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