Fix unbounded stderr_buffer leak; make Daemon stderr reliable

.github/workflows/ci.yml

@@ -20,7 +20,7 @@ jobs:
     name: Compile & Warnings
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v7
 
       - uses: erlef/setup-beam@v1
         with:
@@ -47,7 +47,7 @@ jobs:
     name: Formatting
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v7
 
       - uses: erlef/setup-beam@v1
         with:
@@ -69,7 +69,7 @@ jobs:
     name: Credo
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v7
 
       - uses: erlef/setup-beam@v1
         with:
@@ -106,10 +106,10 @@ jobs:
             elixir: "1.20.0-rc.1"
             otp: "28.3.3"
           - os: macos-latest
-            elixir: "1.17.3"
-            otp: "27.2"
+            elixir: "1.18.4"
+            otp: "28.3.3"
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v7
 
       - uses: erlef/setup-beam@v1
         with:
@@ -135,7 +135,7 @@ jobs:
     name: Dialyzer
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v7
 
       - uses: erlef/setup-beam@v1
         with:
@@ -169,7 +169,7 @@ jobs:
     name: Sanitizers (ASan + UBSan)
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v7
 
       - uses: erlef/setup-beam@v1
         with:
@@ -203,7 +203,7 @@ jobs:
     needs: [compile, format, credo, test, dialyzer, sanitizers]
     if: startsWith(github.ref, 'refs/tags/v')
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v7
 
       - uses: erlef/setup-beam@v1
         with:

lib/net_runner.ex

@@ -34,7 +34,12 @@ defmodule NetRunner do
   ## Options
 
     * `:stderr` - `:consume` (default, drained internally so the child never
-      blocks on a full stderr pipe) or `:disabled`
+      blocks on a full stderr pipe) or `:disabled`. In `:consume` mode only the
+      most-recent `:stderr_tail_bytes` of stderr are retained (see
+      `NetRunner.Process.stderr_tail/1`); the rest is drained and dropped.
+    * `:stderr_tail_bytes` - cap (bytes) on the retained stderr tail. Default
+      `8192`. `0` retains nothing. The tail is raw bytes and may begin
+      mid-character, so treat it as diagnostic text, not valid UTF-8.
     * `:input` - data to write to stdin (binary or enumerable)
     * `:timeout` - maximum wall-clock time in milliseconds. Sends SIGTERM then SIGKILL
       on timeout. Returns `{:error, :timeout}` instead of `{output, exit_status}`.

lib/net_runner/daemon.ex

@@ -3,7 +3,10 @@ defmodule NetRunner.Daemon do
   A supervised long-running OS process.
 
   Wraps `NetRunner.Process` for integration into a supervision tree.
-  Automatically drains stdout/stderr to prevent pipe blocking.
+  Automatically drains stdout/stderr to prevent pipe blocking. Both streams
+  are delivered to the `:on_output` callback; stderr is owned by the Daemon's
+  own drain task (the underlying process runs with `stderr: :disabled`), so
+  `:on_output` sees the complete stderr stream in order.
 
   ## Usage
 
@@ -42,7 +45,15 @@ defmodule NetRunner.Daemon do
     cmd = Keyword.fetch!(opts, :cmd)
     args = Keyword.get(opts, :args, [])
     on_output = Keyword.get(opts, :on_output, :discard)
-    process_opts = Keyword.get(opts, :process_opts, [])
+
+    # Force stderr: :disabled so the underlying Process does NOT start its own
+    # internal stderr consumer. The Daemon's own drain task (below) is then the
+    # sole reader of the stderr pipe, so on_output receives the full stream in
+    # order rather than racing the internal consumer for chunks.
+    process_opts =
+      opts
+      |> Keyword.get(:process_opts, [])
+      |> Keyword.put(:stderr, :disabled)
 
     case Proc.start_link(cmd, args, process_opts) do
       {:ok, proc} ->

lib/net_runner/process.ex

@@ -83,6 +83,23 @@ defmodule NetRunner.Process do
     GenServer.call(process, :stats)
   end
 
