Don't hang (and lose the result) when the output consumer stalls

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)
+}

cmd/ftsb_redisearch/main.go

@@ -106,21 +106,26 @@ 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.
+	// 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 {
 			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)

cmd/ftsb_redisearch/nonblocking_writer.go

@@ -0,0 +1,85 @@
+package main
+
+import (
+	"io"
+	"time"
+)
+
+// 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). 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 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 writeReq
+	writeTimeout time.Duration
+}
+
+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 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 writeReq, bufLines),
+		writeTimeout: writeTimeout,
+	}
+	go func() {
+		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(req.b)
+			close(req.done)
+		}
+	}()
+	return nb
+}
+
+// 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)
+	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 <-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.
+	}
+	return len(p), nil
+}

cmd/ftsb_redisearch/nonblocking_writer_test.go

@@ -0,0 +1,90 @@
+package main
+
+import (
+	"bytes"
+	"sync"
+	"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 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, 50*time.Millisecond)
+
+	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
+}
+
+// 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) {
+	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))
+}
+
+// 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, 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)
+	}
+	if !cw.contains("Fatal error with X") {
+		t.Fatal("healthy-consumer line was not delivered before Write returned")
+	}
+}
+
+// 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, 2*time.Second)
+	const lines = 500
+	for i := 0; i < lines; i++ {
+		_, _ = nb.Write([]byte("x"))
+	}
+	if got := atomic.LoadInt64(&cw.n); got != lines {
+		t.Fatalf("delivered %d lines, want %d (nothing should drop when the consumer keeps up)", got, lines)
+	}
+}