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

A retry run creates a new executor over the full task set with pre-completed
tasks. A pending task whose dependency is already failed takes the skip path in
the first startReadyTasks call, which marks it failed and reports it on the
completion channel. The finished counter was seeded after that call, so the skip
was counted twice and the run could end while other tasks were still starting,
panicking with "send on closed channel" when the next task emitted its preamble.

Seed the counter before starting tasks, and stop closing the output channel:
draining now ends on a dedicated signal, so a line from a goroutine that
outlives the run is dropped rather than fatal.

Author: LauJosefsen

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")
+	}
+}