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">>}};