From 3bdef2dfb7777ce42a9f2fdf6da7c598ccd392c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Kutryj?= Date: Wed, 22 Jul 2026 12:45:00 +0200 Subject: [PATCH 1/2] fix(shell): keep partial end marker across reads so jobs don't hang The shell executor detects command completion by scanning PTY output for the end marker (\001 ). scan() entered buffering mode on the FIRST \001 in the buffer, but that byte can be a stray 0x01 from the command's own output (e.g. binary/protobuf printed to the log). When a real end marker arrived right after such a stray byte and had not yet been fully received, the memory-bound flush dumped the ENTIRE buffer - including the partial real marker - to the job output. The marker's tail then arrived in the next read with no leading \001, was flushed as plain output, and the complete marker never appeared whole in the buffer, so scan() looped forever and the job hung until the global time limit even though the command had already exited. Fix: in buffering mode, flush only up to the LAST \001 in the buffer, retaining a possible partially-received marker for the next read. The retained tail is capped at len(endMark)+10 (the longest a real marker can be), so a pathological stream of \001 bytes cannot grow the buffer without bound. Also make start-marker detection tolerant of a bare \n terminator (not just \r\n): a previous command can leave the PTY in -onlcr mode, in which case the next start marker arrives with \n and was never recognized, producing the same hang. The end-marker regex already tolerated [\r\n]+. Reproduced by three package-local tests that drive scan() over an os.Pipe with scripted read boundaries and a timeout, so a lost marker fails fast instead of hanging the suite. Co-Authored-By: Claude Fable 5 --- pkg/shell/process.go | 62 +++++++++++--- pkg/shell/process_test.go | 168 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 220 insertions(+), 10 deletions(-) diff --git a/pkg/shell/process.go b/pkg/shell/process.go index 123e1f95..01606a6d 100644 --- a/pkg/shell/process.go +++ b/pkg/shell/process.go @@ -52,6 +52,7 @@ type Process struct { Pid int startMark string endMark string + commandStartRegex *regexp.Regexp commandEndRegex *regexp.Regexp inputBuffer []byte outputBuffer *OutputBuffer @@ -66,6 +67,7 @@ func randomMagicMark() string { func NewProcess(config Config) *Process { startMark := randomMagicMark() + "-start" endMark := randomMagicMark() + "-end" + commandStartRegex := regexp.MustCompile(startMark + `\r?\n`) commandEndRegex := regexp.MustCompile(endMark + " " + `(\d+)` + "[\r\n]+") outputBuffer, _ := NewOutputBuffer(config.OnOutput) @@ -76,6 +78,7 @@ func NewProcess(config Config) *Process { ExitCode: 1, startMark: startMark, endMark: endMark, + commandStartRegex: commandStartRegex, commandEndRegex: commandEndRegex, outputBuffer: outputBuffer, UseBase64Encoding: config.UseBase64Encoding, @@ -94,6 +97,40 @@ func (p *Process) flushInputAll() { p.flushInputBufferTill(len(p.inputBuffer)) } +// flushInputBufferKeepingMarkerTail is used in buffering mode when the buffer +// holds a \001 (a possible end-marker start) but no complete end marker has +// matched yet. It flushes buffered output but retains a trailing slice that +// could be a not-yet-complete end marker: everything from the LAST \001 onward. +// +// The leading \001 that put us in buffering mode may be a stray byte from the +// command's own output (e.g. binary data printed to the log), with the real +// marker arriving right after it. Flushing the whole buffer here would leak a +// partially-received real marker into the job output and the marker would +// never be recognized (the original hang). Keeping the tail from the last +// \001 preserves the partial marker across reads. +// +// The tail is capped at maxEndMarkerLen: a real marker is never longer than +// that, so if the tail already exceeds it the leading \001 cannot be a partial +// marker and the whole buffer is safe to flush. This keeps memory bounded even +// under a pathological stream of \001 bytes. +func (p *Process) flushInputBufferKeepingMarkerTail() { + lastMark := strings.LastIndex(string(p.inputBuffer), "\001") + + if lastMark < 0 || len(p.inputBuffer)-lastMark > p.maxEndMarkerLen() { + p.flushInputAll() + return + } + + p.flushInputBufferTill(lastMark) +} + +// maxEndMarkerLen is the largest number of bytes a complete end marker can +// occupy: "\001 " + endMark + " " + exit code digits + line terminator. The +// +10 slack covers the prefix, separators, exit code, and CR/LF. +func (p *Process) maxEndMarkerLen() int { + return len(p.endMark) + 10 +} + func (p *Process) flushInputBufferTill(index int) { if index == 0 { return @@ -415,20 +452,22 @@ func (p *Process) waitForStartMarker() error { } // - // If the inputBuffer has a start marker, the wait is done + // If the inputBuffer has a start marker, the wait is done. // - index := strings.Index(string(p.inputBuffer), p.startMark+"\r\n") - - if index >= 0 { + // The marker's line terminator may be \r\n or a bare \n (a previous + // command can leave the PTY in -onlcr mode), so we match \r?\n rather + // than a literal \r\n. + // + if loc := p.commandStartRegex.FindStringIndex(string(p.inputBuffer)); loc != nil { // - // Cut everything from the buffer before the marker + // Cut everything from the buffer up to and including the marker. // Example: // // buffer before: some test <***marker**> rest of the test // buffer after : rest of the test // - p.inputBuffer = p.inputBuffer[index+len(p.startMark)+2 : len(p.inputBuffer)] + p.inputBuffer = p.inputBuffer[loc[1]:] break } @@ -471,12 +510,15 @@ func (p *Process) scan() error { // // The buffer is much longer than the end mark, at least by 10 - // characters. + // characters, but still doesn't match the full end marker. // - // If it is not matching the full end mark, it is safe to dump. + // The leading \001 was therefore a stray byte, not our marker, so + // it is safe to dump the buffered output. We keep only the tail + // from the last \001, which may be a real marker that split across + // reads - flushing it here would lose the marker and hang the job. // - if len(p.inputBuffer) >= len(p.endMark)+10 { - p.flushInputAll() + if len(p.inputBuffer) >= p.maxEndMarkerLen() { + p.flushInputBufferKeepingMarkerTail() } } else { p.flushInputAll() diff --git a/pkg/shell/process_test.go b/pkg/shell/process_test.go index b4757eb4..c2d0cc58 100644 --- a/pkg/shell/process_test.go +++ b/pkg/shell/process_test.go @@ -4,9 +4,177 @@ import ( "fmt" "os" "strings" + "sync" "testing" + "time" ) +// scanHarness drives Process.scan() without a real bash/PTY. The shell's TTY +// is an os.Pipe, so a test can script exactly which bytes arrive in each read +// (by writing a chunk, then pausing long enough for scan to consume it and +// block on the next read). scan runs in a goroutine so a lost end marker shows +// up as a timeout instead of hanging the test process. +type scanHarness struct { + p *Process + reader *os.File + writer *os.File + scanErr chan error + outputMu sync.Mutex + output strings.Builder +} + +func newScanHarness(t *testing.T) *scanHarness { + t.Helper() + + reader, writer, err := os.Pipe() + if err != nil { + t.Fatalf("os.Pipe: %v", err) + } + + h := &scanHarness{ + reader: reader, + writer: writer, + scanErr: make(chan error, 1), + } + + shell := &Shell{TTY: reader, ExitSignal: make(chan string, 1)} + h.p = NewProcess(Config{ + Shell: shell, + StoragePath: os.TempDir(), + OnOutput: func(s string) { + h.outputMu.Lock() + h.output.WriteString(s) + h.outputMu.Unlock() + }, + }) + + t.Cleanup(func() { + _ = h.writer.Close() + _ = h.reader.Close() + }) + + return h +} + +func (h *scanHarness) run() { + go func() { h.scanErr <- h.p.scan() }() +} + +func (h *scanHarness) write(t *testing.T, chunk string) { + t.Helper() + if _, err := h.writer.WriteString(chunk); err != nil { + t.Fatalf("writing chunk to pty pipe: %v", err) + } + // Give scan time to consume this chunk and block on the next read, so the + // following chunk lands in a separate read() call. + time.Sleep(60 * time.Millisecond) +} + +func (h *scanHarness) collectedOutput() string { + h.outputMu.Lock() + defer h.outputMu.Unlock() + return h.output.String() +} + +// A stray 0x01 byte in the command's own output (e.g. binary data printed to +// the log) must not cause the real end marker to be flushed to the job output +// and lost. Before the fix, the buffering-mode flush dumped the whole buffer - +// including a partially-received real marker sitting after the stray byte - so +// the marker never appeared whole and scan() looped forever. +func Test__Scan_CompletesWhenStrayControlBytePrecedesEndMarker(t *testing.T) { + h := newScanHarness(t) + p := h.p + h.run() + + // Start marker so scan enters the end-marker loop. + h.write(t, "\001 "+p.startMark+"\r\n") + + // Command output: a stray \001, padding that pushes the buffer past the + // flush threshold, then the *partial* real end marker (no exit code yet). + h.write(t, "\001"+strings.Repeat("X", 20)+"\001 "+p.endMark) + + // Remainder of the real end marker arrives in the next read. + h.write(t, " 0\r\n") + + select { + case err := <-h.scanErr: + if err != nil { + t.Fatalf("scan returned error: %v", err) + } + if p.ExitCode != 0 { + t.Fatalf("expected exit code 0, got %d", p.ExitCode) + } + if strings.Contains(h.collectedOutput(), p.endMark) { + t.Fatalf("end marker leaked into job output: %q", h.collectedOutput()) + } + case <-time.After(5 * time.Second): + t.Fatal("scan did not complete: stray \\001 caused the end marker to be lost and the command hung") + } +} + +// A pathological stream of \001 bytes with no completing marker must not grow +// the input buffer without bound: the retained partial-marker tail is capped. +func Test__Scan_BoundsBufferOnRepeatedControlBytes(t *testing.T) { + h := newScanHarness(t) + p := h.p + h.run() + + h.write(t, "\001 "+p.startMark+"\r\n") + + // Many \001 bytes separated by junk longer than any real marker, none of + // which forms a complete end marker. + for i := 0; i < 5; i++ { + h.write(t, "\001"+strings.Repeat("Y", len(p.endMark)+20)) + } + + if got := len(p.inputBuffer); got > len(p.endMark)+10 { + t.Fatalf("input buffer not bounded: %d bytes retained (max %d)", got, len(p.endMark)+10) + } + + // A real marker still completes afterwards. + h.write(t, "\001 "+p.endMark+" 0\r\n") + + select { + case err := <-h.scanErr: + if err != nil { + t.Fatalf("scan returned error: %v", err) + } + if p.ExitCode != 0 { + t.Fatalf("expected exit code 0, got %d", p.ExitCode) + } + case <-time.After(5 * time.Second): + t.Fatal("scan did not complete after repeated control bytes") + } +} + +// The start marker may arrive terminated by a bare \n (not \r\n) when a +// previous command left the PTY in -onlcr mode. It must still be recognized. +func Test__Scan_CompletesWhenStartMarkerHasBareNewline(t *testing.T) { + h := newScanHarness(t) + p := h.p + h.run() + + // Start marker terminated by a bare \n rather than \r\n. + h.write(t, "\001 "+p.startMark+"\n") + + h.write(t, "hello\r\n\001 "+p.endMark+" 0\r\n") + + select { + case err := <-h.scanErr: + if err != nil { + t.Fatalf("scan returned error: %v", err) + } + if p.ExitCode != 0 { + t.Fatalf("expected exit code 0, got %d", p.ExitCode) + } + if !strings.Contains(h.collectedOutput(), "hello") { + t.Fatalf("expected command output to contain 'hello', got %q", h.collectedOutput()) + } + case <-time.After(5 * time.Second): + t.Fatal("scan did not complete: start marker with a bare newline was not recognized") + } +} + func Benchmark__CommandOutput_128Bytes(b *testing.B) { p := createProcess(b, fmt.Sprintf("echo '%s'", strings.Repeat("x", 128))) b.ResetTimer() From 929a680cf7879adcc2eaad3833e90b7679caafb4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Kutryj?= Date: Wed, 22 Jul 2026 14:24:30 +0200 Subject: [PATCH 2/2] fix(shell): match complete end marker; deterministic scan tests Review round 1 fixes for PR #277. Finding 1 (same-read output loss): scan() located the end marker by the first \001 then searched the whole buffer for the marker body, so when a stray \001 preceded the real marker in one read, everything between them was discarded. Match the complete marker (now including its \001 prefix) with FindStringSubmatchIndex and flush everything before the match start as ordinary output. The keep-tail buffering path is unchanged - it still retains from the last \001, which is the real marker's own prefix. Finding 2 (test timing + data race): replace the sleep-based pipe sequencing in the scan tests with a channel-driven read source, so each written chunk is returned as exactly one read() with deterministic boundaries and no goroutine races on inputBuffer. Adds a nil-defaulted readSource seam on Process for tests to drive reads directly. Also fixes a pre-existing data race the deterministic harness exposed: OutputBuffer.IsEmpty() read the shared byte slice without the mutex, and done was read/written without synchronization. Guard IsEmpty() with the lock and make done an atomic flag. Needed for the scan tests to pass under -race. Scan tests pass under `go test -race ./pkg/shell/ -count=5`. Co-Authored-By: Claude Fable 5 --- pkg/shell/output_buffer.go | 13 ++-- pkg/shell/process.go | 63 +++++++++++++------- pkg/shell/process_test.go | 118 ++++++++++++++++++++++++++++--------- 3 files changed, 140 insertions(+), 54 deletions(-) diff --git a/pkg/shell/output_buffer.go b/pkg/shell/output_buffer.go index c09854a1..7d6ce8fc 100644 --- a/pkg/shell/output_buffer.go +++ b/pkg/shell/output_buffer.go @@ -6,6 +6,7 @@ import ( "math" "strings" "sync" + "sync/atomic" "time" "unicode/utf8" @@ -38,7 +39,7 @@ type OutputBuffer struct { Consumer func(string) bytes []byte mu sync.Mutex - done bool + done atomic.Bool lastAppend *time.Time flushTimeout time.Duration } @@ -73,6 +74,8 @@ func (b *OutputBuffer) Append(bytes []byte) { } func (b *OutputBuffer) IsEmpty() bool { + b.mu.Lock() + defer b.mu.Unlock() return len(b.bytes) == 0 } @@ -80,7 +83,7 @@ func (b *OutputBuffer) Flush() { backoffStrategy := b.exponentialBackoff() for { - if b.done { + if b.done.Load() { log.Debugf("The output buffer was closed - stopping") break } @@ -157,7 +160,7 @@ func (b *OutputBuffer) flush() { * is because we are cutting it here to fit it into the chunk. */ - if !b.done || (b.done && cutLength == chunkSize) { + if !b.done.Load() || (b.done.Load() && cutLength == chunkSize) { for i := 0; i < 4; i++ { if utf8.Valid(b.bytes[0:cutLength]) { break @@ -227,7 +230,7 @@ func (b *OutputBuffer) timeSinceLastAppend() time.Duration { func (b *OutputBuffer) chunkSize() int { // If the output buffer was already closed, we should // use bigger chunks to make sure we can flush everything left in time. - if b.done { + if b.done.Load() { return OutputBufferDefaultCutLength * 10 } @@ -235,7 +238,7 @@ func (b *OutputBuffer) chunkSize() int { } func (b *OutputBuffer) Close() error { - b.done = true + b.done.Store(true) ctx, cancelFunc := context.WithTimeout(context.Background(), b.flushTimeout) defer cancelFunc() diff --git a/pkg/shell/process.go b/pkg/shell/process.go index 01606a6d..f6745a60 100644 --- a/pkg/shell/process.go +++ b/pkg/shell/process.go @@ -58,6 +58,10 @@ type Process struct { outputBuffer *OutputBuffer SysProcAttr *syscall.SysProcAttr UseBase64Encoding bool + + // readSource, when non-nil, replaces reading from the shell TTY. Tests set + // it to drive scan() one read at a time with deterministic read boundaries. + readSource func() ([]byte, error) } func randomMagicMark() string { @@ -68,7 +72,12 @@ func NewProcess(config Config) *Process { startMark := randomMagicMark() + "-start" endMark := randomMagicMark() + "-end" commandStartRegex := regexp.MustCompile(startMark + `\r?\n`) - commandEndRegex := regexp.MustCompile(endMark + " " + `(\d+)` + "[\r\n]+") + + // The end marker is emitted as `\001 \n` (see + // constructShellInstruction). Matching it *including* the \001 prefix lets + // scan locate the complete marker anywhere in the buffer and flush anything + // before it - even a stray \001 - as ordinary output rather than dropping it. + commandEndRegex := regexp.MustCompile("\001 " + endMark + " " + `(\d+)` + "[\r\n]+") outputBuffer, _ := NewOutputBuffer(config.OnOutput) return &Process{ @@ -424,6 +433,12 @@ func (p *Process) readNonPTY(reader *io.PipeReader, done chan bool) { // Read state from shell into the inputBuffer func (p *Process) read() error { + if p.readSource != nil { + data, err := p.readSource() + p.inputBuffer = append(p.inputBuffer, data...) + return err + } + buffer := make([]byte, p.readBufferSize()) log.Debug("Reading started") @@ -493,29 +508,37 @@ func (p *Process) scan() error { exitCode := "" for { - if index := p.endMarkerHeaderIndex(); index >= 0 { - if index > 0 { - // publish everything until the end mark - p.flushInputBufferTill(index) - } - - log.Debug("Start of end marker detected, entering buffering mode.") - - if match := p.commandEndRegex.FindStringSubmatch(string(p.inputBuffer)); len(match) == 2 { - log.Debug("End marker detected. Exit code: ", match[1]) + // + // A complete end marker (with its \001 prefix) ends the command. + // Matching the whole marker means anything before it - including a + // stray \001 emitted by the command's own output - is flushed as + // ordinary output instead of being discarded. + // + buffered := string(p.inputBuffer) + if loc := p.commandEndRegex.FindStringSubmatchIndex(buffered); loc != nil { + exitCode = buffered[loc[2]:loc[3]] + log.Debug("End marker detected. Exit code: ", exitCode) - exitCode = match[1] - break - } + // Everything before the marker is genuine command output. + p.flushInputBufferTill(loc[0]) + break + } + if index := p.endMarkerHeaderIndex(); index >= 0 { + // + // A \001 is present but no complete marker has arrived yet - it may + // still be streaming in across reads. Flush everything before the + // first \001 as ordinary output and stay in buffering mode. // - // The buffer is much longer than the end mark, at least by 10 - // characters, but still doesn't match the full end marker. + log.Debug("Start of end marker detected, entering buffering mode.") + p.flushInputBufferTill(index) + // - // The leading \001 was therefore a stray byte, not our marker, so - // it is safe to dump the buffered output. We keep only the tail - // from the last \001, which may be a real marker that split across - // reads - flushing it here would lose the marker and hang the job. + // The buffer has grown past the longest possible marker without + // matching, so the leading \001 was a stray byte. Flush it, keeping + // only the tail from the last \001, which may be a real marker that + // split across reads - flushing it here would lose the marker and + // hang the job. // if len(p.inputBuffer) >= p.maxEndMarkerLen() { p.flushInputBufferKeepingMarkerTail() diff --git a/pkg/shell/process_test.go b/pkg/shell/process_test.go index c2d0cc58..a9177829 100644 --- a/pkg/shell/process_test.go +++ b/pkg/shell/process_test.go @@ -2,6 +2,7 @@ package shell import ( "fmt" + "io" "os" "strings" "sync" @@ -9,16 +10,24 @@ import ( "time" ) -// scanHarness drives Process.scan() without a real bash/PTY. The shell's TTY -// is an os.Pipe, so a test can script exactly which bytes arrive in each read -// (by writing a chunk, then pausing long enough for scan to consume it and -// block on the next read). scan runs in a goroutine so a lost end marker shows -// up as a timeout instead of hanging the test process. +// scanHarness drives Process.scan() without a real bash/PTY by replacing the +// process's read source with a channel. Each chunk the test writes is returned +// as exactly one read() call, so read boundaries are deterministic - no sleeps, +// no reliance on pipe scheduling. scan runs in a goroutine so a lost end marker +// shows up as a timeout instead of hanging the test process. +// +// Handshake: before every read, scan signals readReady, then blocks receiving +// the next chunk. write() waits for that signal before sending, guaranteeing +// the chunk lands in the read scan is about to perform. While scan is parked +// waiting for a chunk it is not touching inputBuffer, so a test may inspect +// inputBuffer between readReady and the next write without racing the scanner. type scanHarness struct { - p *Process - reader *os.File - writer *os.File - scanErr chan error + p *Process + scanErr chan error + chunks chan []byte + readReady chan struct{} + closeOnce sync.Once + outputMu sync.Mutex output strings.Builder } @@ -26,18 +35,13 @@ type scanHarness struct { func newScanHarness(t *testing.T) *scanHarness { t.Helper() - reader, writer, err := os.Pipe() - if err != nil { - t.Fatalf("os.Pipe: %v", err) - } - h := &scanHarness{ - reader: reader, - writer: writer, - scanErr: make(chan error, 1), + scanErr: make(chan error, 1), + chunks: make(chan []byte), + readReady: make(chan struct{}, 1), } - shell := &Shell{TTY: reader, ExitSignal: make(chan string, 1)} + shell := &Shell{ExitSignal: make(chan string, 1)} h.p = NewProcess(Config{ Shell: shell, StoragePath: os.TempDir(), @@ -48,9 +52,19 @@ func newScanHarness(t *testing.T) *scanHarness { }, }) + h.p.readSource = func() ([]byte, error) { + h.readReady <- struct{}{} + chunk, ok := <-h.chunks + if !ok { + return nil, io.EOF + } + return chunk, nil + } + + // If a test bails out before feeding a completing marker, unblock the + // parked scanner so its goroutine exits instead of leaking. t.Cleanup(func() { - _ = h.writer.Close() - _ = h.reader.Close() + h.closeOnce.Do(func() { close(h.chunks) }) }) return h @@ -60,14 +74,21 @@ func (h *scanHarness) run() { go func() { h.scanErr <- h.p.scan() }() } +// awaitRead blocks until scan is parked waiting for its next read(). +func (h *scanHarness) awaitRead() { + <-h.readReady +} + +// feed hands scan the bytes for the read it is currently waiting on. It must be +// called only after awaitRead (write() does both). +func (h *scanHarness) feed(chunk string) { + h.chunks <- []byte(chunk) +} + func (h *scanHarness) write(t *testing.T, chunk string) { t.Helper() - if _, err := h.writer.WriteString(chunk); err != nil { - t.Fatalf("writing chunk to pty pipe: %v", err) - } - // Give scan time to consume this chunk and block on the next read, so the - // following chunk lands in a separate read() call. - time.Sleep(60 * time.Millisecond) + h.awaitRead() + h.feed(chunk) } func (h *scanHarness) collectedOutput() string { @@ -112,6 +133,42 @@ func Test__Scan_CompletesWhenStrayControlBytePrecedesEndMarker(t *testing.T) { } } +// When a stray \001, ordinary command output, and the complete end marker all +// arrive in a single read, scan must locate the *whole* marker (not the first +// \001) so the output before the marker is published rather than discarded. +// Only the marker itself is stripped. +func Test__Scan_PreservesOutputAroundStrayControlByteInSameRead(t *testing.T) { + h := newScanHarness(t) + p := h.p + h.run() + + h.write(t, "\001 "+p.startMark+"\r\n") + + // One read: a stray \001 inside ordinary output, followed by the complete + // real end marker. "hello\001world " is genuine output; the marker begins + // at the second \001 (the only \001 followed by " "). + h.write(t, "hello\001world \001 "+p.endMark+" 0\r\n") + + select { + case err := <-h.scanErr: + if err != nil { + t.Fatalf("scan returned error: %v", err) + } + if p.ExitCode != 0 { + t.Fatalf("expected exit code 0, got %d", p.ExitCode) + } + out := h.collectedOutput() + if !strings.Contains(out, "hello\001world ") { + t.Fatalf("ordinary output around the stray \\001 was lost: %q", out) + } + if strings.Contains(out, p.endMark) { + t.Fatalf("end marker leaked into job output: %q", out) + } + case <-time.After(5 * time.Second): + t.Fatal("scan did not complete when a stray \\001 preceded the end marker in one read") + } +} + // A pathological stream of \001 bytes with no completing marker must not grow // the input buffer without bound: the retained partial-marker tail is capped. func Test__Scan_BoundsBufferOnRepeatedControlBytes(t *testing.T) { @@ -127,12 +184,15 @@ func Test__Scan_BoundsBufferOnRepeatedControlBytes(t *testing.T) { h.write(t, "\001"+strings.Repeat("Y", len(p.endMark)+20)) } - if got := len(p.inputBuffer); got > len(p.endMark)+10 { - t.Fatalf("input buffer not bounded: %d bytes retained (max %d)", got, len(p.endMark)+10) + // Wait until scan has consumed all of the above and parked on the next + // read. It is not touching inputBuffer now, so this read is race-free. + h.awaitRead() + if got := len(p.inputBuffer); got > p.maxEndMarkerLen() { + t.Fatalf("input buffer not bounded: %d bytes retained (max %d)", got, p.maxEndMarkerLen()) } // A real marker still completes afterwards. - h.write(t, "\001 "+p.endMark+" 0\r\n") + h.feed("\001 " + p.endMark + " 0\r\n") select { case err := <-h.scanErr: