From 82124a872ff8bc85b08597ec00797c473bff2384 Mon Sep 17 00:00:00 2001 From: deepfates Date: Tue, 29 Sep 2026 12:36:13 -0700 Subject: [PATCH 1/4] Load the model catalog before the suite starts LLMDB loads ReqLLM's model catalog on first use under a :global.trans lock, and a process that finds the lock taken sleeps a random backoff that doubles up to 8 s. Async tests making their first model lookup together at the start of the suite waited several times the 1.4 s load, and on a loaded runner past a 5 s Task.await in ReqLLMStreamEndTest. --- test/test_helper.exs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/test/test_helper.exs b/test/test_helper.exs index 4e459308..9b0044e2 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -78,5 +78,14 @@ external_excludes = ExUnit.configure(exclude: external_excludes) +# ReqLLM reads its model catalog from LLMDB, which loads it on first use under +# a :global.trans lock. A process that finds the lock taken sleeps a random +# backoff that doubles up to 8 s before it asks again, so when the async tests +# at the start of the suite all make their first model lookup at once, some +# wait several times as long as the load itself (a 1.4 s load, waits of up to +# 4.6 s among 16 first lookups on an idle machine), and on a loaded runner past +# the 5 s of a Task.await. Loading it here, before any test runs, leaves nothing to contend for. +{:ok, _catalog} = LLMDB.load() + Imp.Test.OwnLog.install() ExUnit.start() From 6c6c8ca2a21849e2375fda4d5099b8cddf28f6bd Mon Sep 17 00:00:00 2001 From: deepfates Date: Tue, 29 Sep 2026 12:36:13 -0700 Subject: [PATCH 2/4] TimedOutConnectionTest: no timer decides the result The first request is held until the test ends rather than for 2 s, so the first call times out however slow the machine is, and the second call keeps ReqLLM's own receive timeout instead of one the test picks. A second request on the stuck connection is still never answered. --- test/timed_out_connection_test.exs | 26 ++++++++++++++++---------- 1 file changed, 16 insertions(+), 10 deletions(-) diff --git a/test/timed_out_connection_test.exs b/test/timed_out_connection_test.exs index d84a4bf8..965fa9be 100644 --- a/test/timed_out_connection_test.exs +++ b/test/timed_out_connection_test.exs @@ -23,21 +23,24 @@ defmodule Imp.TimedOutConnectionTest do } test "the call after a timed-out call is answered on a new connection" do + test_pid = self() {:ok, seen} = Agent.start_link(fn -> [] end) # Bandit serves each HTTP/1 connection from one process, so the handler's - # pid names the connection a request came in on. The first request is held - # past the client's timeout; every later one is answered at once. + # pid names the connection a request came in on. The first request is not + # answered while the test runs, so the first call times out however slow + # the machine is, and its connection stays busy: a request written behind + # it would not be answered either. Every later request is answered at once. base_url = Imp.Test.LocalHTTP.start(fn _request -> connection = self() count = Agent.get_and_update(seen, &{length(&1) + 1, &1 ++ [connection]}) if count == 1 do + test_ref = Process.monitor(test_pid) + receive do - :release -> :ok - after - 2_000 -> :ok + {:DOWN, ^test_ref, :process, _pid, _reason} -> :ok end end @@ -61,13 +64,16 @@ defmodule Imp.TimedOutConnectionTest do # connection only when it draws the same shard. Choosing the first shard # for every request makes that meeting certain. finch = [name: ReqLLM.Application.finch_name(), pool_strategy: &hd/1] - opts = [timeout: 200, max_retries: 0, req_http_options: [finch: finch]] + opts = [max_retries: 0, req_http_options: [finch: finch]] messages = [%{role: :user, content: "ping"}] - assert {:error, %Imp.LMError{retryable: true}} = Imp.LM.generate(lm, messages, opts) - # Only the first call has to time out. The second gets room to be answered - # on a loaded machine; landing on the stuck connection still fails below. - assert {:ok, _reply} = Imp.LM.generate(lm, messages, Keyword.put(opts, :timeout, 5_000)) + assert {:error, %Imp.LMError{retryable: true}} = + Imp.LM.generate(lm, messages, Keyword.put(opts, :timeout, 200)) + + # The second call keeps ReqLLM's own receive timeout. On a new connection + # it is answered at once; on the stuck one it is not answered, and fails + # when that timeout runs out. + assert {:ok, _reply} = Imp.LM.generate(lm, messages, opts) assert [first, second] = Agent.get(seen, & &1) refute first == second From be813ba9208748beb147a578c1dec49816489da3 Mon Sep 17 00:00:00 2001 From: deepfates Date: Tue, 29 Sep 2026 12:36:19 -0700 Subject: [PATCH 3/4] CSV walk test: the walk's exit reason decides, not a time window The walk is found for as long as the caller is alive, since the caller parses the whole file before it starts it. The walk is then suspended, so it cannot finish and exit on its own before the caller is killed, and the test asserts that it ended killed. --- test/datasets_contract_test.exs | 42 ++++++++++++++++++++------------- 1 file changed, 25 insertions(+), 17 deletions(-) diff --git a/test/datasets_contract_test.exs b/test/datasets_contract_test.exs index 3cdc7d8b..f5012289 100644 --- a/test/datasets_contract_test.exs +++ b/test/datasets_contract_test.exs @@ -206,11 +206,17 @@ defmodule DatasetsContractTest do File.write!(path, ["q,a\n" | rows] ++ ["bad,row,extra\n"]) caller = spawn(fn -> Datasets.csv(path, [:q]) end) - walk = wait_for_monitored(caller) + walk = wait_for_walk(caller) ref = Process.monitor(walk) + # Suspended, the walk cannot run to its end and exit on its own, so it + # ends only when something kills it. The suspension is lifted if the test + # process exits. + :erlang.suspend_process(walk) Process.exit(caller, :kill) - assert_receive {:DOWN, ^ref, :process, ^walk, :killed}, 1_000 + receive do + {:DOWN, ^ref, :process, ^walk, reason} -> assert reason == :killed + end after cleanup_tmp("killed-caller.csv") end @@ -434,13 +440,15 @@ defmodule DatasetsContractTest do cleanup_tmp("typed-math.jsonl") end - # The process `pid` monitors, once it monitors one. - # The caller also monitors other processes for a moment (the file server - # while it reads), so the walk is the monitored process running the CSV code. - defp wait_for_monitored(pid, tries \\ 500) do - walk = - case Process.info(pid, :monitors) do - {:monitors, monitors} -> + # The walk `caller` starts, once it has started it. The caller also monitors + # other processes for a moment (the file server while it reads), so the walk + # is the monitored process running the CSV code. The caller parses the whole + # file before it starts the walk, which takes as long as the machine is slow, + # so this waits for as long as the caller is alive. + defp wait_for_walk(caller) do + case Process.info(caller, :monitors) do + {:monitors, monitors} -> + walk = Enum.find_value(monitors, fn {:process, monitored} when is_pid(monitored) -> if csv_walk?(monitored), do: monitored @@ -449,15 +457,15 @@ defmodule DatasetsContractTest do nil end) - nil -> - nil - end + if walk do + walk + else + Process.sleep(5) + wait_for_walk(caller) + end - if walk || tries == 0 do - walk - else - Process.sleep(5) - wait_for_monitored(pid, tries - 1) + nil -> + flunk("the CSV caller ended before it started a walk") end end From 31eda098fff65ce6a3762519a976a4c23fdf1c60 Mon Sep 17 00:00:00 2001 From: deepfates Date: Tue, 29 Sep 2026 12:36:19 -0700 Subject: [PATCH 4/4] ParitySidecarTest: time a tree out once it is ready A 150 ms timeout fired on a loaded runner before Python had started the tree, so the output held no "tree ready". The three tree tests run with no timer, wait for the tree, and send the port owner the message its timer sends; a separate test checks that the timeout option ends a command. Waiting for the process ids has no attempt limit. --- test/parity_sidecar_test.exs | 78 +++++++++++++++++++++++------------- 1 file changed, 50 insertions(+), 28 deletions(-) diff --git a/test/parity_sidecar_test.exs b/test/parity_sidecar_test.exs index 1935d9d9..d462feba 100644 --- a/test/parity_sidecar_test.exs +++ b/test/parity_sidecar_test.exs @@ -76,14 +76,19 @@ defmodule Imp.BenchmarkTruth.ParitySidecarTest do refute process_group_alive?(group_pid) end + test "the timeout option ends a command that does not finish" do + assert {:error, :timeout, %Output{}} = + ParitySidecar.run(@python, ["-c", "import time; time.sleep(60)"], timeout: 150) + end + test "timeout returns captured output and terminates the process tree" do pid_path = tmp_path("timeout") - assert {:error, :timeout, %Output{} = output} = - ParitySidecar.run(@python, process_tree_args(pid_path), timeout: 150) + {result, [parent_pid, child_pid, group_pid]} = + time_out_ready_tree(process_tree_args(pid_path), pid_path) + assert {:error, :timeout, %Output{} = output} = result assert output.text =~ "tree ready" - [parent_pid, child_pid, group_pid] = await_pids(pid_path) refute await_alive?(parent_pid) refute await_alive?(child_pid) refute process_group_alive?(group_pid) @@ -92,14 +97,13 @@ defmodule Imp.BenchmarkTruth.ParitySidecarTest do test "bounded cleanup kills a fully TERM-resistant process tree" do pid_path = tmp_path("term-resistant") - assert {:error, :timeout, %Output{}} = - ParitySidecar.run( - @python, - process_tree_args(pid_path, leader_ignore_term: true, child_ignore_term: true), - timeout: 150 - ) + {result, [parent_pid, child_pid, group_pid]} = + time_out_ready_tree( + process_tree_args(pid_path, leader_ignore_term: true, child_ignore_term: true), + pid_path + ) - [parent_pid, child_pid, group_pid] = await_pids(pid_path) + assert {:error, :timeout, %Output{}} = result refute await_alive?(parent_pid) refute await_alive?(child_pid) refute process_group_alive?(group_pid) @@ -108,18 +112,14 @@ defmodule Imp.BenchmarkTruth.ParitySidecarTest do test "Port exit does not hide a TERM-resistant child with redirected stdio" do pid_path = tmp_path("mixed-tree") - assert {:error, :timeout, %Output{text: text}} = - ParitySidecar.run( - @python, - process_tree_args(pid_path, - child_ignore_term: true, - child_redirect_stdio: true - ), - timeout: 150 - ) + {result, [parent_pid, child_pid, group_pid]} = + time_out_ready_tree( + process_tree_args(pid_path, child_ignore_term: true, child_redirect_stdio: true), + pid_path + ) + assert {:error, :timeout, %Output{text: text}} = result assert text =~ "tree ready" - [parent_pid, child_pid, group_pid] = await_pids(pid_path) refute await_alive?(parent_pid) refute await_alive?(child_pid) refute process_group_alive?(group_pid) @@ -203,11 +203,11 @@ defmodule Imp.BenchmarkTruth.ParitySidecarTest do #{if leader_ignore_term, do: "signal.signal(signal.SIGTERM, signal.SIG_IGN)", else: ""} child_code = #{inspect(child_code(child_ignore_term))} child = subprocess.Popen([sys.executable, "-c", child_code]#{child_stdio}) + print("tree ready", flush=True) with open(#{inspect(pid_path)}, "w", encoding="utf-8") as handle: handle.write(f"{os.getpid()}\\n{child.pid}\\n{os.getpgrp()}\\n") handle.flush() os.fsync(handle.fileno()) - print("tree ready", flush=True) time.sleep(60) """ @@ -220,25 +220,47 @@ defmodule Imp.BenchmarkTruth.ParitySidecarTest do defp child_code(false), do: "import time; time.sleep(60)" - defp await_pids(path, attempts \\ 100) - defp await_pids(_path, 0), do: flunk("sidecar did not publish process ids") + # A timeout is a timer in the process that owns the sidecar's port, counted + # from the start. A timer short enough for a fast test fires, on a loaded + # machine, before Python has started the tree, and a long one makes every + # run wait it out. So these tests run the tree with no timer, wait until it + # is ready, and send its owner the message the timer sends. + defp time_out_ready_tree(args, pid_path) do + task = Task.async(fn -> ParitySidecar.run(@python, args) end) + [parent_pid | _rest] = pids = await_pids(pid_path) + send(port_owner(parent_pid), :command_timeout) + {Task.await(task, :infinity), pids} + end + + defp port_owner(os_pid) do + Enum.find_value(Port.list(), fn port -> + with {:os_pid, ^os_pid} <- Port.info(port, :os_pid), + {:connected, owner} <- Port.info(port, :connected) do + owner + else + _other -> nil + end + end) || flunk("no port runs OS process #{os_pid}") + end - defp await_pids(path, attempts) do + # The sidecar writes its process ids once the tree is running, which takes + # as long as the machine is slow. + defp await_pids(path) do case File.read(path) do {:ok, contents} -> case contents |> String.split("\n", trim: true) |> Enum.map(&Integer.parse/1) do [{parent, ""}, {child, ""}, {group, ""}] -> [parent, child, group] - _other -> retry_pids(path, attempts) + _other -> retry_pids(path) end _other -> - retry_pids(path, attempts) + retry_pids(path) end end - defp retry_pids(path, attempts) do + defp retry_pids(path) do Process.sleep(20) - await_pids(path, attempts - 1) + await_pids(path) end defp await_alive?(pid, attempts \\ 100)