+  @doc """
+  Returns the retained tail of consumed stderr.
+
+  In the default `:consume` stderr mode, stderr is drained to keep the child
+  from blocking on a full pipe; only the most-recent `:stderr_tail_bytes`
+  bytes (default 8 KB) are retained and returned here. Useful for diagnosing
+  why a command failed.
+
+  The tail is raw bytes and may begin mid-character if stderr was truncated,
+  so treat it as diagnostic text rather than guaranteed-valid UTF-8. Returns
+  `""` in `:disabled` mode.
+  """
+  @spec stderr_tail(GenServer.server()) :: binary()
+  def stderr_tail(process) do
+    GenServer.call(process, :stderr_tail)
+  end
+
   @doc "Set PTY window size (rows, cols). Only works in PTY mode."
   def set_window_size(process, rows, cols) do
     GenServer.call(process, {:set_window_size, rows, cols})
@@ -204,6 +221,10 @@ defmodule NetRunner.Process do
     {:reply, state.stats, state}
   end
 
+  def handle_call(:stderr_tail, _from, state) do
+    {:reply, state.stderr_tail, state}
+  end
+
   def handle_call({:set_window_size, rows, cols}, _from, state) do
     send_shepherd_command(state, <<0x03, rows::big-16, cols::big-16>>)
     {:reply, :ok, state}
@@ -276,7 +297,7 @@ defmodule NetRunner.Process do
   # the data would be silently dropped by the catch-all below.
   def handle_info({:stderr_data, data}, state) when is_binary(data) do
     stats = Stats.record_read_stderr(state.stats, byte_size(data))
-    state = %{state | stderr_buffer: [data | state.stderr_buffer], stats: stats}
+    state = %{state | stderr_tail: append_stderr_tail(state, data), stats: stats}
     # Drain anything else buffered and re-arm enif_select on EAGAIN.
     {:noreply, consume_stderr(state)}
   end
@@ -478,11 +499,28 @@ defmodule NetRunner.Process do
     end
   end
 
+  # Appends `data` to the retained stderr tail, keeping only the most-recent
+  # `stderr_tail_bytes` bytes. A cap of 0 retains nothing (drain-and-drop);
+  # the pipe is still drained so the child never blocks. All bytes are still
+  # counted in stats — only retention is bounded.
+  defp append_stderr_tail(%{stderr_tail: tail, stderr_tail_bytes: cap}, data) do
+    combined = tail <> data
+    size = byte_size(combined)
+
+    if size > cap do
+      # Keep the last `cap` bytes. For cap == 0 this is
+      # binary_part(combined, size, 0), which is valid and returns <<>>.
+      binary_part(combined, size - cap, cap)
+    else
+      combined
+    end
+  end
+
   defp consume_stderr(state) do
     case Pipe.read(state.stderr) do
       {:ok, data} ->
         stats = Stats.record_read_stderr(state.stats, byte_size(data))
-        consume_stderr(%{state | stderr_buffer: [data | state.stderr_buffer], stats: stats})
+        consume_stderr(%{state | stderr_tail: append_stderr_tail(state, data), stats: stats})
 
       :eof ->
         state

lib/net_runner/process/exec.ex

@@ -28,6 +28,7 @@ defmodule NetRunner.Process.Exec do
     result =
       with :ok <- validate_cmd_and_args(cmd, args),
            :ok <- validate_stderr_mode(Keyword.get(opts, :stderr, :consume), pty_mode),
+           :ok <- validate_stderr_tail_bytes(Keyword.get(opts, :stderr_tail_bytes, 8_192)),
            :ok <- validate_cgroup_path(Keyword.get(opts, :cgroup_path, nil)),
            {:ok, listen_socket} <- create_uds_listener(uds_path),
            shepherd_port <- open_shepherd(uds_path, cmd, args, opts),
@@ -98,6 +99,7 @@ defmodule NetRunner.Process.Exec do
          cmd: cmd,
          args: args,
          stderr_mode: stderr_mode,
