From 167aec65d88e37d7ef6d51f2654913b1edc55c86 Mon Sep 17 00:00:00 2001
From: "detail-app[bot]" <180357370+detail-app[bot]@users.noreply.github.com>
Date: Wed, 16 Sep 2026 02:04:08 +0000
Subject: [PATCH] fix(tail): flush stdin write buffer on SIGINT/SIGTERM

---
 bench/logging/tests/test_stdin.py | 132 ++++++++++++++++++-
 src/core/lifecycle.zig            |   9 ++
 src/tail/framer.zig               |  60 +++++++++
 src/tail/runtime.zig              | 210 +++++++++++++++++++++++++++++-
 4 files changed, 409 insertions(+), 2 deletions(-)

diff --git a/bench/logging/tests/test_stdin.py b/bench/logging/tests/test_stdin.py
index 87f07cd3..acd4d848 100644
--- a/bench/logging/tests/test_stdin.py
+++ b/bench/logging/tests/test_stdin.py
@@ -1,7 +1,10 @@
+import os
+import signal
 import subprocess
+import time
 from pathlib import Path
 
-from .helpers import read_lines
+from .helpers import read_lines, wait_for_process_exit
 
 
 def test_stdin_pipe(edge_tail_bin: Path, tmp_path: Path) -> None:
@@ -13,3 +16,130 @@ def test_stdin_pipe(edge_tail_bin: Path, tmp_path: Path) -> None:
         check=True,
     )
     assert read_lines(out) == ["a", "b"]
+
+
+def _stdin_input() -> bytes:
+    # Larger than the default write_buf (64 KiB) so the writer drains at least
+    # once mid-stream and then holds a residual, which is exactly the buffer
+    # the signal must flush. Ends with a newline so no partial line is stranded
+    # in the framer scratch.
+    lines = []
+    total = 0
+    i = 0
+    while total < 130_000:
+        line = f"line-{i:06d}-padding-content-for-buffer-test\n"
+        lines.append(line)
+        total += len(line)
+        i += 1
+    return "".join(lines).encode()
+
+
+def _drain_then_signal(edge_tail_bin: Path, tmp_path: Path, sig: int) -> None:
+    data = _stdin_input()
+    out = tmp_path / "stdin_signal.out"
+    out_f = open(out, "wb")
+    proc = subprocess.Popen(
+        [str(edge_tail_bin), "-"],
+        stdin=subprocess.PIPE,
+        stdout=out_f,
+        stderr=subprocess.DEVNULL,
+    )
+    try:
+        # Write all input but keep the write end of the pipe open: edge-tail
+        # reads it, drains the writer on overflow, then blocks on the next
+        # readv waiting for more data.
+        assert proc.stdin is not None
+        proc.stdin.write(data)
+        proc.stdin.flush()
+
+        # Wait until the pump has read all the input (output grows then stops)
+        # — at that point it is blocked on the empty pipe, with the residual in
+        # the write buffer. Bounded by the input size; this is fast.
+        deadline = time.monotonic() + 5.0
+        last = -1
+        stable_since = time.monotonic()
+        while time.monotonic() < deadline:
+            assert proc.poll() is None, "edge-tail exited early mid-stream"
+            time.sleep(0.05)
+            sz = out.stat().st_size
+            if sz != last:
+                last = sz
+                stable_since = time.monotonic()
+            elif sz > 0 and time.monotonic() - stable_since >= 0.2:
+                break
+        pre = out.stat().st_size
+        assert pre > 0, "no data drained before signal"
+        assert proc.poll() is None, "edge-tail exited before signal"
+
+        proc.send_signal(sig)
+        rc = wait_for_process_exit(proc, timeout_s=4.0)
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+            proc.wait()
+        out_f.close()
+
+    # Clean exit (cooperative shutdown), not hard-killed by the signal, and the
+    # full input reached the output: the writer drained mid-stream and the
+    # residual write buffer was flushed on the main thread after the cancel.
+    assert rc == 0
+    assert out.stat().st_size == len(data)
+    assert out.read_bytes() == data
+
+
+def test_stdin_sigint_flushes_residual(edge_tail_bin: Path, tmp_path: Path) -> None:
+    _drain_then_signal(edge_tail_bin, tmp_path, signal.SIGINT)
+
+
+def test_stdin_sigterm_flushes_residual(edge_tail_bin: Path, tmp_path: Path) -> None:
+    _drain_then_signal(edge_tail_bin, tmp_path, signal.SIGTERM)
+
+
+def test_stdin_fifo_sigint_flushes_residual(edge_tail_bin: Path, tmp_path: Path) -> None:
+    # The bug report's scenario: a named pipe with the write end held open
+    # (e.g. `tail -f app.log | edge-tail - > out.log`), stopped with Ctrl-C.
+    data = _stdin_input()
+    datafile = tmp_path / "data.in"
+    datafile.write_bytes(data)
+
+    fifo = tmp_path / "et.fifo"
+    os.mkfifo(fifo)
+    out = tmp_path / "fifo_signal.out"
+
+    # Feeder writes the file into the fifo, then holds the write end open
+    # (sleep 30) so stdin never reaches EOF — the exact foreground-pipeline
+    # Ctrl-C case.
+    feeder = subprocess.Popen(["sh", "-c", f"exec 3>{fifo}; cat {datafile} >&3; sleep 30"])
+    try:
+        out_f = open(fifo, "rb")
+        try:
+            proc = subprocess.Popen(
+                [str(edge_tail_bin), "-"],
+                stdin=out_f,
+                stdout=open(out, "wb"),
+                stderr=subprocess.DEVNULL,
+            )
+            deadline = time.monotonic() + 5.0
+            last = -1
+            stable_since = time.monotonic()
+            while time.monotonic() < deadline:
+                assert proc.poll() is None, "edge-tail exited early mid-stream"
+                time.sleep(0.05)
+                sz = out.stat().st_size
+                if sz != last:
+                    last = sz
+                    stable_since = time.monotonic()
+                elif sz > 0 and time.monotonic() - stable_since >= 0.2:
+                    break
+            assert proc.poll() is None
+            proc.send_signal(signal.SIGINT)
+            rc = wait_for_process_exit(proc, timeout_s=4.0)
+        finally:
+            out_f.close()
+        assert rc == 0
+        assert out.stat().st_size == len(data)
+        assert out.read_bytes() == data
+    finally:
+        if feeder.poll() is None:
+            feeder.kill()
+            feeder.wait()
diff --git a/src/core/lifecycle.zig b/src/core/lifecycle.zig
index f330a75d..851eae2c 100644
--- a/src/core/lifecycle.zig
+++ b/src/core/lifecycle.zig
@@ -45,6 +45,15 @@ pub const Lifecycle = struct {
         self.shutdown_event.set(io);
     }
 
