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