fix(executor): stop counting skipped tasks twice on retry runs

executor/executor.go

@@ -113,6 +113,17 @@ func (e *Executor) Execute(ctx context.Context) error {
 		semaphore = make(chan struct{}, e.opts.MaxParallelization)
 	}
 
+	// Count pre-completed tasks before starting anything. startReadyTasks marks
+	// dependency-skipped tasks as failed *and* reports them on completionCh, so
+	// counting after it would count those tasks twice and end the run early.
+	finished := 0
+	for _, run := range e.tasks {
+		if run.status == statusSuccess || run.status == statusFailed {
+			finished++
+		}
+	}
+	total := len(e.tasks)
+
 	if err := e.startReadyTasks(ctx, completionCh, outputCh, semaphore, executorDone); err != nil {
 		return err
 	}
@@ -121,19 +132,13 @@ func (e *Executor) Execute(ctx context.Context) error {
 		return err
 	}
 
+	drainStop := make(chan struct{})
 	drainDone := make(chan struct{})
 	go func() {
-		e.drainOutput(ctx, outputCh)
+		e.drainOutput(ctx, outputCh, drainStop)
 		close(drainDone)
 	}()
 
-	finished := 0
-	for _, run := range e.tasks {
-		if run.status == statusSuccess || run.status == statusFailed {
-			finished++
-		}
-	}
-	total := len(e.tasks)
 	var errs []error
 
 	for finished < total {
@@ -198,7 +203,7 @@ func (e *Executor) Execute(ctx context.Context) error {
 		}
 	}
 
-	close(outputCh)
+	close(drainStop)
 	<-drainDone // wait for all output to be processed before returning
 	return errors.Join(errs...)
 }
@@ -365,28 +370,34 @@ func (e *Executor) pendingNames() string {
 	return strings.Join(names, ", ")
 }
 
-func (e *Executor) drainOutput(ctx context.Context, outputCh <-chan Output) {
-	for {
-		select {
-		case out, ok := <-outputCh:
-			if !ok {
+// drainOutput forwards task output to the output handler until stop is closed or
+// ctx is cancelled. outputCh is deliberately never closed: a task goroutine that
+// outlives the run would panic sending on a closed channel, so it is left open
+// and such a send is simply dropped once draining has stopped.
+func (e *Executor) drainOutput(ctx context.Context, outputCh <-chan Output, stop <-chan struct{}) {
+	// Drain any lines already queued before exiting so the last output of a
+	// failing command is not lost.
+	flush := func(handlerCtx context.Context) {
+		for {
+			select {
+			case out := <-outputCh:
+				_ = e.outputHandler.HandleOutput(handlerCtx, out)
+			default:
 				return
 			}
+		}
+	}
+
+	for {
+		select {
+		case out := <-outputCh:
 			_ = e.outputHandler.HandleOutput(ctx, out)
+		case <-stop:
+			flush(ctx)
+			return
 		case <-ctx.Done():
-			// Drain any lines already queued before exiting so the last output of a
-			// failing command is not lost when the context is cancelled.
-			for {
-				select {
-				case out, ok := <-outputCh:
-					if !ok {
-						return
-					}
-					_ = e.outputHandler.HandleOutput(context.Background(), out)
-				default:
-					return
-				}
-			}
+			flush(context.Background())
+			return
 		}
 	}
 }

executor/executor_test.go

@@ -437,3 +437,115 @@ func TestExecutor_DependencyBlocking(t *testing.T) {
 		t.Error("dep started before base completed")
 	}
 }
+
+type collectingHandler struct {
+	mu    sync.Mutex
+	lines []string
+}
+
+func (c *collectingHandler) HandleOutput(_ context.Context, out Output) error {
+	c.mu.Lock()
+	c.lines = append(c.lines, string(out.Output))
+	c.mu.Unlock()
+	return nil
+}
+
+func (c *collectingHandler) snapshot() []string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return append([]string(nil), c.lines...)
+}
+
+// A dependent of a pre-completed failed task is skipped during the very first
+// startReadyTasks call, which both marks it failed and reports it on the completion
+// channel. Counting it twice ends the run while other tasks are still executing.
+func TestExecutor_SkippedDependentOfPreCompletedFailureDoesNotEndRunEarly(t *testing.T) {
+	started := make(chan struct{})
+	release := make(chan struct{})
+
+	tasks := []Task{
+		{
+			Name: "slow-task",
+			ExecuteFn: func(ctx context.Context, name string, h OutputHandler) error {
+				close(started)
+				<-release
+				return h.HandleOutput(ctx, Output{Output: []byte("last line"), CmdName: name})
+			},
+		},
+		{
+			Name:      "failed-task",
+			ExecuteFn: func(ctx context.Context, name string, h OutputHandler) error { return nil },
+		},
+		{
+			Name:      "dependent",
+			Needs:     []string{"failed-task"},
+			ExecuteFn: func(ctx context.Context, name string, h OutputHandler) error { return nil },
+		},
+	}
+
+	exec, err := NewExecutor(tasks, ExecutorOptions{})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	exec.WithPreCompleted(nil, []string{"failed-task"})
+	handler := &collectingHandler{}
+	exec.WithOutputHandler(handler)
+
+	done := make(chan error, 1)
+	go func() { done <- exec.Execute(context.Background()) }()
+
+	<-started
+	select {
+	case <-done:
+		t.Fatal("Execute returned while slow-task was still running")
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	close(release)
+	select {
+	case err := <-done:
+		if !errors.Is(err, ErrTaskSkipped) {
+			t.Errorf("expected skipped dependent error, got %v", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("Execute did not return after slow-task finished")
+	}
+
+	if got := handler.snapshot(); len(got) != 1 || got[0] != "last line" {
+		t.Errorf("expected slow-task output to be drained, got %v", got)
+	}
+}
+
+// Output emitted after the run has finished must be dropped, not fatal.
+func TestExecutor_OutputAfterRunFinishesIsDropped(t *testing.T) {
+	emitted := make(chan struct{})
+
+	tasks := []Task{
+		{
+			Name: "task",
+			ExecuteFn: func(ctx context.Context, name string, h OutputHandler) error {
+				go func() {
+					time.Sleep(20 * time.Millisecond)
+					_ = h.HandleOutput(ctx, Output{Output: []byte("stray"), CmdName: name})
+					close(emitted)
+				}()
+				return nil
+			},
+		},
+	}
+
+	exec, err := NewExecutor(tasks, ExecutorOptions{})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if err := exec.Execute(context.Background()); err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+
+	select {
+	case <-emitted:
+	case <-time.After(5 * time.Second):
+		t.Fatal("stray output goroutine never completed")
+	}
+}