Fix a bug that causes piped crashes while rolling back (#617)

**What this PR does / why we need it**:

Scenario:
- A deployment is triggered
- The first stage (e.g. `K8S_SYNC`) is cancelled from web UI while `ensurePreparing` is running
- So `prepareOnce` is marked as executed but `s.deploymentConfig = cfg` is still not configured
- The rollback stage starts running
- Because the `prepareOnce` was marked as executed so `s.deploymentConfig` is still nil
- That causes a panic when accessing `s.deploymentConfig` in rollback execution

**Which issue(s) this PR fixes**:

Fixes #

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
-->
```release-note
NONE
```

/cc @nakabonne 
This PR was merged by Kapetanios.

pkg/app/piped/controller/scheduler.go

@@ -31,7 +31,6 @@ import (
 	pln "github.com/pipe-cd/pipe/pkg/app/piped/planner"
 	"github.com/pipe-cd/pipe/pkg/cache"
 	"github.com/pipe-cd/pipe/pkg/config"
-	"github.com/pipe-cd/pipe/pkg/git"
 	"github.com/pipe-cd/pipe/pkg/model"
 )
 
@@ -67,7 +66,8 @@ type scheduler struct {
 
 	deploymentConfig *config.Config
 	pipelineable     config.Pipelineable
-	prepareOnce      sync.Once
+	prepareMu        sync.Mutex
+	prepared         bool
 	// Current status of each stages.
 	// We stores their current statuses into this field
 	// because the deployment model is readonly to avoid data race.
@@ -468,61 +468,57 @@ func (s *scheduler) executeStage(sig executor.StopSignal, ps model.PipelineStage
 // The log of this preparing process will be written to the first executing stage
 // when a new scheduler has been created.
 func (s *scheduler) ensurePreparing(ctx context.Context, lp logpersister.StageLogPersister) error {
-	var err error
-	s.prepareOnce.Do(func() {
-		lp.Info("Start preparing for deployment")
+	s.prepareMu.Lock()
+	defer s.prepareMu.Unlock()
+	if s.prepared {
+		return nil
+	}
 
-		// Clone repository and checkout to the target revision.
-		var (
-			gitRepo     git.Repo
-			repoDirPath = filepath.Join(s.workingDir, workspaceGitRepoDirName)
-		)
-		gitRepo, err = prepareDeployRepository(ctx, s.deployment, s.gitClient, repoDirPath, s.pipedConfig)
-		if err != nil {
-			lp.Error(err.Error())
-			return
-		}
-		lp.Successf("Successfully cloned repository %s", s.deployment.GitPath.Repo.Id)
+	lp.Info("Start preparing for the deployment")
 
-		// Copy and checkout the running revision.
-		if s.deployment.RunningCommitHash != "" {
-			var (
-				runningGitRepo  git.Repo
-				runningRepoPath = filepath.Join(s.workingDir, workspaceGitRunningRepoDirName)
-			)
-			runningGitRepo, err = gitRepo.Copy(runningRepoPath)
-			if err != nil {
-				lp.Error(err.Error())
-				return
-			}
-			if err = runningGitRepo.Checkout(ctx, s.deployment.RunningCommitHash); err != nil {
-				lp.Error(err.Error())
-				return
-			}
-		}
+	// Clone repository and checkout to the target revision.
+	repoDirPath := filepath.Join(s.workingDir, workspaceGitRepoDirName)
+	gitRepo, err := prepareDeployRepository(ctx, s.deployment, s.gitClient, repoDirPath, s.pipedConfig)
+	if err != nil {
+		lp.Errorf("Unable to prepare repository (%v)", err)
+		return err
+	}
+	lp.Successf("Successfully cloned repository %s", s.deployment.GitPath.Repo.Id)
 
-		// Load deployment configuration for this application.
-		var cfg *config.Config
-		cfg, err = loadDeploymentConfiguration(gitRepo.GetPath(), s.deployment)
+	// Copy and checkout the running revision.
+	if s.deployment.RunningCommitHash != "" {
+		runningRepoPath := filepath.Join(s.workingDir, workspaceGitRunningRepoDirName)
+		runningGitRepo, err := gitRepo.Copy(runningRepoPath)
 		if err != nil {
-			err = fmt.Errorf("failed to load deployment configuration (%w)", err)
-			lp.Error(err.Error())
-			return
+			lp.Errorf("Unable to copy repository (%v)", err)
+			return err
 		}
-		s.deploymentConfig = cfg
-
-		pipelineable, ok := cfg.GetPipelineable()
-		if !ok {
-			err = fmt.Errorf("Unsupport non pipelineable application %s", cfg.Kind)
-			lp.Error(err.Error())
-			return
+		if err = runningGitRepo.Checkout(ctx, s.deployment.RunningCommitHash); err != nil {
+			lp.Errorf("Unable to checkout repository (%v)", err)
+			return err
 		}
-		s.pipelineable = pipelineable
-		lp.Success("Successfully loaded deployment configuration")
+	}
 
-		lp.Info("All preparations have been completed successfully")
-	})
-	return err
+	// Load deployment configuration for this application.
+	cfg, err := loadDeploymentConfiguration(gitRepo.GetPath(), s.deployment)
+	if err != nil {
+		lp.Errorf("Failed to load deployment configuration (%v)", err)
+		return fmt.Errorf("failed to load deployment configuration (%w)", err)
+	}
+	s.deploymentConfig = cfg
+
+	pp, ok := cfg.GetPipelineable()
+	if !ok {
+		lp.Errorf("Unsupport non pipelineable application %s", cfg.Kind)
+		return fmt.Errorf("unsupport non pipelineable application %s", cfg.Kind)
+	}
+	s.pipelineable = pp
+	lp.Success("Successfully loaded deployment configuration")
+
+	s.prepared = true
+	lp.Info("All preparations have been completed successfully")
+
+	return nil
 }
 
 func (s *scheduler) reportStageStatus(ctx context.Context, stageID string, status model.StageStatus, requires []string) error {