Tablet 50: a socket reply is one line, however long — and a failed retry closes its socket

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 <noreply@anthropic.com>

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

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