From 511c3cddc08104bd2a97bd0168ce57b3afa5aeee Mon Sep 17 00:00:00 2001 From: Bobby Ho Date: Wed, 15 Jul 2026 09:08:11 -0700 Subject: [PATCH] fix(testutil/expecter): remove nested pipe from output logging path (#27204) ## Summary Partial fix for the flake in [PLAT-251](https://linear.app/codercom/issue/PLAT-251/flake-test-go-pg-macos-job-timeoutcancelled) / [coder/internal#1365](https://github.com/coder/internal/issues/1365) (`test-go-pg (macos-latest)` timing out and getting cancelled after 25 minutes). ## Problem `testutil/expecter.Expecter` drains a command's stdout/stderr through `io.Copy` into an internal buffer, but it also teed every write through a **second, independent unbuffered `io.Pipe()`**, read by a `bufio.Scanner`, purely to produce human-readable debug logs: ```go logr, logw := io.Pipe() w := io.MultiWriter(logw, out) // out never blocks; logw is a second unbuffered pipe go func() { io.Copy(w, r) }() // drains the command's real stdout/stderr go func() { bufio.NewScanner(logr).Scan() ... }() // only reads when scheduled ``` `io.MultiWriter` only returns once **every** writer succeeds. If the scanner goroutine is ever delayed (GC pause, scheduler contention under CI's `-parallel=16` test config), the write into `logw` blocks, which blocks the `io.Copy` write, which stops it from reading the command's actual output pipe, which means the **command's own write can never complete either**, since nobody is left reading it. Nothing on this path has a timeout, so once wedged it stays wedged until `go test`'s own `-timeout 20m` kills the whole binary. This reproduced locally in `TestConfigSSH_FileWriteAndOptionsFlow` under the macOS CI job's exact parallelism, stuck writing a routine "executable not in `$PATH`" warning (`cli.currentBinPath`) that the test wasn't actively reading at that instant. It's a flake, not a deterministic failure, because under light load the scanner always keeps up trivially. ### Race condition ```mermaid sequenceDiagram participant TG as Test goroutine participant CmdG as Command goroutine
(inv.Run) participant P1 as Pipe 1
(inv.Stdout) participant Copy as io.Copy goroutine participant P2 as Pipe 2
(logw/logr, debug logging only) participant Scan as Scanner goroutine participant Buf as stdbuf (out)
unbounded, never blocks CmdG->>P1: Write(warning output) P1-->>Copy: Read() unblocks Copy->>P2: MultiWriter step 1: write to logw Copy->>Buf: MultiWriter step 2: write to out rect rgb(255, 230, 230) Note over Scan: Under heavy parallel test load,
Scan()'s next Read() is delayed P2--xCopy: Write(logw) blocks: nobody reading yet end Copy--xP1: Read() no longer called: Copy is stuck writing to P2 CmdG--xP1: Write() can't complete either: nobody reads Pipe 1 Note over CmdG: Command write blocks FOREVER
(confirmed: 17-18 min in a goroutine dump) TG->>TG: ExpectMatch times out (got ""), fails the test Note over TG,CmdG: Test goroutine exits, but CmdG is
orphaned and permanently blocked until
go test's own -timeout kills the binary ``` ## Fix Remove the second pipe. `io.Copy` now writes only to the unbounded, non-blocking `stdbuf`. Debug logging is forwarded via a **bounded, non-blocking channel** instead of a second unbuffered pipe: a full channel just drops the chunk rather than propagating backpressure. This works because the only thing the command's real output pipe depends on is `io.Copy(out, r)`, and `out.Write()` can never block, so `io.Copy` always keeps calling `Read()`, so the command's write always has an active reader. Debug logging becomes provably unable to backpressure the command under test, since losing an occasional log line under extreme load is an acceptable tradeoff, unlike losing a whole CI job to a silent deadlock. The fix lives in the shared harness rather than in `cli/configssh.go` because the warning it's tripping over is legitimate, unrelated product behavior; every other test using this harness that happens to emit output the test isn't actively matching at that instant was exposed to the same bug. **Note:** this addresses one of two distinct root causes bundled under PLAT-251. The other is a Depot `GOCACHEPROG` shutdown hang in CI infrastructure (outside this repo, previously diagnosed by a maintainer on the Feb 2026 incident), which this change cannot affect. Locally this fix measurably improves things, but a residual, load-dependent hang was still observed under extreme synthetic contention on a shared dev machine; watching several real CI runs is the next step before considering the flake fully resolved. --------- Co-authored-by: Cian Johnston --- testutil/expecter/expecter.go | 83 ++++++++++++++++++++++++++--------- 1 file changed, 63 insertions(+), 20 deletions(-) diff --git a/testutil/expecter/expecter.go b/testutil/expecter/expecter.go index bbcbdb5b21..858c82467a 100644 --- a/testutil/expecter/expecter.go +++ b/testutil/expecter/expecter.go @@ -22,15 +22,54 @@ import ( "github.com/coder/serpent" ) -func New(t *testing.T, r io.Reader, name string) *Expecter { - // Use pipe for logging. - logDone := make(chan struct{}) - logr, logw := io.Pipe() +// logTee forwards a best-effort copy of each write to lines for +// line-oriented debug logging, in addition to the real write to out. +// See PLAT-251 for more details. +type logTee struct { + out io.Writer + lines chan<- []byte +} - // Write to log and output buffer. +func (t logTee) Write(p []byte) (int, error) { + n, err := t.out.Write(p) + if n > 0 { + select { + case t.lines <- append([]byte(nil), p[:n]...): + default: + } + } + return n, err +} + +// chanReader adapts a channel of byte chunks to an io.Reader so a +// bufio.Scanner can split it into lines, without needing a pipe. +type chanReader struct { + ch <-chan []byte + buf []byte +} + +func (r *chanReader) Read(p []byte) (int, error) { + for len(r.buf) == 0 { + chunk, ok := <-r.ch + if !ok { + return 0, io.EOF + } + r.buf = chunk + } + n := copy(p, r.buf) + r.buf = r.buf[n:] + return n, nil +} + +func New(t *testing.T, r io.Reader, name string) *Expecter { copyDone := make(chan struct{}) + logDone := make(chan struct{}) out := newStdbuf() - w := io.MultiWriter(logw, out) + logLines := make(chan []byte, 256) + // Wait for drain goroutine to reach its receive loop first, + // so a burst of output at startup can't fill the buffer and + // get silently dropped before anything is listening. + logReady := make(chan struct{}) ex := &Expecter{ t: t, @@ -40,14 +79,26 @@ func New(t *testing.T, r io.Reader, name string) *Expecter { runeReader: bufio.NewReaderSize(out, utf8.UTFMax), logDone: logDone, copyDone: copyDone, - logr: logr, - logw: logw, } go func() { defer close(copyDone) - _, err := io.Copy(w, r) + defer close(logLines) + <-logReady + _, err := io.Copy(logTee{out: out, lines: logLines}, r) ex.Logf("copy done: %v", err) + if err != nil { + // out rejected a write (e.g. doMatchWithDeadline closed it + // after giving up on a match) while the command may still + // be running and writing. io.Copy stops on any destination + // error, so without this the command's next write blocks + // forever: nobody would be left reading r. Keep draining + // and discarding until the command's pipe actually closes, + // so its writes can never block on us. + ex.Logf("out closed early, draining remainder: %v", err) + _, err = io.Copy(io.Discard, r) + ex.Logf("drain done: %v", err) + } ex.Logf("closing out") err = out.closeErr(err) ex.Logf("closed out: %v", err) @@ -56,7 +107,8 @@ func New(t *testing.T, r io.Reader, name string) *Expecter { // Log all output as part of test for easier debugging on errors. go func() { defer close(logDone) - s := bufio.NewScanner(logr) + close(logReady) + s := bufio.NewScanner(&chanReader{ch: logLines}) for s.Scan() { ex.Logf("%q", stripansi.Strip(s.Text())) } @@ -104,7 +156,6 @@ type Expecter struct { runeReader *bufio.Reader copyDone, logDone chan struct{} - logr, logw io.Closer } // Rename the expecter. Make sure you set this before anything starts writing to the @@ -127,11 +178,9 @@ func (e *Expecter) Close(reason string) { case <-e.copyDone: } - e.logClose("logw", e.logw) - e.logClose("logr", e.logr) select { case <-ctx.Done(): - e.fatalf("close", "log pipe did not close in time") + e.fatalf("close", "log drain did not finish in time") return case <-e.logDone: } @@ -139,12 +188,6 @@ func (e *Expecter) Close(reason string) { e.Logf("closed expecter") } -func (e *Expecter) logClose(name string, c io.Closer) { - e.Logf("closing %s", name) - err := c.Close() - e.Logf("closed %s: %v", name, err) -} - func (e *Expecter) ExpectMatch(ctx context.Context, str string) string { return e.expectMatcherFunc(ctx, str, strings.Contains) }