From 4d84234a46d9cc7b96a23503904a2e5b5c936473 Mon Sep 17 00:00:00 2001 From: rsbin1178 <3029898593@qq.com> Date: Thu, 8 Oct 2026 22:46:55 +0800 Subject: [PATCH] fix(coding/execution): prefer the sink error over the operation deadline The runner took whichever of {process exit, output trigger, context done} arrived first and kept only that error, so a caller whose sink failed could be told "context deadline exceeded" instead: the fixture's one-second operation deadline can expire before the dispatcher goroutine reports the failure, which is what the macOS sandbox runner hit. The dispatcher now records the sink's failure beside reporting it on the trigger, and the run prefers that record: the sink error leads the returned error and any context error that also happened is joined behind it, so nothing is dropped. The error classification still sees the preferred error and the status stays canceled. Measured: a test cancels the caller's context while its sink is running and lets the sink fail afterwards. Without the preference applied it fails three runs out of three with "Target error should be in err chain", and with it the caller sees both the sink failure and the cancellation, twenty -race repetitions. --- .../execution/runner_sink_failure_test.go | 75 +++++++++++++++++++ internal/coding/execution/runner_unix.go | 49 +++++++++++- 2 files changed, 122 insertions(+), 2 deletions(-) create mode 100644 internal/coding/execution/runner_sink_failure_test.go diff --git a/internal/coding/execution/runner_sink_failure_test.go b/internal/coding/execution/runner_sink_failure_test.go new file mode 100644 index 0000000..9692af4 --- /dev/null +++ b/internal/coding/execution/runner_sink_failure_test.go @@ -0,0 +1,75 @@ +//go:build darwin || linux + +package execution + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestPreferSinkFailure pins the rule: the caller's failing sink is the reason the +// run stopped, so it leads the returned error even when a context error was recorded +// first, and nothing is dropped. +func TestPreferSinkFailure(t *testing.T) { + t.Parallel() + + sinkErr := errors.New("sink stopped") + deadline := context.DeadlineExceeded + + assert.Equal(t, deadline, preferSinkFailure(deadline, nil)) + assert.Equal(t, sinkErr, preferSinkFailure(nil, sinkErr)) + assert.Equal(t, sinkErr, preferSinkFailure(sinkErr, sinkErr)) + + joined := preferSinkFailure(deadline, sinkErr) + require.ErrorIs(t, joined, sinkErr) + require.ErrorIs(t, joined, deadline) + assert.Equal(t, sinkErr.Error(), strings.SplitN(joined.Error(), "\n", 2)[0], "the sink failure leads") +} + +// TestRunnerPrefersTheSinkFailureOverCallerCancellation pins the same rule through +// the runner: a caller that cancels while its sink is still running, and a sink that +// then reports its own failure, still get their failure back as the reason the run +// stopped. Which of the two the runner observed first is a schedule, and the answer +// must not depend on it. +func TestRunnerPrefersTheSinkFailureOverCallerCancellation(t *testing.T) { + t.Parallel() + + fixture := newExecutorFixture(t) + executor := fixture.executor(t, unavailableBackend{}, systemRunnerDependencies()) + operation, authorization := fixture.fullAccessOperation(t, fixture.operationSpec("printf hello; /bin/sleep 10")) + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + sinkErr := errors.New("sink stopped late") + started := make(chan struct{}) + + var once sync.Once + + go func() { + <-started + cancel() + }() + + result, err := executor.Execute(ctx, operation, authorization, SinkFunc(func(sinkCtx context.Context, _ OutputChunk) error { + once.Do(func() { close(started) }) + + // The cancellation reaches the runner first, then this failure lands while it + // is draining the dispatcher. + <-sinkCtx.Done() + time.Sleep(10 * time.Millisecond) + + return sinkErr + })) + + require.ErrorIs(t, err, sinkErr) + require.ErrorIs(t, err, context.Canceled, "the cancellation is still reported") + assert.Equal(t, StatusCanceled, result.Status) +} diff --git a/internal/coding/execution/runner_unix.go b/internal/coding/execution/runner_unix.go index 4302181..9cdd82b 100644 --- a/internal/coding/execution/runner_unix.go +++ b/internal/coding/execution/runner_unix.go @@ -56,7 +56,8 @@ func (e *Executor) run(ctx context.Context, plan *Plan, sink Sink) (Result, erro trigger := make(chan error, 1) chunks := make(chan OutputChunk, plan.operation.output.QueueDepth) readersDone := startOutputReaders(pipes, collector, plan.operation.output.ChunkBytes, chunks, trigger) - dispatchDone := startDispatcher(runCtx, sink, chunks, trigger) + failures := &outputFailure{} + dispatchDone := startDispatcher(runCtx, sink, chunks, trigger, failures) stdinDone := startStdinWriter(pipes.stdinParent, plan.launch.stdin, trigger) waitDone := make(chan error, 1) @@ -66,6 +67,7 @@ func (e *Executor) run(ctx context.Context, plan *Plan, sink Sink) (Result, erro cleanupErr := cleanupProcessGroup(command.Process.Pid, e.termGrace) readerErr := e.finishIO(runCtx, ctx, pipes, readersDone, dispatchDone, stdinDone, chunks, trigger, &runErr) closeErr = errors.Join(closeErr, readerErr, pipes.closeAll()) + runErr = preferSinkFailure(runErr, failures.failure()) result.Duration = e.deps.now().Sub(startedAt) result.Stdout, result.Stderr = collector.results() @@ -179,6 +181,7 @@ func startDispatcher( sink Sink, chunks <-chan OutputChunk, trigger chan<- error, + failures *outputFailure, ) <-chan struct{} { done := make(chan struct{}) go func() { @@ -190,7 +193,9 @@ func startDispatcher( } if err := sink.WriteOutput(ctx, chunk); err != nil { - notifyTrigger(trigger, fmt.Errorf("coding execution: output sink: %w", err)) + failure := fmt.Errorf("coding execution: output sink: %w", err) + failures.record(failure) + notifyTrigger(trigger, failure) return } @@ -423,6 +428,46 @@ func setFirstError(target *error, candidate error) { } } +// outputFailure remembers the sink's own failure. The dispatcher reports it on the +// trigger as well, but the first error to reach the runner wins that race, and the +// caller's failing sink is the more useful answer than an operation deadline that +// expired while the runner was draining. +type outputFailure struct { + mu sync.Mutex + err error +} + +func (f *outputFailure) record(err error) { + f.mu.Lock() + defer f.mu.Unlock() + + if f.err == nil { + f.err = err + } +} + +func (f *outputFailure) failure() error { + f.mu.Lock() + defer f.mu.Unlock() + + return f.err +} + +// preferSinkFailure returns the sink's failure as the reason the run stopped, with +// anything else that happened joined behind it. +func preferSinkFailure(runErr, sinkErr error) error { + switch { + case sinkErr == nil: + return runErr + case runErr == nil: + return sinkErr + case errors.Is(runErr, sinkErr): + return runErr + default: + return errors.Join(sinkErr, runErr) + } +} + // processGroupSignalUnavailable reports whether a failed process-group signal // means the group is gone or is no longer ours to signal. Cleanup runs after the // direct child is reaped, so its pid can be recycled into an unrelated process