From 2c0bd58dfc1580e5708da2f24433c61cf4771663 Mon Sep 17 00:00:00 2001 From: Avery Yen <976748+haplesshero13@users.noreply.github.com> Date: Thu, 30 Jul 2026 17:53:13 -0400 Subject: [PATCH] Stop an abandoned chat turn from blocking the next one Closing a tab mid-reply left chat unusable for up to a minute. The turn holds a global slot, and the only release paths were finishing normally or noticing a failed SSE write. Neither fires promptly: Mist keeps the SSE process alive after a disconnect, and writes to a half-closed socket keep succeeding, so the turn just runs to completion while every new message gets 409 AGENT_BUSY. A new turn now signals the incumbent to stop and takes the slot. Two things had to change for that signal to land: - Cancel flags moved from the process dictionary to ETS keyed by pid. The process dictionary is per-process, so one turn could never cancel another. - The SSE process claims ownership itself via claim_turn/0. try_acquire_turn/0 runs in Mist's per-request handler process, which is not the process that blocks on the model, so the pid it recorded was never the one to cancel. Also shortens the heartbeat from 15s to 5s, which is what gives the incumbent a chance to observe the cancel flag between model chunks. Strict serialization is gone by choice: for a single household, blocking the person is worse than the interleaving it prevented. Co-Authored-By: Claude Fable 5 --- server/src/cacheboard/server.gleam | 4 ++ server/src/cacheboard_server_ffi.erl | 85 ++++++++++++++++++++++++---- 2 files changed, 77 insertions(+), 12 deletions(-) diff --git a/server/src/cacheboard/server.gleam b/server/src/cacheboard/server.gleam index 13f1019..2e0aa57 100644 --- a/server/src/cacheboard/server.gleam +++ b/server/src/cacheboard/server.gleam @@ -38,6 +38,9 @@ fn complete_model( @external(erlang, "cacheboard_server_ffi", "reset_cancelled") fn reset_cancelled() -> Nil +@external(erlang, "cacheboard_server_ffi", "claim_turn") +fn claim_turn() -> Nil + @external(erlang, "cacheboard_server_ffi", "mark_cancelled") fn mark_cancelled() -> Nil @@ -232,6 +235,7 @@ fn chat( case event { ChatRun(message) -> { reset_cancelled() + claim_turn() reset_events() let primary_url = envoy.get("LM_BASE_URL") diff --git a/server/src/cacheboard_server_ffi.erl b/server/src/cacheboard_server_ffi.erl index db77c46..400055e 100644 --- a/server/src/cacheboard_server_ffi.erl +++ b/server/src/cacheboard_server_ffi.erl @@ -1,7 +1,7 @@ -module(cacheboard_server_ffi). -export([ secret/0, stream_model/4, complete_model/3, reset_cancelled/0, - mark_cancelled/0, init_turn_lock/0, try_acquire_turn/0, release_turn/0, reset_events/0, + mark_cancelled/0, init_turn_lock/0, try_acquire_turn/0, claim_turn/0, release_turn/0, reset_events/0, record_event/2, events_json/0, simplefin_request/4, unix_date/1, unix_now/0, current_month/0, date_plus_days/1, classify_account/1 ]). @@ -9,32 +9,89 @@ secret() -> base64:encode(crypto:strong_rand_bytes(64)). +%% Cancel flags live in ETS keyed by pid, not the process dictionary, so a +%% newly arriving turn can cancel the turn that currently holds the lock. reset_cancelled() -> - erase(cacheboard_cancelled), + ensure_turn_table(), + ets:delete(cacheboard_agent_cancel, self()), nil. mark_cancelled() -> - put(cacheboard_cancelled, true), + ensure_turn_table(), + ets:insert(cacheboard_agent_cancel, {self(), true}), nil. +is_cancelled() -> + ensure_turn_table(), + case ets:lookup(cacheboard_agent_cancel, self()) of + [{_, true}] -> true; + _ -> false + end. + init_turn_lock() -> ensure_turn_table(), nil. +%% The turn lock records its owning pid rather than a bare `true`. +%% +%% A chat turn runs synchronously inside the SSE process, and release_turn/0 +%% only runs on the normal path. When a client disconnects, Mist tears that +%% process down without unwinding, so a bare flag leaks and every later turn +%% gets AGENT_BUSY forever - one abandoned request wedged chat permanently. +%% Storing the pid lets a new turn detect a dead owner and take over. +%% Newest turn wins. +%% +%% Strict serialization blocked the user whenever a previous turn was +%% abandoned: a closed tab leaves the SSE process alive, writes to a +%% half-closed socket keep succeeding, and the turn holds the slot until the +%% model finishes on its own. Making the person wait is worse than the +%% interleaving serialization was protecting against. +%% +%% Note the process split: this runs in Mist's per-request handler process, +%% while the turn itself runs in the SSE process. Only the SSE process can +%% claim ownership (see claim_turn/0), because that is the pid a cancel +%% signal has to reach. try_acquire_turn() -> ensure_turn_table(), - ets:insert_new(cacheboard_agent_turn, {busy, true}). + case ets:lookup(cacheboard_agent_turn, busy) of + [{busy, Pid}] when is_pid(Pid) -> + case is_process_alive(Pid) of + true -> ets:insert(cacheboard_agent_cancel, {Pid, true}); + false -> true + end; + _ -> true + end, + true. + +%% Called from the SSE process at the start of a turn: it records itself as +%% the owner, so a later turn's cancel signal reaches the process that is +%% actually blocked on the model. +claim_turn() -> + ensure_turn_table(), + ets:insert(cacheboard_agent_turn, {busy, self()}), + ets:delete(cacheboard_agent_cancel, self()), + nil. +%% Only the owner may release, so a turn that finishes late cannot clear a +%% lock that has since been taken over by a newer turn. release_turn() -> ensure_turn_table(), - ets:delete(cacheboard_agent_turn, busy), + Self = self(), + case ets:lookup(cacheboard_agent_turn, busy) of + [{busy, Self}] -> ets:delete(cacheboard_agent_turn, busy); + _ -> true + end, nil. ensure_turn_table() -> - case ets:whereis(cacheboard_agent_turn) of + ensure_table(cacheboard_agent_turn), + ensure_table(cacheboard_agent_cancel). + +ensure_table(Name) -> + case ets:whereis(Name) of undefined -> - try ets:new(cacheboard_agent_turn, [named_table, public, set]) - catch error:badarg -> cacheboard_agent_turn + try ets:new(Name, [named_table, public, set]) + catch error:badarg -> Name end; Table -> Table end. @@ -143,7 +200,7 @@ classify_account_loop(Label, Allowed) -> complete_model(BaseUrl, RequestBody, OnDelta) -> application:ensure_all_started(inets), application:ensure_all_started(ssl), - case get(cacheboard_cancelled) of + case is_cancelled() of true -> {error, {false, <<"client disconnected">>}}; _ -> Url = binary_to_list(<>), @@ -180,7 +237,7 @@ complete_loop(RequestId, OnDelta, Buffer, Content, Fragments, Started) -> Fragments, Started ), - case get(cacheboard_cancelled) of + case is_cancelled() of true -> httpc:cancel_request(RequestId), {error, {NextStarted, <<"client disconnected">>}}; @@ -222,9 +279,13 @@ complete_loop(RequestId, OnDelta, Buffer, Content, Fragments, Started) -> false -> {error, {Started, <<"model returned HTTP ", (integer_to_binary(Status))/binary>>}} end - after 15000 -> + after 5000 -> + %% Heartbeat. Writing an empty delta is how a dead socket gets noticed: + %% the Gleam emit path attempts an SSE write and marks the turn cancelled + %% when it fails. Without this the turn blocks on the model, never writes, + %% and keeps the single turn slot until it finishes on its own. OnDelta(<<>>), - case get(cacheboard_cancelled) of + case is_cancelled() of true -> httpc:cancel_request(RequestId), {error, {Started, <<"client disconnected">>}}; -- 2.54.0