From 78f14154289949bb76ce64d2750815773ebeb28b Mon Sep 17 00:00:00 2001 From: Lau Josefsen Date: Mon, 10 Aug 2026 11:29:01 +0200 Subject: [PATCH] fix(executor): stop counting skipped tasks twice on retry runs 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. --- executor/executor.go | 65 +++++++++++++--------- executor/executor_test.go | 112 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 150 insertions(+), 27 deletions(-) diff --git a/executor/executor.go b/executor/executor.go index fe2269a0..99fcc72c 100644 --- a/executor/executor.go +++ b/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 } } } diff --git a/executor/executor_test.go b/executor/executor_test.go index 9c405592..2df91d2a 100644 --- a/executor/executor_test.go +++ b/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") + } +}