From 1807f3633b286b50c2aa4218d549103749bd646c Mon Sep 17 00:00:00 2001 From: Niko Maroulis Date: Sun, 28 Jun 2026 12:54:40 -0400 Subject: [PATCH 1/2] Fix unbounded stderr_buffer leak; make Daemon stderr reliable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In the default :consume stderr mode, every stderr chunk was prepended to state.stderr_buffer — a list that was never read anywhere in lib/. For long-running processes (especially NetRunner.Daemon) this grew the GenServer heap without bound, retaining data nothing could use. Replace it with a bounded tail: - Retain only the most-recent :stderr_tail_bytes bytes (default 8192) as a single binary; drain-and-drop the rest. Stats still count every byte. - Expose it via NetRunner.Process.stderr_tail/1 (the long-orphaned AbnormalExit.stderr feature; kept defined-but-unraised for a future opt-in). - Add and validate the :stderr_tail_bytes option (non-negative integer; 0 disables retention while still draining). Also fix a stderr-pipe race in NetRunner.Daemon: it ran its own stderr drain task while leaving the default :consume internal consumer on, so the two raced the same FD and on_output saw an arbitrary subset. Daemon now forces stderr: :disabled so its drain task is the sole reader and on_output receives the full stream in order. No C/shepherd/protocol changes. PTY mode unaffected. --- lib/net_runner.ex | 7 +- lib/net_runner/daemon.ex | 15 ++- lib/net_runner/process.ex | 42 ++++++++- lib/net_runner/process/exec.ex | 11 +++ lib/net_runner/process/state.ex | 9 +- test/stderr_tail_test.exs | 158 ++++++++++++++++++++++++++++++++ 6 files changed, 235 insertions(+), 7 deletions(-) create mode 100644 test/stderr_tail_test.exs diff --git a/lib/net_runner.ex b/lib/net_runner.ex index f88c587..869594b 100644 --- a/lib/net_runner.ex +++ b/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}`. diff --git a/lib/net_runner/daemon.ex b/lib/net_runner/daemon.ex index 9262e4d..64b80df 100644 --- a/lib/net_runner/daemon.ex +++ b/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} -> diff --git a/lib/net_runner/process.ex b/lib/net_runner/process.ex index 31c9345..a0e8f02 100644 --- a/lib/net_runner/process.ex +++ b/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 diff --git a/lib/net_runner/process/exec.ex b/lib/net_runner/process/exec.ex index e767260..b3afbd6 100644 --- a/lib/net_runner/process/exec.ex +++ b/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 diff --git a/lib/net_runner/process/state.ex b/lib/net_runner/process/state.ex index bec7772..c8fb743 100644 --- a/lib/net_runner/process/state.ex +++ b/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 diff --git a/test/stderr_tail_test.exs b/test/stderr_tail_test.exs new file mode 100644 index 0000000..ef4d956 --- /dev/null +++ b/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 From e2a6f4a8935793f6e5b14e8ddc313e59e0e7194e Mon Sep 17 00:00:00 2001 From: Niko Maroulis Date: Sun, 28 Jun 2026 12:54:46 -0400 Subject: [PATCH 2/2] ci: bump macOS test matrix to OTP 28 / Elixir 1.18, checkout to v7 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The macOS test job ran OTP 27.2 / Elixir 1.17.3, which setup-beam does not reliably provide for the now-arm64 macos-latest runner — bump it to OTP 28.3.3 / Elixir 1.18.4 to match a supported build. Also bump actions/checkout from the aging v4 to v7 across all jobs. --- .github/workflows/ci.yml | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6b335fc..666d749 100644 --- a/.github/workflows/ci.yml +++ b/.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: