From b8522ea098dd298441470cecb2aae08fafa9d59a Mon Sep 17 00:00:00 2001 From: Scott Karp Date: Tue, 6 Oct 2026 22:46:54 +0000 Subject: [PATCH] =?UTF-8?q?Tablet=2050:=20a=20socket=20reply=20is=20one=20?= =?UTF-8?q?line,=20however=20long=20=E2=80=94=20and=20a=20failed=20retry?= =?UTF-8?q?=20closes=20its=20socket?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both Unix-socket clients (Korban Processor, Talmudic Engine) read a reply with a single :gen_tcp.recv/3, which returns at most one receive buffer — 1460 bytes, measured — whatever the packet mode. A bedikah of forty claims answered in 18 KB: the client decoded the first 1460 bytes, failed, reconnected, failed again, and returned an error with the retry's socket still open. The single-threaded Korban Processor then sat on that dead conversation and every later bedikah, however small, timed out at 15 s. Live, for every caller, from the first long output. Clients now read until the newline (recv_line/2, 16 MiB ceiling) in raw mode, and close the retry's socket on failure. The Korban Processor drops a connection that says nothing for 120 s, so one stuck client can no longer hold the door. tablet50_socket_line_test stands up a 60 KB-line server and asks both clients. Elixir 552/0 (L3 on and off), korban 280. Co-Authored-By: Claude Fable 5.1 --- .../src/korban_processor/server.py | 7 +- .../vav_pipeline/korban_processor_socket.ex | 36 +++++++- .../vav_pipeline/talmudic_engine_socket.ex | 37 +++++++- .../test/tablet50_socket_line_test.exs | 85 +++++++++++++++++++ 4 files changed, 159 insertions(+), 6 deletions(-) create mode 100644 vav_pipeline/test/tablet50_socket_line_test.exs diff --git a/korban_processor/src/korban_processor/server.py b/korban_processor/src/korban_processor/server.py index 48db0ccf..7aec1adb 100644 --- a/korban_processor/src/korban_processor/server.py +++ b/korban_processor/src/korban_processor/server.py @@ -35,6 +35,7 @@ from korban_processor.bedikah import examine from korban_processor.korban import Provenance DEFAULT_SOCKET_PATH = "/tmp/shorashim/korban.sock" +IDLE_SECONDS = 120.0 # a client that says nothing for this long is dropped (tablet 50) def handle_request(line: str) -> dict: @@ -84,6 +85,10 @@ def handle_request(line: str) -> dict: def _serve_connection(conn: socket.socket) -> None: + # One reader at a time means one stuck reader stops everyone (tablet + # 50: a client that left a connection open held the door for every + # later caller). A conversation that goes quiet is closed. + conn.settimeout(IDLE_SECONDS) with conn, conn.makefile("rb") as reader, conn.makefile("wb") as writer: for raw in reader: line = raw.decode("utf-8", errors="replace").strip() @@ -117,7 +122,7 @@ def serve(socket_path: str = DEFAULT_SOCKET_PATH, stop: Optional[threading.Event continue try: _serve_connection(conn) - except (BrokenPipeError, ConnectionResetError): + except (BrokenPipeError, ConnectionResetError, socket.timeout, TimeoutError): pass finally: server.close() diff --git a/vav_pipeline/lib/vav_pipeline/korban_processor_socket.ex b/vav_pipeline/lib/vav_pipeline/korban_processor_socket.ex index 22e03dde..988fd256 100644 --- a/vav_pipeline/lib/vav_pipeline/korban_processor_socket.ex +++ b/vav_pipeline/lib/vav_pipeline/korban_processor_socket.ex @@ -71,6 +71,11 @@ defmodule VavPipeline.KorbanProcessorSocket do {:reply, {:ok, reply}, state} else {:error, reason2} -> + # Close the retry's socket too. Left open (tablet 50), it held + # the single-threaded processor on a conversation that would + # never continue, and every later caller queued behind it. + if state.socket, do: :gen_tcp.close(state.socket) + {:reply, {:error, "Korban Processor unreachable: #{inspect(reason)} / #{inspect(reason2)}"}, %{state | socket: nil}} end @@ -84,7 +89,7 @@ defmodule VavPipeline.KorbanProcessorSocket do end defp ensure_connected(%{socket: nil, socket_path: path} = state) do - case :gen_tcp.connect({:local, path}, 0, [:binary, active: false, packet: :line], @connect_timeout) do + case :gen_tcp.connect({:local, path}, 0, [:binary, active: false, packet: :raw], @connect_timeout) do {:ok, socket} -> {:ok, %{state | socket: socket}} {:error, reason} -> {:error, reason} end @@ -94,7 +99,7 @@ defmodule VavPipeline.KorbanProcessorSocket do defp send_and_receive(socket, json_line) do with :ok <- :gen_tcp.send(socket, json_line <> "\n"), - {:ok, data} <- :gen_tcp.recv(socket, 0, @call_timeout), + {:ok, data} <- recv_line(socket, @call_timeout), {:ok, parsed} <- Jason.decode(String.trim_trailing(data, "\n")) do {:ok, parsed} else @@ -102,4 +107,31 @@ defmodule VavPipeline.KorbanProcessorSocket do {:error, reason} -> {:error, reason} end end + + # One reply is one line. `:gen_tcp.recv/3` hands back at most one + # receive buffer (1460 bytes by default — measured, 2026-10-06, tablet + # 50) whatever the packet mode, so a reply longer than that arrived in + # pieces, the first piece failed to decode, and the client reconnected + # mid-conversation; every later call timed out. Read until the newline. + @max_line_bytes 16_777_216 + + defp recv_line(socket, timeout), do: recv_line(socket, timeout, [], 0) + + defp recv_line(_socket, _timeout, _acc, size) when size > @max_line_bytes, + do: {:error, :reply_too_long} + + defp recv_line(socket, timeout, acc, size) do + case :gen_tcp.recv(socket, 0, timeout) do + {:ok, chunk} -> + if String.ends_with?(chunk, "\n") do + {:ok, IO.iodata_to_binary(Enum.reverse([chunk | acc]))} + else + recv_line(socket, timeout, [chunk | acc], size + byte_size(chunk)) + end + + {:error, reason} -> + {:error, reason} + end + end + end diff --git a/vav_pipeline/lib/vav_pipeline/talmudic_engine_socket.ex b/vav_pipeline/lib/vav_pipeline/talmudic_engine_socket.ex index 45a80a0f..df069cc8 100644 --- a/vav_pipeline/lib/vav_pipeline/talmudic_engine_socket.ex +++ b/vav_pipeline/lib/vav_pipeline/talmudic_engine_socket.ex @@ -130,7 +130,11 @@ defmodule VavPipeline.TalmudicEngineSocket do {:ok, state} -> case send_and_receive(state.socket, json_line) do {:ok, response} -> {:reply, {:ok, response}, state} - {:error, reason2} -> {:reply, {:error, "Retry failed: #{inspect(reason2)}"}, %{state | socket: nil}} + {:error, reason2} -> + # Close the retry's socket too (tablet 50): left open it + # holds the engine on a dead conversation. + :gen_tcp.close(state.socket) + {:reply, {:error, "Retry failed: #{inspect(reason2)}"}, %{state | socket: nil}} end {:error, reason2} -> @@ -155,7 +159,7 @@ defmodule VavPipeline.TalmudicEngineSocket do # --- Private --- defp ensure_connected(%{socket: nil, socket_path: path} = state) do - case :gen_tcp.connect({:local, path}, 0, [:binary, active: false, packet: :line], @connect_timeout) do + case :gen_tcp.connect({:local, path}, 0, [:binary, active: false, packet: :raw], @connect_timeout) do {:ok, socket} -> {:ok, %{state | socket: socket}} @@ -169,7 +173,7 @@ defmodule VavPipeline.TalmudicEngineSocket do defp send_and_receive(socket, json_line) do case :gen_tcp.send(socket, json_line <> "\n") do :ok -> - case :gen_tcp.recv(socket, 0, @call_timeout) do + case recv_line(socket, @call_timeout) do {:ok, data} -> line = String.trim_trailing(data, "\n") case Jason.decode(line) do @@ -203,4 +207,31 @@ defmodule VavPipeline.TalmudicEngineSocket do defp call_engine(json_line) do GenServer.call(__MODULE__, {:request, json_line}, @call_timeout) end + + # One reply is one line. `:gen_tcp.recv/3` hands back at most one + # receive buffer (1460 bytes by default — measured, 2026-10-06, tablet + # 50) whatever the packet mode, so a reply longer than that arrived in + # pieces, the first piece failed to decode, and the client reconnected + # mid-conversation; every later call timed out. Read until the newline. + @max_line_bytes 16_777_216 + + defp recv_line(socket, timeout), do: recv_line(socket, timeout, [], 0) + + defp recv_line(_socket, _timeout, _acc, size) when size > @max_line_bytes, + do: {:error, :reply_too_long} + + defp recv_line(socket, timeout, acc, size) do + case :gen_tcp.recv(socket, 0, timeout) do + {:ok, chunk} -> + if String.ends_with?(chunk, "\n") do + {:ok, IO.iodata_to_binary(Enum.reverse([chunk | acc]))} + else + recv_line(socket, timeout, [chunk | acc], size + byte_size(chunk)) + end + + {:error, reason} -> + {:error, reason} + end + end + end diff --git a/vav_pipeline/test/tablet50_socket_line_test.exs b/vav_pipeline/test/tablet50_socket_line_test.exs new file mode 100644 index 00000000..58afd2f1 --- /dev/null +++ b/vav_pipeline/test/tablet50_socket_line_test.exs @@ -0,0 +1,85 @@ +defmodule VavPipeline.Tablet50SocketLineTest do + @moduledoc """ + Tablet 50 (2026-10-06) — one reply is one line, however long. + + Both socket clients read a reply with a single `:gen_tcp.recv/3`, which + returns at most one receive buffer (1460 bytes). A bedikah of forty + claims answered in 18 KB; the client decoded the first 1460 bytes, + failed, reconnected, failed again, and left the retry's socket open — + the single-threaded Korban Processor then waited on that dead + conversation and every later bedikah timed out. Live, for everyone. + + This test stands up a one-line server that answers with a long line + and a short one, and asks both clients for them. + """ + + use ExUnit.Case, async: false + + alias VavPipeline.{KorbanProcessorSocket, TalmudicEngineSocket} + + @long_pad String.duplicate("x", 60_000) + + setup do + path = Path.join(System.tmp_dir!(), "tablet50_#{System.unique_integer([:positive])}.sock") + File.rm(path) + + {:ok, listen} = + :gen_tcp.listen(0, [:binary, active: false, packet: :raw, ifaddr: {:local, String.to_charlist(path)}]) + + parent = self() + + server = + spawn_link(fn -> + loop = fn loop -> + {:ok, conn} = :gen_tcp.accept(listen) + serve = fn serve -> + case read_line(conn) do + {:ok, line} -> + send(parent, {:server_got, byte_size(line)}) + reply = Jason.encode!(%{"status" => "ok", "examination" => %{"pad" => @long_pad, "claims" => []}, "violations" => [], "action" => "t"}) + :ok = :gen_tcp.send(conn, reply <> "\n") + serve.(serve) + _ -> :gen_tcp.close(conn) + end + end + serve.(serve) + loop.(loop) + end + loop.(loop) + end) + + on_exit(fn -> Process.exit(server, :kill); File.rm(path) end) + %{path: path} + end + + test "the Korban client reads a 60 KB reply whole, and the next call still works", %{path: path} do + {:ok, pid} = GenServer.start_link(KorbanProcessorSocket, [socket_path: path], name: :tablet50_korban) + on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end) + + for _ <- 1..3 do + assert {:ok, %{"status" => "ok", "examination" => %{"pad" => pad}}} = + GenServer.call(:tablet50_korban, {:request, Jason.encode!(%{"action" => "examine", "output" => String.duplicate("y", 20_000)})}, 15_000) + + assert byte_size(pad) == 60_000 + assert_receive {:server_got, got}, 5_000 + assert got > 20_000 + end + end + + test "the Talmudic client reads a 60 KB reply whole", %{path: path} do + {:ok, pid} = GenServer.start_link(TalmudicEngineSocket, [socket_path: path], name: :tablet50_talmudic) + on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end) + + assert {:ok, %{"status" => "ok"} = reply} = GenServer.call(:tablet50_talmudic, {:request, Jason.encode!(%{"action" => "halt_check"})}, 15_000) + assert byte_size(reply["examination"]["pad"]) == 60_000 + end + + defp read_line(conn, acc \\ "") do + case :gen_tcp.recv(conn, 0, 5_000) do + {:ok, chunk} -> + acc = acc <> chunk + if String.ends_with?(acc, "\n"), do: {:ok, acc}, else: read_line(conn, acc) + err -> err + end + end +end