From ba2ed3f6de7a28eb52eb310d2b270cb30ed33490 Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 22:10:10 +0100 Subject: [PATCH 1/3] fix: don't hang (and lose the result) when the output consumer stalls (#121) If stdout/stderr is a pipe whose consumer stops draining (a wedged terminal, an SSH / run-remote stream stall, a full CI log buffer), ftsb wedged inside a blocking log.Printf: the unbounded progress reporter, or summary() (which logs before it writes the --json-out-file result), blocked forever -- the process never exited and no result file was written. Operationally worst case in the on-the-fly-EC2 model: a hung run yields no result AND keeps the bench-server billing until manually destroyed. Route all console logging through a non-blocking best-effort writer: log lines are buffered and dropped (whole lines only) rather than blocking the benchmark when the consumer stalls; a detached drain goroutine does the actual (possibly blocking) writes and is reaped at exit. The --log-file, if any, is written directly (a regular file does not stall) so it stays complete. Normal runs are unaffected (the drain keeps up, nothing drops). Validated: with a never-drained os.Pipe consumer, master HANGS with no result; this branch COMPLETES and writes the JSON. Adds a unit test (Write never blocks on a stalled underlying writer; delivers everything when drained) and an integration test (ftsb completes + writes the result under a stalled pipe). --- benchmark_runner/redisearch_test.go | 46 +++++++++++++++ cmd/ftsb_redisearch/main.go | 17 +++--- cmd/ftsb_redisearch/nonblocking_writer.go | 48 ++++++++++++++++ .../nonblocking_writer_test.go | 57 +++++++++++++++++++ 4 files changed, 160 insertions(+), 8 deletions(-) create mode 100644 cmd/ftsb_redisearch/nonblocking_writer.go create mode 100644 cmd/ftsb_redisearch/nonblocking_writer_test.go diff --git a/benchmark_runner/redisearch_test.go b/benchmark_runner/redisearch_test.go index 711a12f..5e3d4d0 100644 --- a/benchmark_runner/redisearch_test.go +++ b/benchmark_runner/redisearch_test.go @@ -638,3 +638,49 @@ func TestFTSBLatencyCapHighOptIn(t *testing.T) { t.Errorf("Expected q100 >= 1500 ms with --max-latency-seconds=60 (DEBUG SLEEP 2 should record ~2000 ms), got %.2f ms", q100) } } + +// Regression guard for #121: a stalled output consumer must not prevent the run +// from completing and writing its --json-out-file result. Output goes to a pipe +// that is never drained; the fast reporting period fills the 64KB pipe buffer +// mid-run. Pre-fix, the blocking progress log wedged the process and no result +// file was written (the test times out). +func TestFTSBCompletesWithStalledOutputConsumer(t *testing.T) { + startRedisContainer(t) + + pr, pw, err := os.Pipe() + if err != nil { + t.Fatalf("pipe: %v", err) + } + defer pr.Close() // held open, never read -> the pipe fills and stays full + + jsonPath := "../testdata/results.stalled.json" + _ = os.Remove(jsonPath) + + cmd := exec.Command("../bin/ftsb_redisearch", + "--input", "../testdata/minimal.csv", + "--duration", "3s", + "--reporting-period", "1ms", // thousands of progress lines -> overflows the pipe buffer + "--json-out-file", jsonPath, + ) + cmd.Stdout = pw + cmd.Stderr = pw + if err := cmd.Start(); err != nil { + t.Fatalf("start: %v", err) + } + pw.Close() // the child holds its own fd; drop the parent's copy + + done := make(chan error, 1) + go func() { done <- cmd.Wait() }() + select { + case <-done: + // completed despite the never-drained consumer + case <-time.After(45 * time.Second): + _ = cmd.Process.Kill() + t.Fatal("ftsb hung with a stalled output consumer (never exited)") + } + + if _, statErr := os.Stat(jsonPath); statErr != nil { + t.Fatalf("result file not written under a stalled consumer: %v", statErr) + } + _ = os.Remove(jsonPath) +} diff --git a/cmd/ftsb_redisearch/main.go b/cmd/ftsb_redisearch/main.go index d5407df..7f93d8f 100644 --- a/cmd/ftsb_redisearch/main.go +++ b/cmd/ftsb_redisearch/main.go @@ -106,21 +106,22 @@ func main() { git_dirty_str = "-dirty" } - // Setup log file if specified + // Route console logging through a non-blocking writer so a stalled output + // consumer (wedged terminal, run-remote/SSH stream stall, full CI buffer) + // can never wedge the benchmark and prevent the result from being written + // (issue #121). The log file, if any, is written directly -- a regular file + // does not stall, and keeping it off the drop path leaves it complete. + console := newNonBlockingWriter(os.Stderr, 1024) if logFile != "" { f, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) if err != nil { log.Fatalf("Failed to open log file %s: %v", logFile, err) } defer f.Close() - - // Create a multi-writer that writes to both stdout and the log file - multiWriter := io.MultiWriter(os.Stdout, f) - - // Redirect log output to both stdout and file - log.SetOutput(multiWriter) - + log.SetOutput(io.MultiWriter(console, f)) log.Printf("Logging to file: %s\n", logFile) + } else { + log.SetOutput(console) } log.Printf("ftsb (git_sha1:%s%s)\n", git_sha, git_dirty_str) diff --git a/cmd/ftsb_redisearch/nonblocking_writer.go b/cmd/ftsb_redisearch/nonblocking_writer.go new file mode 100644 index 0000000..988212a --- /dev/null +++ b/cmd/ftsb_redisearch/nonblocking_writer.go @@ -0,0 +1,48 @@ +package main + +import "io" + +// nonBlockingWriter forwards whole log lines to an underlying writer through a +// bounded buffer, dropping output rather than blocking when the consumer stalls. +// +// ftsb's progress reporter logs unboundedly (one line per --reporting-period) +// and summary() logs before it writes the --json-out-file result. If stdout/ +// stderr is a pipe whose consumer stops draining (a wedged terminal, an SSH / +// run-remote stream stall, a full CI log buffer), a plain blocking log.Printf +// wedges the whole run: the process never exits and the result is never written +// (issue #121). Routing console logs through this writer makes every log.Printf +// non-blocking, so the benchmark always completes, writes its result, and exits. +// The background drain goroutine may block on the stalled consumer, but it is +// detached and touches no benchmark state, so it is reaped harmlessly at exit. +type nonBlockingWriter struct { + ch chan []byte +} + +// newNonBlockingWriter starts a drain goroutine that forwards buffered lines to +// w. bufLines is the number of pending lines tolerated before new lines are +// dropped. +func newNonBlockingWriter(w io.Writer, bufLines int) *nonBlockingWriter { + nb := &nonBlockingWriter{ch: make(chan []byte, bufLines)} + go func() { + for b := range nb.ch { + // A stalled consumer blocks here; that's fine -- only this detached + // goroutine waits, never the benchmark. Write errors are ignored, + // matching the standard log package's best-effort semantics. + _, _ = w.Write(b) + } + }() + return nb +} + +// Write never blocks. The log package calls Write once per fully-formatted line, +// so a dropped write loses a whole line, never a partial one. The line is copied +// because log reuses its formatting buffer across calls. +func (nb *nonBlockingWriter) Write(p []byte) (int, error) { + b := make([]byte, len(p)) + copy(b, p) + select { + case nb.ch <- b: + default: // buffer full (consumer stalled) -> drop this line + } + return len(p), nil +} diff --git a/cmd/ftsb_redisearch/nonblocking_writer_test.go b/cmd/ftsb_redisearch/nonblocking_writer_test.go new file mode 100644 index 0000000..a536806 --- /dev/null +++ b/cmd/ftsb_redisearch/nonblocking_writer_test.go @@ -0,0 +1,57 @@ +package main + +import ( + "sync/atomic" + "testing" + "time" +) + +// blockingWriter blocks every Write until released -- models a stalled console. +type blockingWriter struct{ release chan struct{} } + +func (b *blockingWriter) Write(p []byte) (int, error) { <-b.release; return len(p), nil } + +// The whole point of #121: Write must never block the caller even when the +// underlying consumer is wedged. +func TestNonBlockingWriterNeverBlocks(t *testing.T) { + bw := &blockingWriter{release: make(chan struct{})} + nb := newNonBlockingWriter(bw, 4) + + done := make(chan struct{}) + go func() { + for i := 0; i < 100000; i++ { // far more than the buffer -> excess is dropped + _, _ = nb.Write([]byte("progress line\n")) + } + close(done) + }() + + select { + case <-done: + case <-time.After(3 * time.Second): + close(bw.release) + t.Fatal("Write blocked while the underlying writer was stalled") + } + close(bw.release) // let the drain goroutine exit +} + +type countingWriter struct{ n int64 } + +func (c *countingWriter) Write(p []byte) (int, error) { atomic.AddInt64(&c.n, 1); return len(p), nil } + +// When the consumer keeps up, no lines are dropped and order/content is intact. +func TestNonBlockingWriterDeliversWhenDrained(t *testing.T) { + cw := &countingWriter{} + nb := newNonBlockingWriter(cw, 1024) + const lines = 500 + for i := 0; i < lines; i++ { + _, _ = nb.Write([]byte("x")) + } + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if atomic.LoadInt64(&cw.n) == lines { + return // all delivered + } + time.Sleep(5 * time.Millisecond) + } + t.Fatalf("delivered %d lines, want %d (nothing should drop when the consumer drains)", atomic.LoadInt64(&cw.n), lines) +} From e0b76f2c0e10696b474adb36e0054d178b0da6ff Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 22:28:12 +0100 Subject: [PATCH 2/3] fix: flush non-blocking console writer on exit so the summary isn't lost The initial #121 fix routed console logging through a fire-and-forget writer whose drain goroutine wrote asynchronously. On a clean run RunBenchmark returns normally (no os.Exit), so the process could exit before the goroutine flushed the final buffered lines -- dropping the "Issued ..."/Summary output. This broke TestFTSBWithDuration and TestFTSBWithBatchSize, which assert the summary reaches stderr. Add nonBlockingWriter.Close(timeout): it stops accepting lines and waits for the drain goroutine to flush the buffer, bounded by a timeout so a genuinely stalled consumer still can't wedge shutdown. main defers Close(2s) after RunBenchmark, so a healthy consumer gets the complete summary while the stalled-consumer guarantee is preserved. A small mutex makes Write/Close race-free and makes post-Close writes drop instead of panicking on a closed channel. Tests: TestNonBlockingWriterCloseFlushesBufferedLines (healthy -> tail delivered), TestNonBlockingWriterCloseReturnsDespiteStall (stalled -> returns within timeout), TestNonBlockingWriterWriteAfterCloseIsSafe. The two integration tests and TestFTSBCompletesWithStalledOutputConsumer all pass. --- cmd/ftsb_redisearch/main.go | 4 ++ cmd/ftsb_redisearch/nonblocking_writer.go | 56 ++++++++++++++++--- .../nonblocking_writer_test.go | 46 +++++++++++++++ 3 files changed, 98 insertions(+), 8 deletions(-) diff --git a/cmd/ftsb_redisearch/main.go b/cmd/ftsb_redisearch/main.go index 7f93d8f..37aea8c 100644 --- a/cmd/ftsb_redisearch/main.go +++ b/cmd/ftsb_redisearch/main.go @@ -112,6 +112,10 @@ func main() { // (issue #121). The log file, if any, is written directly -- a regular file // does not stall, and keeping it off the drop path leaves it complete. console := newNonBlockingWriter(os.Stderr, 1024) + // Flush buffered console lines (notably the final summary) before exit. On a + // healthy consumer this delivers everything; on a stalled one it returns + // after the timeout so shutdown still can't hang. + defer console.Close(2 * time.Second) if logFile != "" { f, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) if err != nil { diff --git a/cmd/ftsb_redisearch/nonblocking_writer.go b/cmd/ftsb_redisearch/nonblocking_writer.go index 988212a..7d6410d 100644 --- a/cmd/ftsb_redisearch/nonblocking_writer.go +++ b/cmd/ftsb_redisearch/nonblocking_writer.go @@ -1,6 +1,10 @@ package main -import "io" +import ( + "io" + "sync" + "time" +) // nonBlockingWriter forwards whole log lines to an underlying writer through a // bounded buffer, dropping output rather than blocking when the consumer stalls. @@ -12,18 +16,29 @@ import "io" // wedges the whole run: the process never exits and the result is never written // (issue #121). Routing console logs through this writer makes every log.Printf // non-blocking, so the benchmark always completes, writes its result, and exits. -// The background drain goroutine may block on the stalled consumer, but it is -// detached and touches no benchmark state, so it is reaped harmlessly at exit. +// +// On a HEALTHY consumer nothing is lost: the drain goroutine keeps up, so the +// buffer never fills, and Close() flushes the tail (including the final summary) +// before the process exits. On a STALLED consumer Close() still returns within +// its timeout, so shutdown can never hang. type nonBlockingWriter struct { - ch chan []byte + ch chan []byte + done chan struct{} // closed when the drain goroutine has finished + + mu sync.Mutex // guards closed + the send on ch; never held across w.Write + closed bool } // newNonBlockingWriter starts a drain goroutine that forwards buffered lines to // w. bufLines is the number of pending lines tolerated before new lines are // dropped. func newNonBlockingWriter(w io.Writer, bufLines int) *nonBlockingWriter { - nb := &nonBlockingWriter{ch: make(chan []byte, bufLines)} + nb := &nonBlockingWriter{ + ch: make(chan []byte, bufLines), + done: make(chan struct{}), + } go func() { + defer close(nb.done) for b := range nb.ch { // A stalled consumer blocks here; that's fine -- only this detached // goroutine waits, never the benchmark. Write errors are ignored, @@ -40,9 +55,34 @@ func newNonBlockingWriter(w io.Writer, bufLines int) *nonBlockingWriter { func (nb *nonBlockingWriter) Write(p []byte) (int, error) { b := make([]byte, len(p)) copy(b, p) - select { - case nb.ch <- b: - default: // buffer full (consumer stalled) -> drop this line + nb.mu.Lock() + if !nb.closed { + select { + case nb.ch <- b: + default: // buffer full (consumer stalled) -> drop this line + } } + nb.mu.Unlock() return len(p), nil } + +// Close stops accepting new lines and waits up to timeout for the drain +// goroutine to flush whatever is still buffered. On a healthy consumer this +// delivers the final lines (notably the summary) before the process exits; on a +// stalled consumer it returns after timeout so shutdown never hangs (#121). +// Safe to call more than once; writes after Close are silently dropped. +func (nb *nonBlockingWriter) Close(timeout time.Duration) { + nb.mu.Lock() + if nb.closed { + nb.mu.Unlock() + return + } + nb.closed = true + close(nb.ch) + nb.mu.Unlock() + + select { + case <-nb.done: + case <-time.After(timeout): + } +} diff --git a/cmd/ftsb_redisearch/nonblocking_writer_test.go b/cmd/ftsb_redisearch/nonblocking_writer_test.go index a536806..587898d 100644 --- a/cmd/ftsb_redisearch/nonblocking_writer_test.go +++ b/cmd/ftsb_redisearch/nonblocking_writer_test.go @@ -55,3 +55,49 @@ func TestNonBlockingWriterDeliversWhenDrained(t *testing.T) { } t.Fatalf("delivered %d lines, want %d (nothing should drop when the consumer drains)", atomic.LoadInt64(&cw.n), lines) } + +// Close must flush buffered lines to a healthy consumer before returning -- this +// is what guarantees the final summary line reaches stderr before the process +// exits (the #122 CI regression: "Issued ..." was queued but never flushed). +func TestNonBlockingWriterCloseFlushesBufferedLines(t *testing.T) { + cw := &countingWriter{} + nb := newNonBlockingWriter(cw, 1024) + const lines = 200 + for i := 0; i < lines; i++ { + _, _ = nb.Write([]byte("x")) + } + nb.Close(2 * time.Second) + if got := atomic.LoadInt64(&cw.n); got != lines { + t.Fatalf("after Close delivered %d lines, want %d (Close must flush the tail)", got, lines) + } +} + +// Close must return within its timeout even if the consumer is wedged, so +// shutdown can never hang on a stalled output stream. +func TestNonBlockingWriterCloseReturnsDespiteStall(t *testing.T) { + bw := &blockingWriter{release: make(chan struct{})} + nb := newNonBlockingWriter(bw, 4) + for i := 0; i < 100; i++ { // fill + overflow the buffer + _, _ = nb.Write([]byte("x")) + } + done := make(chan struct{}) + go func() { nb.Close(200 * time.Millisecond); close(done) }() + select { + case <-done: + case <-time.After(3 * time.Second): + close(bw.release) + t.Fatal("Close blocked while the underlying writer was stalled") + } + close(bw.release) // let the drain goroutine exit +} + +// Writes after Close are dropped, not panicking on a send to a closed channel. +func TestNonBlockingWriterWriteAfterCloseIsSafe(t *testing.T) { + cw := &countingWriter{} + nb := newNonBlockingWriter(cw, 16) + nb.Close(time.Second) + if _, err := nb.Write([]byte("late")); err != nil { + t.Fatalf("Write after Close returned error: %v", err) + } + nb.Close(time.Second) // idempotent +} From 208659b62d31540c545684de6e5c1bcc4f28645e Mon Sep 17 00:00:00 2001 From: fcostaoliveira Date: Mon, 13 Jul 2026 22:58:09 +0100 Subject: [PATCH 3/3] fix: deliver console output synchronously so fatal/summary lines aren't lost (#121) Adversarial review found the async fire-and-forget writer dropped log.Fatal diagnostics: on the default (no --log-file) path, log.SetOutput(console) routed fatal messages through the buffered channel, and log.Fatalf enqueues then calls os.Exit(1) -- which bypasses the deferred console.Close AND the detached drain goroutine, so the process exited before the message reached stderr. A missing --input or an unwritable --json-out-file exited 1 with no reason (a regression: pre-#121 log defaulted to synchronous os.Stderr). The same weakness left the Close-flush test racy (it could pass against a Close that never flushed). os.Stderr is in blocking mode, so its write can't be bounded with a deadline. Instead the blocking write stays on the drain goroutine, but Write now WAITS for its line to be written, bounded by writeTimeout. On a healthy consumer delivery is synchronous, so the summary and any pre-os.Exit fatal line reach stderr before exit; on a stalled consumer Write returns after writeTimeout (then drops once the buffer fills), so the run still completes and writes its result. This removes the Close/flush machinery (and its racy test) entirely. Tests: DeliversBeforeReturnOnHealthy (line on the writer by the time Write returns -> fatal-safe), DeliversAllWhenDrained (nothing dropped when keeping up), DoesNotBlockOnStall (bounded under a wedged writer). Verified E2E: `--input missing.csv` now prints "cannot open file for read ..." to stderr and exits 1. The two integration tests that caught the original drop (TestFTSBWithDuration, TestFTSBWithBatchSize) and the stalled-consumer test all pass under -race. --- cmd/ftsb_redisearch/main.go | 10 +- cmd/ftsb_redisearch/nonblocking_writer.go | 97 ++++++++--------- .../nonblocking_writer_test.go | 101 ++++++++---------- 3 files changed, 96 insertions(+), 112 deletions(-) diff --git a/cmd/ftsb_redisearch/main.go b/cmd/ftsb_redisearch/main.go index 37aea8c..f76ee4b 100644 --- a/cmd/ftsb_redisearch/main.go +++ b/cmd/ftsb_redisearch/main.go @@ -111,11 +111,11 @@ func main() { // can never wedge the benchmark and prevent the result from being written // (issue #121). The log file, if any, is written directly -- a regular file // does not stall, and keeping it off the drop path leaves it complete. - console := newNonBlockingWriter(os.Stderr, 1024) - // Flush buffered console lines (notably the final summary) before exit. On a - // healthy consumer this delivers everything; on a stalled one it returns - // after the timeout so shutdown still can't hang. - defer console.Close(2 * time.Second) + // On a healthy consumer each Write is delivered synchronously (so the summary + // and any pre-os.Exit fatal message reach stderr); on a stalled consumer a + // Write waits at most writeTimeout and then drops, so the run still completes + // and writes its result (issue #121). + console := newNonBlockingWriter(os.Stderr, 1024, 2*time.Second) if logFile != "" { f, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) if err != nil { diff --git a/cmd/ftsb_redisearch/nonblocking_writer.go b/cmd/ftsb_redisearch/nonblocking_writer.go index 7d6410d..fad8111 100644 --- a/cmd/ftsb_redisearch/nonblocking_writer.go +++ b/cmd/ftsb_redisearch/nonblocking_writer.go @@ -2,87 +2,84 @@ package main import ( "io" - "sync" "time" ) -// nonBlockingWriter forwards whole log lines to an underlying writer through a -// bounded buffer, dropping output rather than blocking when the consumer stalls. +// nonBlockingWriter forwards whole log lines to an underlying writer on a +// dedicated drain goroutine, so a stalled consumer can never wedge the caller. // // ftsb's progress reporter logs unboundedly (one line per --reporting-period) // and summary() logs before it writes the --json-out-file result. If stdout/ // stderr is a pipe whose consumer stops draining (a wedged terminal, an SSH / // run-remote stream stall, a full CI log buffer), a plain blocking log.Printf // wedges the whole run: the process never exits and the result is never written -// (issue #121). Routing console logs through this writer makes every log.Printf -// non-blocking, so the benchmark always completes, writes its result, and exits. +// (issue #121). os.Stderr is in blocking mode, so its write can't be bounded +// with a deadline; instead the blocking write happens on the drain goroutine and +// Write only waits for delivery up to writeTimeout. // -// On a HEALTHY consumer nothing is lost: the drain goroutine keeps up, so the -// buffer never fills, and Close() flushes the tail (including the final summary) -// before the process exits. On a STALLED consumer Close() still returns within -// its timeout, so shutdown can never hang. +// On a HEALTHY consumer the drain goroutine writes each line immediately, so +// Write returns only after the line has actually reached the underlying writer. +// That matters for messages logged right before the process exits -- notably the +// final summary and any log.Fatal message, which is emitted via log.Output and +// then os.Exit(1), bypassing any deferred flush. Because Write is synchronous on +// a healthy consumer, those lines are delivered before the exit. +// +// On a STALLED consumer Write waits at most writeTimeout and then returns +// (dropping delivery of that line) so the benchmark still completes and writes +// its result; once the small buffer fills, further lines are dropped without +// waiting at all. type nonBlockingWriter struct { - ch chan []byte - done chan struct{} // closed when the drain goroutine has finished + ch chan writeReq + writeTimeout time.Duration +} - mu sync.Mutex // guards closed + the send on ch; never held across w.Write - closed bool +type writeReq struct { + b []byte + done chan struct{} // closed by the drain goroutine once b has been written } // newNonBlockingWriter starts a drain goroutine that forwards buffered lines to -// w. bufLines is the number of pending lines tolerated before new lines are -// dropped. -func newNonBlockingWriter(w io.Writer, bufLines int) *nonBlockingWriter { +// w. bufLines is how many pending lines are tolerated before a stalled consumer +// causes new lines to be dropped; writeTimeout bounds how long a single Write +// waits for its line to be delivered before giving up. +func newNonBlockingWriter(w io.Writer, bufLines int, writeTimeout time.Duration) *nonBlockingWriter { nb := &nonBlockingWriter{ - ch: make(chan []byte, bufLines), - done: make(chan struct{}), + ch: make(chan writeReq, bufLines), + writeTimeout: writeTimeout, } go func() { - defer close(nb.done) - for b := range nb.ch { + for req := range nb.ch { // A stalled consumer blocks here; that's fine -- only this detached // goroutine waits, never the benchmark. Write errors are ignored, // matching the standard log package's best-effort semantics. - _, _ = w.Write(b) + _, _ = w.Write(req.b) + close(req.done) } }() return nb } -// Write never blocks. The log package calls Write once per fully-formatted line, -// so a dropped write loses a whole line, never a partial one. The line is copied -// because log reuses its formatting buffer across calls. +// Write copies the line (the log package reuses its formatting buffer), hands it +// to the drain goroutine, and waits up to writeTimeout for it to be written. It +// never blocks longer than writeTimeout, and never loses a partial line -- the +// log package calls Write once per fully-formatted line, so a dropped line is +// dropped whole. func (nb *nonBlockingWriter) Write(p []byte) (int, error) { b := make([]byte, len(p)) copy(b, p) - nb.mu.Lock() - if !nb.closed { + req := writeReq{b: b, done: make(chan struct{})} + select { + case nb.ch <- req: + // Delivered to the drain goroutine; wait (bounded) for it to be written + // so healthy-consumer output -- including a pre-os.Exit fatal line -- is + // actually flushed before we return. select { - case nb.ch <- b: - default: // buffer full (consumer stalled) -> drop this line + case <-req.done: + case <-time.After(nb.writeTimeout): } + default: + // Buffer full: the consumer is stalled and we're already behind. Drop + // this line rather than wait, so the benchmark keeps making progress. } - nb.mu.Unlock() return len(p), nil } - -// Close stops accepting new lines and waits up to timeout for the drain -// goroutine to flush whatever is still buffered. On a healthy consumer this -// delivers the final lines (notably the summary) before the process exits; on a -// stalled consumer it returns after timeout so shutdown never hangs (#121). -// Safe to call more than once; writes after Close are silently dropped. -func (nb *nonBlockingWriter) Close(timeout time.Duration) { - nb.mu.Lock() - if nb.closed { - nb.mu.Unlock() - return - } - nb.closed = true - close(nb.ch) - nb.mu.Unlock() - - select { - case <-nb.done: - case <-time.After(timeout): - } -} diff --git a/cmd/ftsb_redisearch/nonblocking_writer_test.go b/cmd/ftsb_redisearch/nonblocking_writer_test.go index 587898d..c061db3 100644 --- a/cmd/ftsb_redisearch/nonblocking_writer_test.go +++ b/cmd/ftsb_redisearch/nonblocking_writer_test.go @@ -1,6 +1,8 @@ package main import ( + "bytes" + "sync" "sync/atomic" "testing" "time" @@ -11,11 +13,11 @@ type blockingWriter struct{ release chan struct{} } func (b *blockingWriter) Write(p []byte) (int, error) { <-b.release; return len(p), nil } -// The whole point of #121: Write must never block the caller even when the -// underlying consumer is wedged. -func TestNonBlockingWriterNeverBlocks(t *testing.T) { +// The whole point of #121: Write must never block the caller for longer than the +// write timeout, even when the underlying consumer is wedged. +func TestNonBlockingWriterDoesNotBlockOnStall(t *testing.T) { bw := &blockingWriter{release: make(chan struct{})} - nb := newNonBlockingWriter(bw, 4) + nb := newNonBlockingWriter(bw, 4, 50*time.Millisecond) done := make(chan struct{}) go func() { @@ -34,70 +36,55 @@ func TestNonBlockingWriterNeverBlocks(t *testing.T) { close(bw.release) // let the drain goroutine exit } -type countingWriter struct{ n int64 } +// countingWriter records how many lines were delivered and their bytes. +type countingWriter struct { + mu sync.Mutex + n int64 + buf bytes.Buffer +} -func (c *countingWriter) Write(p []byte) (int, error) { atomic.AddInt64(&c.n, 1); return len(p), nil } +func (c *countingWriter) Write(p []byte) (int, error) { + c.mu.Lock() + defer c.mu.Unlock() + atomic.AddInt64(&c.n, 1) + c.buf.Write(p) + return len(p), nil +} +func (c *countingWriter) contains(s string) bool { + c.mu.Lock() + defer c.mu.Unlock() + return bytes.Contains(c.buf.Bytes(), []byte(s)) +} -// When the consumer keeps up, no lines are dropped and order/content is intact. -func TestNonBlockingWriterDeliversWhenDrained(t *testing.T) { +// The regression this replaces (#122 CI failure): a message must be on the +// underlying writer by the time Write RETURNS on a healthy consumer, so a line +// logged immediately before os.Exit (log.Fatal, the final summary) is delivered +// even though os.Exit bypasses any deferred flush. Synchronous delivery makes +// this deterministic -- no polling. +func TestNonBlockingWriterDeliversBeforeReturnOnHealthy(t *testing.T) { cw := &countingWriter{} - nb := newNonBlockingWriter(cw, 1024) - const lines = 500 - for i := 0; i < lines; i++ { - _, _ = nb.Write([]byte("x")) + nb := newNonBlockingWriter(cw, 16, 2*time.Second) + + _, _ = nb.Write([]byte("Fatal error with X\n")) + + if got := atomic.LoadInt64(&cw.n); got != 1 { + t.Fatalf("delivered %d lines by the time Write returned, want 1", got) } - deadline := time.Now().Add(2 * time.Second) - for time.Now().Before(deadline) { - if atomic.LoadInt64(&cw.n) == lines { - return // all delivered - } - time.Sleep(5 * time.Millisecond) + if !cw.contains("Fatal error with X") { + t.Fatal("healthy-consumer line was not delivered before Write returned") } - t.Fatalf("delivered %d lines, want %d (nothing should drop when the consumer drains)", atomic.LoadInt64(&cw.n), lines) } -// Close must flush buffered lines to a healthy consumer before returning -- this -// is what guarantees the final summary line reaches stderr before the process -// exits (the #122 CI regression: "Issued ..." was queued but never flushed). -func TestNonBlockingWriterCloseFlushesBufferedLines(t *testing.T) { +// When the consumer keeps up, nothing is dropped and every line is delivered by +// the time the write loop finishes -- again deterministic, no polling. +func TestNonBlockingWriterDeliversAllWhenDrained(t *testing.T) { cw := &countingWriter{} - nb := newNonBlockingWriter(cw, 1024) - const lines = 200 + nb := newNonBlockingWriter(cw, 1024, 2*time.Second) + const lines = 500 for i := 0; i < lines; i++ { _, _ = nb.Write([]byte("x")) } - nb.Close(2 * time.Second) if got := atomic.LoadInt64(&cw.n); got != lines { - t.Fatalf("after Close delivered %d lines, want %d (Close must flush the tail)", got, lines) - } -} - -// Close must return within its timeout even if the consumer is wedged, so -// shutdown can never hang on a stalled output stream. -func TestNonBlockingWriterCloseReturnsDespiteStall(t *testing.T) { - bw := &blockingWriter{release: make(chan struct{})} - nb := newNonBlockingWriter(bw, 4) - for i := 0; i < 100; i++ { // fill + overflow the buffer - _, _ = nb.Write([]byte("x")) - } - done := make(chan struct{}) - go func() { nb.Close(200 * time.Millisecond); close(done) }() - select { - case <-done: - case <-time.After(3 * time.Second): - close(bw.release) - t.Fatal("Close blocked while the underlying writer was stalled") - } - close(bw.release) // let the drain goroutine exit -} - -// Writes after Close are dropped, not panicking on a send to a closed channel. -func TestNonBlockingWriterWriteAfterCloseIsSafe(t *testing.T) { - cw := &countingWriter{} - nb := newNonBlockingWriter(cw, 16) - nb.Close(time.Second) - if _, err := nb.Write([]byte("late")); err != nil { - t.Fatalf("Write after Close returned error: %v", err) + t.Fatalf("delivered %d lines, want %d (nothing should drop when the consumer keeps up)", got, lines) } - nb.Close(time.Second) // idempotent }