+         stderr_tail_bytes: Keyword.get(opts, :stderr_tail_bytes, 8_192),
          status: :running
        }}
     else
@@ -130,6 +132,15 @@ defmodule NetRunner.Process.Exec do
     {:error, {:invalid_stderr, "must be :consume or :disabled, got: #{inspect(mode)}"}}
   end
 
+  # The bounded stderr tail cap. 0 disables retention (drain-and-drop) while
+  # still draining the pipe so the child never blocks.
+  defp validate_stderr_tail_bytes(bytes) when is_integer(bytes) and bytes >= 0, do: :ok
+
+  defp validate_stderr_tail_bytes(bytes) do
+    {:error,
+     {:invalid_stderr_tail_bytes, "must be a non-negative integer, got: #{inspect(bytes)}"}}
+  end
+
   defp validate_cgroup_path(nil), do: :ok
 
   defp validate_cgroup_path(path) do

lib/net_runner/process/state.ex

@@ -17,7 +17,11 @@ defmodule NetRunner.Process.State do
     operations: %Operations{},
     awaiting_exit: [],
     stderr_mode: :consume,
-    stderr_buffer: [],
+    # Bounded tail of consumed stderr: only the most-recent
+    # `stderr_tail_bytes` bytes are retained (the rest is drained and
+    # dropped). Stats still count every byte. Held as a single binary.
+    stderr_tail: <<>>,
+    stderr_tail_bytes: 8_192,
     status: :starting,
     stats: %Stats{}
   ]
@@ -37,7 +41,8 @@ defmodule NetRunner.Process.State do
           operations: Operations.t(),
           awaiting_exit: [GenServer.from()],
           stderr_mode: :consume | :disabled,
-          stderr_buffer: [binary()],
+          stderr_tail: binary(),
+          stderr_tail_bytes: non_neg_integer(),
           status: status()
         }
 end

test/stderr_tail_test.exs

