Correctly handle fail-fast

Author: avoidik

pkg/common/executor.go

@@ -131,6 +131,87 @@ func NewParallelExecutor(parallel int, executors ...Executor) Executor {
 	}
 }
 
+// NewFailFastParallelExecutor creates a parallel executor that respects fail-fast semantics
+// When fail-fast is enabled via context, it will cancel remaining work on first error
+func NewFailFastParallelExecutor(parallel int, executors ...Executor) Executor {
+	return func(ctx context.Context) error {
+		failFast := IsFailFast(ctx)
+
+		// If fail-fast is disabled, use the standard parallel executor
+		if !failFast {
+			return NewParallelExecutor(parallel, executors...)(ctx)
+		}
+
+		// Fail-fast mode: create a cancellable context for workers
+		workCtx, cancelWork := context.WithCancel(ctx)
+		defer cancelWork()
+
+		work := make(chan Executor, len(executors))
+		errs := make(chan error, len(executors))
+
+		if 1 > parallel {
+			log.Debugf("Parallel tasks (%d) below minimum, setting to 1", parallel)
+			parallel = 1
+		}
+
+		// Start worker goroutines
+		for i := 0; i < parallel; i++ {
+			go func(work <-chan Executor, errs chan<- error) {
+				for executor := range work {
+					// Check if work context was cancelled (fail-fast triggered)
+					if workCtx.Err() != nil {
+						errs <- workCtx.Err()
+						continue
+					}
+					errs <- executor(workCtx)
+				}
+			}(work, errs)
+		}
+
+		// Queue work and monitor for failures
+		go func() {
+			defer close(work)
+			for i := 0; i < len(executors); i++ {
+				// Check if we should stop queuing due to failure
+				if workCtx.Err() != nil {
+					// Don't queue remaining work, but send cancelled errors for remaining executors
+					for j := i; j < len(executors); j++ {
+						errs <- workCtx.Err()
+					}
+					return
+				}
+				work <- executors[i]
+			}
+		}()
+
+		// Collect results and trigger fail-fast on first error
+		var firstErr error
+		for i := 0; i < len(executors); i++ {
+			err := <-errs
+
+			// Track the first non-Warning error
+			if err != nil && firstErr == nil {
+				switch err.(type) {
+				case Warning:
+					// Warnings don't trigger fail-fast
+					log.Warning(err.Error())
+				default:
+					firstErr = err
+					// Cancel remaining work on first real error
+					cancelWork()
+				}
+			}
+		}
+
+		// Check if parent context was cancelled
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+
+		return firstErr
+	}
+}
+
 func NewFieldExecutor(name string, value interface{}, exec Executor) Executor {
 	return func(ctx context.Context) error {
 		return exec(WithLogger(ctx, Logger(ctx).WithField(name, value)))

pkg/common/executor_test.go

@@ -3,6 +3,7 @@ package common
 import (
 	"context"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -150,3 +151,123 @@ func TestNewParallelExecutorCanceled(t *testing.T) {
 	assert.Equal(3, count)
 	assert.Error(errExpected, err)
 }
+
+func TestNewFailFastParallelExecutorWithFailFastTrue(t *testing.T) {
+	assert := assert.New(t)
+
+	ctx := WithFailFast(context.Background(), true)
+
+	executedCount := 0
+	var mu sync.Mutex
+
+	// Create executors: some succeed, one fails, rest should be cancelled
+	executors := []Executor{
+		func(ctx context.Context) error {
+			mu.Lock()
+			executedCount++
+			mu.Unlock()
+			time.Sleep(100 * time.Millisecond)
+			return nil
+		},
+		func(ctx context.Context) error {
+			mu.Lock()
+			executedCount++
+			mu.Unlock()
+			time.Sleep(100 * time.Millisecond)
+			return fmt.Errorf("intentional failure")
+		},
+		func(ctx context.Context) error {
+			mu.Lock()
+			executedCount++
+			mu.Unlock()
+			time.Sleep(2 * time.Second) // Should be cancelled
+			return nil
+		},
+		func(ctx context.Context) error {
+			mu.Lock()
+			executedCount++
+			mu.Unlock()
+			time.Sleep(2 * time.Second) // Should be cancelled
+			return nil
+		},
+	}
+
+	err := NewFailFastParallelExecutor(2, executors...)(ctx)
+
+	assert.Error(err)
+	assert.Contains(err.Error(), "intentional failure")
+}
+
+func TestNewFailFastParallelExecutorWithFailFastFalse(t *testing.T) {
+	assert := assert.New(t)
+
+	ctx := WithFailFast(context.Background(), false)
+
+	executedCount := 0
+	var mu sync.Mutex
+
+	executors := []Executor{
+		func(ctx context.Context) error {
+			mu.Lock()
+			executedCount++
+			mu.Unlock()
+			return nil
+		},
+		func(ctx context.Context) error {
+			mu.Lock()
+			executedCount++
+			mu.Unlock()
+			return fmt.Errorf("intentional failure")
+		},
+		func(ctx context.Context) error {
+			mu.Lock()
+			executedCount++
+			mu.Unlock()
+			return nil
+		},
+	}
+
+	err := NewFailFastParallelExecutor(2, executors...)(ctx)
+
+	assert.Error(err)
+	mu.Lock()
+	assert.Equal(3, executedCount, "all executors should run when fail-fast is false")
+	mu.Unlock()
+}
+
+func TestNewFailFastParallelExecutorNoFailFastInContext(t *testing.T) {
+	assert := assert.New(t)
+
+	ctx := context.Background()
+
+	executedCount := 0
+	var mu sync.Mutex
+
+	executors := []Executor{
+		func(ctx context.Context) error {
+			mu.Lock()
+			executedCount++
+			mu.Unlock()
+			return nil
+		},
+		func(ctx context.Context) error {
+			mu.Lock()
+			executedCount++
+			mu.Unlock()
+			return fmt.Errorf("intentional failure")
+		},
+		func(ctx context.Context) error {
+			mu.Lock()
+			executedCount++
+			mu.Unlock()
+			return nil
+		},
+	}
+
+	err := NewFailFastParallelExecutor(2, executors...)(ctx)
+
+	assert.Error(err)
+	mu.Lock()
+	assert.Equal(3, executedCount, "all executors should run when fail-fast not in context")
+	mu.Unlock()
+}

pkg/common/job_error.go

@@ -12,6 +12,26 @@ type jobCancelCtx string
 
 const JobCancelCtxVal = jobCancelCtx("job.cancel")
 
+type failFastContextKey string
+
+const FailFastContextKeyVal = failFastContextKey("job.failfast")
+
+// WithFailFast adds fail-fast configuration to the context
+func WithFailFast(ctx context.Context, failFast bool) context.Context {
+	return context.WithValue(ctx, FailFastContextKeyVal, failFast)
+}
+
+// IsFailFast returns whether fail-fast is enabled for this context
+func IsFailFast(ctx context.Context) bool {
+	val := ctx.Value(FailFastContextKeyVal)
+	if val != nil {
+		if ff, ok := val.(bool); ok {
+			return ff
+		}
+	}
+	return false
+}
+
 // JobError returns the job error for current context if any
 func JobError(ctx context.Context) error {
 	val := ctx.Value(jobErrorContextKeyVal)

pkg/runner/runner.go

@@ -207,7 +207,20 @@ func (runner *runnerImpl) NewPlanExecutor(plan *model.Plan) common.Executor {
 						return executor(common.WithJobErrorContainer(WithJobLogger(ctx, rc.Run.JobID, jobName, rc.Config, &rc.Masks, matrix)))
 					})
 				}
-				pipeline = append(pipeline, common.NewParallelExecutor(maxParallel, stageExecutor...))
+
+				// Use fail-fast executor for matrix jobs
+				matrixExecutor := common.NewFailFastParallelExecutor(maxParallel, stageExecutor...)
+
+				// Wrap with fail-fast context based on strategy
+				pipeline = append(pipeline, func(ctx context.Context) error {
+					// Inject fail-fast setting into context
+					failFast := false
+					if job.Strategy != nil {
+						failFast = job.Strategy.FailFast
+					}
+					ctx = common.WithFailFast(ctx, failFast)
+					return matrixExecutor(ctx)
+				})
 			}
 
 			log.Debugf("PlanExecutor concurrency: %d", runner.config.GetConcurrentJobs())
@@ -223,6 +236,11 @@ func handleFailure(plan *model.Plan) common.Executor {
 		for _, stage := range plan.Stages {
 			for _, run := range stage.Runs {
 				if run.Job().Result == "failure" {
+					job := run.Job()
+					// Check if this was a matrix job with fail-fast
+					if job.Strategy != nil && job.Strategy.FailFast {
+						return fmt.Errorf("Job '%s' failed (fail-fast enabled, remaining matrix jobs may have been cancelled)", run.String())
+					}
 					return fmt.Errorf("Job '%s' failed", run.String())
 				}
 			}