From 010e459e957347bcdde06fc6304780a5b06285fe Mon Sep 17 00:00:00 2001
From: "KADOTA, Kyohei" <lufia@users.noreply.github.com>
Date: Tue, 6 Apr 2021 22:43:02 +0900
Subject: [PATCH] Clean the repository cache if it is in incomplete state
 (#595)

---
 pkg/common/executor.go      | 16 ++++++++--------
 pkg/common/executor_test.go | 38 +++++++++++++++++++++++++++++++++++++
 pkg/common/git.go           |  6 +++---
 3 files changed, 49 insertions(+), 11 deletions(-)

diff --git a/pkg/common/executor.go b/pkg/common/executor.go
index 197cd5b..cd92a6c 100644
--- a/pkg/common/executor.go
+++ b/pkg/common/executor.go
@@ -105,17 +105,17 @@ func NewParallelExecutor(executors ...Executor) Executor {
 			}()
 		}
 
+		// Executor waits all executors to cleanup these resources.
+		var firstErr error
 		for i := 0; i < len(executors); i++ {
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			case err := <-errChan:
-				if err != nil {
-					return err
-				}
+			if err := <-errChan; err != nil && firstErr == nil {
+				firstErr = err
 			}
 		}
-		return nil
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		return firstErr
 	}
 }
 
diff --git a/pkg/common/executor_test.go b/pkg/common/executor_test.go
index 215d4fc..17df3b7 100644
--- a/pkg/common/executor_test.go
+++ b/pkg/common/executor_test.go
@@ -89,3 +89,41 @@ func TestNewParallelExecutor(t *testing.T) {
 
 	assert.Nil(err)
 }
+
+func TestNewParallelExecutorFailed(t *testing.T) {
+	assert := assert.New(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	count := 0
+	errorWorkflow := NewPipelineExecutor(func(ctx context.Context) error {
+		count++
+		return fmt.Errorf("fake error")
+	})
+	err := NewParallelExecutor(errorWorkflow)(ctx)
+	assert.Equal(1, count)
+	assert.ErrorIs(context.Canceled, err)
+}
+
+func TestNewParallelExecutorCanceled(t *testing.T) {
+	assert := assert.New(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	errExpected := fmt.Errorf("fake error")
+
+	count := 0
+	successWorkflow := NewPipelineExecutor(func(ctx context.Context) error {
+		count++
+		return nil
+	})
+	errorWorkflow := NewPipelineExecutor(func(ctx context.Context) error {
+		count++
+		return errExpected
+	})
+	err := NewParallelExecutor(errorWorkflow, successWorkflow, successWorkflow)(ctx)
+	assert.Equal(3, count)
+	assert.Error(errExpected, err)
+}
diff --git a/pkg/common/git.go b/pkg/common/git.go
index 1c12b9e..e88266a 100644
--- a/pkg/common/git.go
+++ b/pkg/common/git.go
@@ -190,7 +190,7 @@ type NewGitCloneExecutorInput struct {
 }
 
 // CloneIfRequired ...
-func CloneIfRequired(refName plumbing.ReferenceName, input NewGitCloneExecutorInput, logger log.FieldLogger) (*git.Repository, error) {
+func CloneIfRequired(ctx context.Context, refName plumbing.ReferenceName, input NewGitCloneExecutorInput, logger log.FieldLogger) (*git.Repository, error) {
 	r, err := git.PlainOpen(input.Dir)
 	if err != nil {
 		var progressWriter io.Writer
@@ -203,7 +203,7 @@ func CloneIfRequired(refName plumbing.ReferenceName, input NewGitCloneExecutorIn
 			progressWriter = os.Stdout
 		}
 
-		r, err = git.PlainClone(input.Dir, false, &git.CloneOptions{
+		r, err = git.PlainCloneContext(ctx, input.Dir, false, &git.CloneOptions{
 			URL:      input.URL,
 			Progress: progressWriter,
 		})
@@ -228,7 +228,7 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) Executor {
 		defer cloneLock.Unlock()
 
 		refName := plumbing.ReferenceName(fmt.Sprintf("refs/heads/%s", input.Ref))
-		r, err := CloneIfRequired(refName, input, logger)
+		r, err := CloneIfRequired(ctx, refName, input, logger)
 		if err != nil {
 			return err
 		}