push architecture slice 2: turn-started/turn-settled emit from the liveness seams (registry-first ordering, no-op paths silent)
This commit is contained in:
@@ -20,6 +20,8 @@
|
|||||||
* chat turns are minutes at most, so 30 minutes is far on the safe side.
|
* chat turns are minutes at most, so 30 minutes is far on the safe side.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
|
import { emit } from "../core/events.ts";
|
||||||
|
|
||||||
interface LhGlobal {
|
interface LhGlobal {
|
||||||
__lhTurnRegistry?: Map<string, number>;
|
__lhTurnRegistry?: Map<string, number>;
|
||||||
}
|
}
|
||||||
@@ -35,19 +37,32 @@ const MAX_TURN_MS = 30 * 60_000;
|
|||||||
/**
|
/**
|
||||||
* Mark a turn as starting. Returns false if one is already running for the
|
* Mark a turn as starting. Returns false if one is already running for the
|
||||||
* chat — the caller rejects the request (only one turn may own a chat).
|
* chat — the caller rejects the request (only one turn may own a chat).
|
||||||
|
* Announces "turn-started" AFTER the registry write: a listener that reads
|
||||||
|
* isTurnRunning or fetches the record sees consistent truth. The rejected
|
||||||
|
* double-begin (the 409 path) stays silent — no turn began.
|
||||||
*/
|
*/
|
||||||
export function beginTurn(chatId: string): boolean {
|
export function beginTurn(chatId: string): boolean {
|
||||||
if (isTurnRunning(chatId)) return false;
|
if (isTurnRunning(chatId)) return false;
|
||||||
inFlight.set(chatId, Date.now());
|
inFlight.set(chatId, Date.now());
|
||||||
|
emit("turn-started", { chatId });
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Mark the chat's turn as finished (runs in onFinish, aborted or not). */
|
/**
|
||||||
|
* Mark the chat's turn as finished (runs in onFinish, aborted or not —
|
||||||
|
* AFTER persistTurn, so "turn-settled" guarantees a subsequent fetch reads
|
||||||
|
* durable truth). Emits only when an entry was actually present: a
|
||||||
|
* defensive no-op endTurn must not fire a spurious settle.
|
||||||
|
*/
|
||||||
export function endTurn(chatId: string): void {
|
export function endTurn(chatId: string): void {
|
||||||
inFlight.delete(chatId);
|
if (inFlight.delete(chatId)) emit("turn-settled", { chatId });
|
||||||
}
|
}
|
||||||
|
|
||||||
/** True while the chat's turn is in flight; lazily evicts stale entries. */
|
/** True while the chat's turn is in flight; lazily evicts stale entries.
|
||||||
|
* The eviction path is DELIBERATELY event-free: it is a leak guard for a
|
||||||
|
* turn whose onFinish never fired (itself pathological), it runs lazily
|
||||||
|
* from a read, and emitting from a read is a side-effect smell. Silence
|
||||||
|
* here is a choice, not an oversight. */
|
||||||
export function isTurnRunning(chatId: string): boolean {
|
export function isTurnRunning(chatId: string): boolean {
|
||||||
const since = inFlight.get(chatId);
|
const since = inFlight.get(chatId);
|
||||||
if (since === undefined) return false;
|
if (since === undefined) return false;
|
||||||
|
|||||||
@@ -2,12 +2,15 @@
|
|||||||
* Turn liveness registry unit tests (src/server/turn-liveness.ts): one turn
|
* Turn liveness registry unit tests (src/server/turn-liveness.ts): one turn
|
||||||
* per chat, ever — begin/end round trip, double-begin rejection, restart
|
* per chat, ever — begin/end round trip, double-begin rejection, restart
|
||||||
* after end. The 30-minute stale-eviction TTL is not tested (it is a leak
|
* after end. The 30-minute stale-eviction TTL is not tested (it is a leak
|
||||||
* guard, not a behavior).
|
* guard, not a behavior). The lifecycle-emitter pins ride the same seams:
|
||||||
|
* turn-started on the accepted begin only, turn-settled on the live end
|
||||||
|
* only, and the registry already consistent when the listener runs.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
import assert from "node:assert/strict";
|
import assert from "node:assert/strict";
|
||||||
import { describe, it } from "node:test";
|
import { describe, it } from "node:test";
|
||||||
import { beginTurn, endTurn, isTurnRunning } from "../src/server/turn-liveness.ts";
|
import { beginTurn, endTurn, isTurnRunning } from "../src/server/turn-liveness.ts";
|
||||||
|
import { subscribe, type ServerEvent } from "../src/core/events.ts";
|
||||||
|
|
||||||
describe("turn liveness registry", () => {
|
describe("turn liveness registry", () => {
|
||||||
it("begin/end round trip; a second begin is rejected until the turn ends", () => {
|
it("begin/end round trip; a second begin is rejected until the turn ends", () => {
|
||||||
@@ -22,3 +25,36 @@ describe("turn liveness registry", () => {
|
|||||||
endTurn(id);
|
endTurn(id);
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("turn lifecycle events", () => {
|
||||||
|
it("beginTurn emits turn-started on success only — the 409 double-begin stays silent", () => {
|
||||||
|
const id = "chat_events_begin";
|
||||||
|
const got: ServerEvent[] = [];
|
||||||
|
const off = subscribe((e) => got.push(e));
|
||||||
|
assert.equal(beginTurn(id), true, "fresh chat accepts the turn");
|
||||||
|
assert.equal(beginTurn(id), false, "the rejected double-begin");
|
||||||
|
assert.deepEqual(got, [{ type: "turn-started", chatId: id }], "exactly one event");
|
||||||
|
endTurn(id); // cleanup (its settle is asserted in the next case's terms)
|
||||||
|
off();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("endTurn emits turn-settled only when a live entry was present", () => {
|
||||||
|
const id = "chat_events_end";
|
||||||
|
const got: ServerEvent[] = [];
|
||||||
|
let runningAtSettle: boolean | undefined;
|
||||||
|
const off = subscribe((e) => {
|
||||||
|
got.push(e);
|
||||||
|
// The settle must be observable as already-idle: the registry
|
||||||
|
// write happens BEFORE the emit, so a listener that fetches truth
|
||||||
|
// (the whole point of notify → fetch) reads consistent state.
|
||||||
|
if (e.type === "turn-settled") runningAtSettle = isTurnRunning(id);
|
||||||
|
});
|
||||||
|
beginTurn(id);
|
||||||
|
got.length = 0; // focus on the settle path
|
||||||
|
endTurn(id); // live entry → settle
|
||||||
|
endTurn(id); // absent entry → defensive no-op, must stay silent
|
||||||
|
assert.deepEqual(got, [{ type: "turn-settled", chatId: id }]);
|
||||||
|
assert.equal(runningAtSettle, false, "listener sees the registry already cleared");
|
||||||
|
off();
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user