From 471df1c1f73435ce30f364fcfb2ef5b2aabcd2ea Mon Sep 17 00:00:00 2001 From: Sung Yoon Whang Date: Tue, 30 Jun 2026 16:02:09 -0700 Subject: [PATCH 1/2] core/bazel: fix race There's a race in `streamOutput` and `streamAndParseTargets` where the stderr buffer was being read / written concurrently. When the context gets cancelled and bails, it's trying to read the buffer while it's still being written to by the other goroutine. The streaming goroutine should block until the other goroutine exits, so that it doesn't attempt to read the byte buffer concurrently. Fixes #134. --- core/bazel/stream.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/core/bazel/stream.go b/core/bazel/stream.go index 2e78fd7..959717c 100644 --- a/core/bazel/stream.go +++ b/core/bazel/stream.go @@ -32,6 +32,7 @@ func streamOutput(ctx context.Context, src io.Reader, dst io.Writer) error { select { case <-ctx.Done(): + <-done return ctx.Err() case err := <-done: return err @@ -52,6 +53,7 @@ func streamAndParseTargets(ctx context.Context, src io.Reader, dst io.Writer) (* select { case <-ctx.Done(): + <-done return nil, ctx.Err() case res := <-done: return res.queryResult, res.err From 03c53ec5e698d17b372b6de82944c53ad35f00cf Mon Sep 17 00:00:00 2001 From: Sung Yoon Whang Date: Tue, 7 Jul 2026 23:01:09 -0700 Subject: [PATCH 2/2] second thought - address code review comments differently. --- core/bazel/bazel.go | 9 +++ core/bazel/query.go | 42 ++++++++-- core/bazel/query_test.go | 169 +++++++++++++++++++++++++++++++++++++++ core/bazel/stream.go | 48 ++++------- core/execcmd/execcmd.go | 13 +-- 5 files changed, 236 insertions(+), 45 deletions(-) diff --git a/core/bazel/bazel.go b/core/bazel/bazel.go index 4b13886..52cf269 100644 --- a/core/bazel/bazel.go +++ b/core/bazel/bazel.go @@ -32,6 +32,13 @@ import ( const ( // default query timeout if not provided in config _queryTimeout = 15 * time.Minute + // _pipeUnblockDelay is how long after the command context ends a query + // waits before force-closing the stdout/stderr pipes to unblock stream + // reads that never saw EOF (a descendant of the killed process can inherit + // the pipe write ends and hold them open, see go.dev/issue/23019). It + // exceeds execcmd.GracePeriod so the SIGTERM→SIGKILL sequence gets a + // chance to end the streams with a natural EOF first. + _pipeUnblockDelay = execcmd.GracePeriod + 5*time.Second ) type QueryRequest struct { @@ -56,6 +63,7 @@ type BazelClient struct { logger *zap.SugaredLogger execCommandContext func(ctx context.Context, name string, arg ...string) commander queryTimeout time.Duration + pipeUnblockDelay time.Duration streamLogs bool } @@ -98,6 +106,7 @@ func NewBazelClient(ctx context.Context, p Params) (*BazelClient, error) { logger: p.Logger, execCommandContext: execCmd, queryTimeout: timeout, + pipeUnblockDelay: _pipeUnblockDelay, streamLogs: p.StreamLogs, }, nil } diff --git a/core/bazel/query.go b/core/bazel/query.go index 37f10a7..2d5e97a 100644 --- a/core/bazel/query.go +++ b/core/bazel/query.go @@ -22,6 +22,7 @@ import ( "io" "os" "strings" + "time" buildpb "github.com/bazelbuild/buildtools/build_proto" "go.uber.org/zap" @@ -74,8 +75,6 @@ func (b *BazelClient) executeQueryInternal(ctx context.Context, query string, st if b.streamLogs { stderrSink = os.Stderr } - // orchestrate `allOfFailFast` - // create a `g` group and a new `gCtx` derived from our 15 minute timeout `ctx`. g, gCtx := errgroup.WithContext(cmdCtx) // Start the process @@ -84,16 +83,47 @@ func (b *BazelClient) executeQueryInternal(ctx context.Context, query string, st } // stream and parse targets g.Go(func() error { - var err error - queryResults, err = streamAndParseTargets(gCtx, stdout, &stdoutBuf) - return err + res, err := getQueryResult(gCtx, stdout, &stdoutBuf) + if err != nil { + return err + } + queryResults = res + return nil }) // stream stderr g.Go(func() error { return streamOutput(gCtx, stderr, stderrSink) }) - waitErr := cmd.Wait() + + // The stream reads normally end with EOF when the process dies and the + // kernel closes the pipe write ends. But a descendant of a killed process + // that inherited the pipes can hold the write ends open indefinitely + // (go.dev/issue/23019). cmd.Wait would force-close the parent ends, but we + // only call Wait after the reads finish, so this watchdog is what + // guarantees the reads can't block forever. + readsDone := make(chan struct{}) + go func() { + select { + case <-readsDone: + return + case <-cmdCtx.Done(): + } + timer := time.NewTimer(b.pipeUnblockDelay) + defer timer.Stop() + select { + case <-readsDone: + case <-timer.C: + stdout.Close() + stderr.Close() + } + }() + + // Drain the streams before reaping the process: cmd.Wait closes the parent + // pipe ends as soon as the process exits, which would truncate whatever + // the readers hadn't consumed yet and race on the output buffers. streamErr := g.Wait() + close(readsDone) + waitErr := cmd.Wait() if waitErr != nil { return queryResults, b.wrapQueryFailure("bazel query failed", waitErr, &stderrBuf) } diff --git a/core/bazel/query_test.go b/core/bazel/query_test.go index c506214..a35737a 100644 --- a/core/bazel/query_test.go +++ b/core/bazel/query_test.go @@ -18,6 +18,7 @@ import ( "bytes" "context" "errors" + "fmt" "io" "strings" "testing" @@ -175,6 +176,174 @@ func TestExecuteQueryInternal_ContextTimeout(t *testing.T) { assert.Contains(t, err.Error(), "deadline exceeded") } +// TestExecuteQueryInternal_PipesHeldOpenAfterCancel simulates a descendant of +// the killed bazel process inheriting the stdout/stderr pipes and holding +// their write ends open past the kill (go.dev/issue/23019). Since cmd.Wait — +// which would force-close the parent ends — only runs after the stream reads +// finish, the pipe watchdog must unblock the reads or the query would hang. +func TestExecuteQueryInternal_PipesHeldOpenAfterCancel(t *testing.T) { + defer goleak.VerifyNone(t) + ctrl := gomock.NewController(t) + mockCmd := commandermock.NewMockcommander(ctrl) + + // The write ends are intentionally never closed by the "process". + prStdout, pwStdout := io.Pipe() + prStderr, pwStderr := io.Pipe() + defer pwStdout.Close() + defer pwStderr.Close() + + var cmdCtx context.Context + gomock.InOrder( + mockCmd.EXPECT().StdoutPipe().Return(prStdout, nil), + mockCmd.EXPECT().StderrPipe().Return(prStderr, nil), + mockCmd.EXPECT().Start().Return(nil), + mockCmd.EXPECT().Wait().DoAndReturn(func() error { + // The process itself dies with the context; only its orphaned + // descendant lives on, holding the pipes. + <-cmdCtx.Done() + return context.DeadlineExceeded + }), + ) + + client, err := NewBazelClient(context.Background(), Params{ + BazelCommand: "bazel", + WorkspacePath: "/tmp/test", + EnvVarsMap: map[string]string{}, + Logger: zap.NewNop().Sugar(), + QueryTimeout: 10 * time.Millisecond, + ExecCommandContext: func(ctx context.Context, name string, arg ...string) commander { + cmdCtx = ctx + return mockCmd + }, + }) + require.NoError(t, err) + client.pipeUnblockDelay = 20 * time.Millisecond + + type queryOutcome struct { + result *buildpb.QueryResult + err error + } + done := make(chan queryOutcome, 1) + go func() { + result, err := client.executeQueryInternal(context.Background(), "//...", nil) + done <- queryOutcome{result: result, err: err} + }() + + select { + case out := <-done: + require.Error(t, out.err) + assert.ErrorIs(t, out.err, context.DeadlineExceeded) + assert.Nil(t, out.result) + case <-time.After(10 * time.Second): + t.Fatal("executeQueryInternal never returned: pipe reads were not unblocked") + } +} + +// TestExecuteQueryInternal_DrainsStreamsBeforeWait verifies that cmd.Wait runs +// only after the stream reads finish. The mock Waits emulate exec.Cmd.Wait, +// which force-closes the parent pipe ends when it returns — had Wait run +// before the reads completed, the still-streaming output would be truncated. +func TestExecuteQueryInternal_DrainsStreamsBeforeWait(t *testing.T) { + t.Run("stdout is fully parsed", func(t *testing.T) { + defer goleak.VerifyNone(t) + ctrl := gomock.NewController(t) + mockCmd := commandermock.NewMockcommander(ctrl) + + prStdout, pwStdout := io.Pipe() + prStderr, pwStderr := io.Pipe() + + // io.Pipe is synchronous, so every write below blocks until the query's + // stream reader consumes it: the "process" is still producing output + // while the query runs. + const targetCount = 100 + go func() { + defer pwStdout.Close() + defer pwStderr.Close() + ruleClass := "go_library" + for i := 0; i < targetCount; i++ { + name := fmt.Sprintf("//pkg:target%d", i) + target := &buildpb.Target{ + Type: buildpb.Target_RULE.Enum(), + Rule: &buildpb.Rule{Name: &name, RuleClass: &ruleClass}, + } + if _, err := protodelim.MarshalTo(pwStdout, target); err != nil { + return + } + } + }() + + gomock.InOrder( + mockCmd.EXPECT().StdoutPipe().Return(prStdout, nil), + mockCmd.EXPECT().StderrPipe().Return(prStderr, nil), + mockCmd.EXPECT().Start().Return(nil), + mockCmd.EXPECT().Wait().DoAndReturn(func() error { + prStdout.Close() + prStderr.Close() + return nil + }), + ) + + client, err := NewBazelClient(context.Background(), Params{ + BazelCommand: "bazel", + WorkspacePath: "/tmp/test", + EnvVarsMap: map[string]string{}, + Logger: zap.NewNop().Sugar(), + ExecCommandContext: func(ctx context.Context, name string, arg ...string) commander { + return mockCmd + }, + }) + require.NoError(t, err) + + result, err := client.executeQueryInternal(context.Background(), "//...", nil) + require.NoError(t, err) + require.NotNil(t, result) + assert.Len(t, result.Target, targetCount) + }) + + t.Run("stderr is fully captured in the failure error", func(t *testing.T) { + defer goleak.VerifyNone(t) + ctrl := gomock.NewController(t) + mockCmd := commandermock.NewMockcommander(ctrl) + + prStderr, pwStderr := io.Pipe() + + const stderrTail = "FINAL STDERR LINE" + go func() { + defer pwStderr.Close() + _, _ = io.WriteString(pwStderr, strings.Repeat("bazel progress line\n", 200)+stderrTail) + }() + + gomock.InOrder( + mockCmd.EXPECT().StdoutPipe().Return(io.NopCloser(strings.NewReader("")), nil), + mockCmd.EXPECT().StderrPipe().Return(prStderr, nil), + mockCmd.EXPECT().Start().Return(nil), + mockCmd.EXPECT().Wait().DoAndReturn(func() error { + prStderr.Close() + return errors.New("exit status 7") + }), + ) + + client, err := NewBazelClient(context.Background(), Params{ + BazelCommand: "bazel", + WorkspacePath: "/tmp/test", + EnvVarsMap: map[string]string{}, + Logger: zap.NewNop().Sugar(), + ExecCommandContext: func(ctx context.Context, name string, arg ...string) commander { + return mockCmd + }, + }) + require.NoError(t, err) + + result, err := client.executeQueryInternal(context.Background(), "//...", nil) + require.Error(t, err) + require.NotNil(t, result) + // This asserts on the payload the error carries (the captured stderr), + // not on the error's own wording: the tail marker only appears if the + // whole stream was drained before the process was reaped. + assert.Contains(t, err.Error(), stderrTail) + }) +} + func TestExecuteQueryInternal_Failures(t *testing.T) { tests := []struct { name string diff --git a/core/bazel/stream.go b/core/bazel/stream.go index 959717c..fb12abc 100644 --- a/core/bazel/stream.go +++ b/core/bazel/stream.go @@ -23,41 +23,17 @@ import ( "google.golang.org/protobuf/encoding/protodelim" ) +// streamOutput copies src into dst until the stream ends. The copy blocks +// until it is unblocked externally — by the process exiting and closing the +// pipe write end, or by the caller force-closing the read end — so when ctx +// has ended, any copy error is just fallout from that shutdown and the +// context error is reported instead. func streamOutput(ctx context.Context, src io.Reader, dst io.Writer) error { - done := make(chan error, 1) - go func() { - _, err := io.Copy(dst, src) - done <- err - }() - - select { - case <-ctx.Done(): - <-done - return ctx.Err() - case err := <-done: - return err - } -} - -func streamAndParseTargets(ctx context.Context, src io.Reader, dst io.Writer) (*buildpb.QueryResult, error) { - type result struct { - queryResult *buildpb.QueryResult - err error - } - done := make(chan result, 1) - - go func() { - queryResult, err := getQueryResult(ctx, src, dst) - done <- result{queryResult: queryResult, err: err} - }() - - select { - case <-ctx.Done(): - <-done - return nil, ctx.Err() - case res := <-done: - return res.queryResult, res.err + _, err := io.Copy(dst, src) + if ctxErr := ctx.Err(); ctxErr != nil { + return ctxErr } + return err } // cancelCheckInterval is how often we poll ctx.Err() inside per-target hot loops. @@ -94,6 +70,10 @@ func getQueryResult(ctx context.Context, src io.Reader, dst io.Writer) (*buildpb } result.Target = append(result.Target, &target) } - + // Like streamOutput, prefer the context error over whatever the shutdown + // did to the stream mid-parse. + if err := ctx.Err(); err != nil { + return result, err + } return result, parseErr } diff --git a/core/execcmd/execcmd.go b/core/execcmd/execcmd.go index 1819b3f..aa48e87 100644 --- a/core/execcmd/execcmd.go +++ b/core/execcmd/execcmd.go @@ -26,12 +26,15 @@ import ( "time" ) -// gracePeriod is how long the child has to exit after SIGTERM before the Go -// runtime escalates to SIGKILL via Cmd.WaitDelay. -const gracePeriod = 10 * time.Second +// GracePeriod is how long the child has to exit after SIGTERM before the Go +// runtime escalates to SIGKILL via Cmd.WaitDelay. Exported so callers that +// layer their own shutdown timeouts on top of a command (e.g. force-closing +// its I/O pipes) can schedule them to fire only after the kill sequence has +// run its course. +const GracePeriod = 10 * time.Second // CommandContext is a drop-in replacement for exec.CommandContext that sends -// SIGTERM when ctx is canceled and escalates to SIGKILL after gracePeriod if +// SIGTERM when ctx is canceled and escalates to SIGKILL after GracePeriod if // the process is still running. This gives child processes (e.g. git, bazel) // a chance to release lock files and disconnect cleanly before being killed. func CommandContext(ctx context.Context, name string, args ...string) *exec.Cmd { @@ -43,6 +46,6 @@ func CommandContext(ctx context.Context, name string, args ...string) *exec.Cmd } return err } - cmd.WaitDelay = gracePeriod + cmd.WaitDelay = GracePeriod return cmd }