+    /// Like `requestShutdown` but without the diagnostic log. Used by
+    /// short-lived paths (e.g. edge-tail's stdin mode) that reach normal
+    /// completion and want to keep stderr clean; the signal path still uses
+    /// the logging `requestShutdown`. Idempotent and safe from any thread.
+    pub fn requestShutdownQuiet(self: *Lifecycle, io: std.Io) void {
+        if (self.shutdown_requested.swap(true, .acq_rel)) return;
+        self.shutdown_event.set(io);
+    }
+
     pub fn isShuttingDown(self: *const Lifecycle) bool {
         return self.shutdown_requested.load(.acquire);
     }
diff --git a/src/tail/framer.zig b/src/tail/framer.zig
index 3c1a673a..586283dd 100644
--- a/src/tail/framer.zig
+++ b/src/tail/framer.zig
@@ -102,6 +102,36 @@ pub const LineFramer = struct {
         try self.finish(writer, filter_ctx, filter_fn);
     }
 
+    /// Pumps from a streaming `File` (e.g. stdin) using short `readStreaming`
+    /// reads directly into `read_buf` — one `readv` per iteration with no
+    /// reader-side internal-buffer prefetch and no `File.Reader` `ReadFailed`
+    /// conversion. This matters for cooperative signal shutdown: a cancel
+    /// interrupts the blocking `readv` and surfaces here as `error.Canceled`
+    /// (not converted to `ReadFailed` by `File.Reader.readVecStreaming`), and
+    /// because each iteration reads only what is immediately available, no
+    /// prefetched bytes are stranded when the pump unwinds. The only residual
+    /// at cancel time is the write buffer, which the caller flushes on the
+    /// (uncanceled) main thread — eliminating the lost-residual-on-signal bug.
+    pub fn pumpFileStreaming(
+        self: *LineFramer,
+        io: std.Io,
+        file: std.Io.File,
+        writer: *std.Io.Writer,
+        filter_ctx: *anyopaque,
+        filter_fn: *const LineFilterFn,
+    ) !void {
+        while (true) {
+            const n = file.readStreaming(io, &.{self.read_buf}) catch |err| switch (err) {
+                error.EndOfStream => break,
+                else => return err, // includes error.Canceled on cooperative shutdown
+            };
+            if (n == 0) break;
+            try self.ingestChunk(self.read_buf[0..n], writer, filter_ctx, filter_fn);
+        }
+
+        try self.finish(writer, filter_ctx, filter_fn);
+    }
+
     /// Reads `[start_offset, end_offset)` from `file` using positional reads
     /// and frames newline-delimited lines to the writer.
     pub fn readRange(
@@ -215,6 +245,36 @@ test "framer public API: readRange emits file bytes as lines" {
     try testing.expectEqualStrings("x\ny\n", out.written());
 }
 
+test "framer public API: pumpFileStreaming frames bytes across short reads to EOF" {
+    var tmp = testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    const io = std.Options.debug_io;
+    {
+        const f = try tmp.dir.createFile(io, "in.log", .{});
+        defer f.close(io);
+        try f.writeStreamingAll(io, "a\nbc\ndef\n");
+    }
+
+    const in_path = try tmp.dir.realPathFileAlloc(io, "in.log", testing.allocator);
+    defer testing.allocator.free(in_path);
+
+    const file = try std.Io.Dir.cwd().openFile(io, in_path, .{ .mode = .read_only });
+    defer file.close(io);
+
+    // Small read_buf forces several short readv iterations across line
+    // boundaries, exercising the same chunk-spanning path the stdin pump uses.
+    var framer = try LineFramer.init(testing.allocator, 4, 1024);
+    defer framer.deinit();
+    var out: std.Io.Writer.Allocating = .init(testing.allocator);
+    defer out.deinit();
+
+    var ctx: u8 = 0;
+    try framer.pumpFileStreaming(io, file, &out.writer, &ctx, keepAll);
+
+    try testing.expectEqualStrings("a\nbc\ndef\n", out.written());
+}
+
 test "framer public API: trailing line without newline is preserved byte-exactly" {
     var framer = try LineFramer.init(testing.allocator, 8, 1024);
     defer framer.deinit();
diff --git a/src/tail/runtime.zig b/src/tail/runtime.zig
index f8770541..1d5c34bd 100644
--- a/src/tail/runtime.zig
+++ b/src/tail/runtime.zig
@@ -336,6 +336,74 @@ const PollLoop = struct {
     }
 };
 
+/// Stdin pump loop as a lifecycle task: the blocking `framer.pumpFileStreaming`
+/// runs on a worker thread so the structured-shutdown signal waiter can cancel
+/// it cooperatively (PLAN.md §9 Phase 6 — analogous to `PollLoop` for file
+/// tailing). `pumpFileStreaming` reads stdin directly via `file.readStreaming`
+/// (one short `readv` per iteration, no reader-side internal-buffer prefetch),
+/// so a cancel interrupts the blocked `readv` and surfaces as `error.Canceled`
+/// (not converted to `ReadFailed` by `File.Reader.readVecStreaming`) and no
+/// prefetched bytes are stranded when the pump unwinds. The main thread then
+/// flushes the residual write buffer after the task joins, so no in-flight
+/// bytes are lost on `SIGINT`/`SIGTERM` — only the unframed tail of a partial
+/// line can be dropped, matching the file loop. A normal stdin EOF completes
+/// the pump and requests shutdown so the main thread's `awaitShutdown` returns
+/// and flushes the residual — the post-EOF flush the old `runStream` did on
+/// the pump thread, now done on the uncanceled main thread.
+///
+/// Two correctness constraints shape `runStdinToOutput`:
+///
+/// 1. The pump must run on a worker (not the main) thread. The Threaded Io
+///    backend only cancels syscalls on registered worker threads
+///    (`Thread.current` is set solely in `Threaded.worker`), so a blocking
+///    `readv` on the main thread cannot be interrupted by `Group.cancel`.
+///    `lifecycle.spawn` puts the read on a worker where
+///    `signalCanceledSyscall` (`tgkill`/`pthread_kill`) can reach it.
+///
+/// 2. `installSignalWaiter` must run BEFORE the worker is spawned. It blocks
+///    `SIGINT`/`SIGTERM`/`SIGUSR1` in the calling thread; threads created
+///    afterwards inherit that mask. The sigwait thread and the pump worker
+///    therefore both have the signals blocked, so a process-directed
+///    SIGINT/SIGTERM stays pending and is consumed by `sigwait` (requesting
+///    shutdown) instead of landing on the worker with default disposition and
+///    hard-killing the process. Spawning the worker first — as the file loop
+///    does — leaves the worker's mask unblocked and breaks signal handling.
+const StdinLoop = struct {
+    runtime: *Runtime,
+    input: *io_mod.Input,
+    output: *io_mod.Output,
+    framer: *framer_mod.LineFramer,
+    evaluator: *eval_stream.StreamEvaluator,
+    lifecycle: *lifecycle_mod.Lifecycle,
+    failure: ?anyerror = null,
+
+    fn run(self: *StdinLoop) std.Io.Cancelable!void {
+        self.framer.pumpFileStreaming(
+            self.runtime.io,
+            self.input.file,
+            self.output.writer(),
+            self.evaluator,
+            Runtime.evalLineFilter,
+        ) catch |err| switch (err) {
+            error.Canceled => return error.Canceled,
+            else => {
+                // During structured shutdown a cancel can surface as a
+                // non-Canceled error if a writer drain was interrupted
+                // mid-writev; reconcile any error seen while shutting down
+                // with the cancel that caused it.
+                if (self.lifecycle.isShuttingDown()) return error.Canceled;
+                self.failure = err;
+                self.lifecycle.requestShutdown(self.runtime.io);
+                return;
+            },
+        };
+        // Normal stdin EOF: wake the main thread's `awaitShutdown`. Use the
+        // quiet variant so a plain `echo ... | edge-tail -` keeps stderr clean;
+        // the signal path logs via `requestShutdown` from the waiter thread.
+        self.lifecycle.requestShutdownQuiet(self.runtime.io);
+    }
+};
+
 pub fn runStdinToOutput(
     allocator: std.mem.Allocator,
     io: std.Io,
@@ -355,7 +423,72 @@ pub fn runStdinToOutput(
     var output = try io_mod.Output.init(allocator, io, out_target, cfg.write_buf);
     defer output.deinit();
 
-    try runtime.runStream(&input, &output);
+    var stdio_bus = initEventBus(io, environ_map);
+    var evaluator = try eval_stream.StreamEvaluator.init(
+        allocator,
+        cfg.input_format,
+        cfg.policy_path,
+        stdio_bus.eventBus(),
+    );
+    defer evaluator.deinit();
+    var framer = try framer_mod.LineFramer.init(allocator, cfg.read_buf, cfg.max_line);
+    defer framer.deinit();
+
+    // Run the pump as a lifecycle task so the blocking read can be canceled
+    // cooperatively on signal (structured shutdown, PLAN.md §9 Phase 6). The
+    // main thread awaits shutdown, then flushes the residual write buffer on
+    // this (uncanceled) thread.
+    var lifecycle: lifecycle_mod.Lifecycle = .init;
+    var loop: StdinLoop = .{
+        .runtime = &runtime,
+        .input = &input,
+        .output = &output,
+        .framer = &framer,
+        .evaluator = &evaluator,
+        .lifecycle = &lifecycle,
+    };
+
+    // Block INT/TERM/USR1 and start the sigwait thread BEFORE spawning the
+    // pump worker: the worker inherits the blocked mask, so a
+    // process-directed SIGINT/SIGTERM is consumed by `sigwait` (→
+    // requestShutdown → cooperative cancel) instead of hitting the worker
+    // with default disposition and hard-killing the process.
+    var signal_count = std.atomic.Value(u32).init(0);
+    var shutdown_waiter = std.atomic.Value(bool).init(false);
+    var signal_waiter: ?SignalWaiterHandle = null;
+    if (installSignalWaiter(io, &lifecycle, &signal_count, &shutdown_waiter)) |waiter| {
+        signal_waiter = waiter;
+    } else |err| switch (err) {
+        error.UnsupportedPlatform => {},
+        else => return err,
+    }
+
+    lifecycle.spawn(io, StdinLoop.run, .{&loop}) catch |err| {
+        if (signal_waiter) |waiter| teardownSignalWaiter(waiter, &shutdown_waiter);
+        return err;
+    };
+
+    lifecycle.awaitShutdown(io) catch |err| switch (err) {
+        error.Canceled => {},
+    };
+    // Cancel (and join) the pump task directly rather than via
+    // `lifecycle.shutdown`, which logs "all tasks drained" on every run —
+    // including a clean stdin EOF, where it would pollute stderr. On EOF the
+    // task is already done (a no-op join); on signal this interrupts the
+    // blocked readv so the pump unwinds with `error.Canceled`. The assert
+    // mirrors the one `shutdown` would have performed.
+    std.debug.assert(lifecycle.isShuttingDown());
+    lifecycle.group.cancel(io);
+
+    if (signal_waiter) |waiter| teardownSignalWaiter(waiter, &shutdown_waiter);
+
+    // The canceled task can't reliably do final IO; drain and flush the
+    // residual write buffer on this (uncanceled) thread. This is the fix for
+    // the lost-residual-on-signal bug: without it, up to `write_buf` bytes
+    // (64 KiB default) were lost on SIGINT/SIGTERM because the process was
+    // hard-killed before the only post-EOF `output.flush()` could run.
+    try output.flush();
+    if (loop.failure) |err| return err;
 }
 
 pub fn runFilesToOutput(
@@ -486,3 +619,78 @@ test "runtime public API: runStream applies policy drops" {
     defer testing.allocator.free(got);
     try testing.expectEqualStrings("ok\nnext\n", got);
 }
+
+test "runtime stdin path: structured StdinLoop task flushes residual on EOF" {
+    // Exercises the structured-shutdown coordination `runStdinToOutput` uses
+    // (spawn pump task → awaitShutdown → cancel/join → flush on the main
+    // thread) without installing the signal waiter, so it doesn't touch the
+    // test runner's process signal mask. The pump task reaches EOF, requests
+    // shutdown, and the main thread flushes the residual write buffer.
+    const io = testing.io;
+    var tmp = testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const in_path = "in.log";
+    const out_path = "out.log";
+    {
+        const f = try tmp.dir.createFile(io, in_path, .{});
+        defer f.close(io);
+        try f.writeStreamingAll(io, "a\nb\n");
+    }
+
+    const abs_in = try tmp.dir.realPathFileAlloc(io, in_path, testing.allocator);
+    defer testing.allocator.free(abs_in);
+    const cwd_abs = try tmp.dir.realPathFileAlloc(io, ".", testing.allocator);
+    defer testing.allocator.free(cwd_abs);
+    const abs_out = try std.fs.path.join(testing.allocator, &.{ cwd_abs, out_path });
+    defer testing.allocator.free(abs_out);
+
+    const cfg: types.TailConfig = .{
+        .output_path = abs_out,
+        .read_buf = 16,
+        .max_line = 1024,
+        .write_buf = 16,
+    };
+    var env_map = std.process.Environ.Map.init(testing.allocator);
+    defer env_map.deinit();
+    var runtime = try Runtime.init(testing.allocator, io, &env_map, cfg);
+    var input = try io_mod.Input.init(testing.allocator, io, .{ .file = abs_in }, cfg.read_buf);
+    defer input.deinit();
+    var output = try io_mod.Output.init(testing.allocator, io, .{ .file_append = abs_out }, cfg.write_buf);
+    defer output.deinit();
+
+    var stdio_bus = initEventBus(io, &env_map);
+    var evaluator = try eval_stream.StreamEvaluator.init(
+        testing.allocator,
+        cfg.input_format,
+        cfg.policy_path,
+        stdio_bus.eventBus(),
+    );
+    defer evaluator.deinit();
+    var framer = try framer_mod.LineFramer.init(testing.allocator, cfg.read_buf, cfg.max_line);
+    defer framer.deinit();
+
+    var lifecycle: lifecycle_mod.Lifecycle = .init;
+    var loop: StdinLoop = .{
+        .runtime = &runtime,
+        .input = &input,
+        .output = &output,
+        .framer = &framer,
+        .evaluator = &evaluator,
+        .lifecycle = &lifecycle,
+    };
+    try lifecycle.spawn(io, StdinLoop.run, .{&loop});
+
+    lifecycle.awaitShutdown(io) catch |err| switch (err) {
+        error.Canceled => {},
+    };
+    // Mirror `runStdinToOutput`: cancel/join the pump task without the
+    // "all tasks drained" log so a clean EOF keeps stderr quiet.
+    lifecycle.group.cancel(io);
+
+    try output.flush();
+    if (loop.failure) |err| return err;
+
+    const got = try tmp.dir.readFileAlloc(io, out_path, testing.allocator, .limited(1024));
+    defer testing.allocator.free(got);
+    try testing.expectEqualStrings("a\nb\n", got);
+}