@@ -0,0 +1,158 @@
+defmodule NetRunner.StderrTailTest do
+  use ExUnit.Case, async: true
+
+  alias NetRunner.Daemon
+  alias NetRunner.Process, as: Proc
+
+  # "errline\n" is 8 bytes; 100_000 / 8 = 12_500 whole lines, and the 8 KB
+  # default cap is 1_024 whole lines — so the tail aligns cleanly and we can
+  # assert exact contents.
+  @line "errline\n"
+  @total_bytes 100_000
+
+  describe "bounded stderr tail (:consume mode)" do
+    test "retains only the most-recent stderr_tail_bytes (default 8 KB)" do
+      pid = start_proc("sh", ["-c", "yes errline 2>/dev/null | head -c #{@total_bytes} 1>&2"])
+
+      assert {:ok, _status} = Proc.await_exit(pid)
+      wait_until_drained(pid, @total_bytes)
+
+      tail = Proc.stderr_tail(pid)
+      assert byte_size(tail) == 8_192
+      # The tail is exactly the last 1_024 lines.
+      assert tail == String.duplicate(@line, 1_024)
+
+      # Stats still count every byte that was drained, not just the retained tail.
+      assert Proc.stats(pid).bytes_err == @total_bytes
+    end
+
+    test "honors a custom :stderr_tail_bytes cap" do
+      pid =
+        start_proc("sh", ["-c", "yes errline 2>/dev/null | head -c #{@total_bytes} 1>&2"],
+          stderr_tail_bytes: 1_024
+        )
+
+      assert {:ok, _status} = Proc.await_exit(pid)
+      wait_until_drained(pid, @total_bytes)
+
+      assert byte_size(Proc.stderr_tail(pid)) == 1_024
+      assert Proc.stats(pid).bytes_err == @total_bytes
+    end
+
+    test ":stderr_tail_bytes of 0 retains nothing but still drains the pipe" do
+      pid =
+        start_proc("sh", ["-c", "yes errline 2>/dev/null | head -c #{@total_bytes} 1>&2"],
+          stderr_tail_bytes: 0
+        )
+
+      assert {:ok, _status} = Proc.await_exit(pid)
+      wait_until_drained(pid, @total_bytes)
+
+      assert Proc.stderr_tail(pid) == ""
+      # Draining still happened (the child did not block on a full pipe).
+      assert Proc.stats(pid).bytes_err == @total_bytes
+    end
+
+    test "retention is bounded while throughput is not (no leak)" do
+      pid = start_proc("sh", ["-c", "yes errline 2>/dev/null | head -c 200000 1>&2"])
+
+      assert {:ok, _status} = Proc.await_exit(pid)
+      wait_until_drained(pid, 200_000)
+
+      # The whole point: O(1) retention vs O(n) throughput.
+      assert Proc.stats(pid).bytes_err == 200_000
+      assert byte_size(Proc.stderr_tail(pid)) == 8_192
+    end
+  end
+
+  describe "validation" do
+    test "rejects a negative :stderr_tail_bytes at start" do
+      assert {:error, {:invalid_stderr_tail_bytes, _}} =
+               Proc.start("echo", ["hi"], stderr_tail_bytes: -1)
+    end
+
+    test "rejects a non-integer :stderr_tail_bytes at start" do
+      assert {:error, {:invalid_stderr_tail_bytes, _}} =
+               Proc.start("echo", ["hi"], stderr_tail_bytes: :lots)
+    end
+  end
+
+  describe ":disabled mode" do
+    test "stderr_tail is empty and explicit read_stderr still works" do
+      pid = start_proc("sh", ["-c", "echo err 1>&2"], stderr: :disabled)
+
+      assert {:ok, "err\n"} = Proc.read_stderr(pid)
+      assert {:ok, _status} = Proc.await_exit(pid)
+      assert Proc.stderr_tail(pid) == ""
+    end
+  end
+
+  describe "Daemon stderr ownership" do
+    test "on_output receives the full stderr stream, in order" do
+      test_pid = self()
+      handler = fn data -> send(test_pid, {:out, data}) end
+
+      {:ok, daemon} =
+        Daemon.start_link(
+          cmd: "sh",
+          args: ["-c", "echo a 1>&2; echo b 1>&2; echo c 1>&2"],
+          on_output: handler
+        )
+
+      # The Daemon's drain task is the sole stderr reader, so nothing is lost
+      # to an internal consumer racing it.
+      assert collect_output("a\nb\nc\n") == "a\nb\nc\n"
+
+      GenServer.stop(daemon)
+    end
+  end
+
+  # Starts a process and registers cleanup so the GenServer (and its NIF FDs)
+  # don't linger past the test — Proc.start/3 is unlinked, so without this the
+  # async suite would leak a GenServer per test.
+  defp start_proc(cmd, args, opts \\ []) do
+    {:ok, pid} = Proc.start(cmd, args, opts)
+
+    on_exit(fn ->
+      # :ok if alive; exits with :noproc if already gone — both fine.
+      try do
+        GenServer.stop(pid, :normal, 1_000)
+      catch
+        :exit, _ -> :ok
+      end
+    end)
+
+    pid
+  end
+
+  # Polls until at least `expected` stderr bytes have been drained, so the
+  # retained tail is final before we assert on it (avoids racing the
+  # select-driven consumer against the OS process exit notification).
+  defp wait_until_drained(pid, expected, attempts \\ 100) do
+    if Proc.stats(pid).bytes_err >= expected do
+      :ok
+    else
+      if attempts == 0 do
+        flunk("stderr not fully drained: #{Proc.stats(pid).bytes_err}/#{expected}")
+      else
+        Process.sleep(20)
+        wait_until_drained(pid, expected, attempts - 1)
+      end
+    end
+  end
+
+  # Accumulates chunks until `expected` is fully received (chunking is
+  # arbitrary), then returns. Falls back to a timeout so a regression that
+  # drops stderr fails the assertion instead of hanging.
+  defp collect_output(expected, acc \\ "") do
+    if acc == expected do
+      acc
+    else
+      receive do
+        {:out, data} -> collect_output(expected, acc <> data)
+      after
+        2_000 -> acc
+      end
+    end
+  end
+end