mptyl/ThothII
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(backend): generation-aware SSE event ids — stale cursors can no longer eat events

Browse Source 
Audit finding 3.1 (high, 3/3 reviewer consensus). Event ids restart at 1
when the backend restarts; a browser auto-reconnect carrying the old
numeric Last-Event-ID was honored whenever the new process had already
emitted that many events, silently suppressing fresh events (same ids,
different content). The previous guard only caught cursor > lastId.

Wire ids are now "<generation>:<seq>" (generation = per-hub instance
token; seq = the existing per-session monotonic counter). The hub parses
raw header/query candidates itself: other-generation and legacy bare-
number cursors are stale → replay from the beginning; same-generation
cursors keep the newest-valid-wins behavior. EventSource treats ids as
opaque, so no frontend change.

Finding 3.2 (eviction) resolved by NOT evicting: close keeps the seq
counter on purpose (sessions reopen; monotonicity is what makes old
cursors detectable) — documented at the call site; buffers are emptied by
clear() and ring-bounded at 200.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
marcopan
2026-07-20 01:32:47 +02:00
co-authored by Claude Fable 5
parent 3e072fe652
commit 2958b32fd5
4 changed files with 171 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/test/sse-route.test.ts
+48 -28
View File
@@ -3,6 +3,8 @@ import { buildApp } from "../src/app.js";
import { loadConfig } from "../src/config.js";
import { SseHub } from "../src/sse/sse-hub.js";
const seqOf = (id: string) => Number(id.split(":")[1]);
async function readUntil(
reader: ReadableStreamDefaultReader<Uint8Array>,
predicate: (text: string) => boolean,
@@ -24,11 +26,15 @@ async function readUntil(
return text;
}
async function captureReplay(options: { query?: string; lastEventId?: string }): Promise<string> {
async function captureReplay(
options: { query?: (ids: string[]) => string; lastEventId?: (ids: string[]) => string },
): Promise<{ body: string; ids: string[] }> {
const hub = new SseHub();
hub.publish("s1", "info", { type: "info", text: "one" });
hub.publish("s1", "info", { type: "info", text: "two" });
hub.publish("s1", "info", { type: "info", text: "three" });
const ids = [
hub.publish("s1", "info", { type: "info", text: "one" }),
hub.publish("s1", "info", { type: "info", text: "two" }),
hub.publish("s1", "info", { type: "info", text: "three" }),
];
const app = buildApp(loadConfig({ THT_HARNESS_DIR: "/tmp/h" }), {
hub,
thtRunner: {} as any,
@@ -38,17 +44,19 @@ async function captureReplay(options: { query?: string; lastEventId?: string }):
const controller = new AbortController();
try {
const lastEventId = options.lastEventId?.(ids);
const query = options.query ? `?lastEventId=${encodeURIComponent(options.query(ids))}` : "";
const response = await fetch(
`http://127.0.0.1:${port}/sessions/s1/events${options.query ?? ""}`,
`http://127.0.0.1:${port}/sessions/s1/events${query}`,
{
headers: options.lastEventId ? { "Last-Event-ID": options.lastEventId } : undefined,
headers: lastEventId ? { "Last-Event-ID": lastEventId } : undefined,
signal: controller.signal,
},
);
const reader = response.body!.getReader();
const body = await readUntil(reader, (text) => text.includes('"three"'));
await reader.cancel();
return body;
return { body, ids };
} finally {
controller.abort();
await app.close();
@@ -67,36 +75,45 @@ async function expectStreamEnd(reader: ReadableStreamDefaultReader<Uint8Array>):
}
test("SSE emits ids and honors the native Last-Event-ID replay cursor", async () => {
const body = await captureReplay({ lastEventId: "1" });
const { body, ids } = await captureReplay({ lastEventId: (published) => published[0] });
expect(body).not.toContain('"one"');
expect(body).toContain("id: 2\nevent: info\n");
expect(body).toContain(`id: ${ids[1]}\nevent: info\n`);
expect(body).toContain('data: {"type":"info","text":"two"}');
expect(body).toContain("id: 3\nevent: info\n");
expect(body).toContain(`id: ${ids[2]}\nevent: info\n`);
expect(body).toContain('data: {"type":"info","text":"three"}');
});
test("SSE honors the manual lastEventId query cursor", async () => {
const body = await captureReplay({ query: "?lastEventId=2" });
const { body, ids } = await captureReplay({ query: (published) => published[1] });
expect(body).not.toContain('"one"');
expect(body).not.toContain('"two"');
expect(body).toContain("id: 3\nevent: info\n");
expect(body).toContain(`id: ${ids[2]}\nevent: info\n`);
expect(body).toContain('data: {"type":"info","text":"three"}');
});
test("SSE uses the newer valid cursor when header and query are both present", async () => {
const body = await captureReplay({ query: "?lastEventId=2", lastEventId: "1" });
const { body } = await captureReplay({
query: (published) => published[1],
lastEventId: (published) => published[0],
});
expect(body).not.toContain('"two"');
expect(body).toContain("id: 3\nevent: info\n");
expect(body).toContain('"three"');
});
test("SSE resets a cursor from an older process and emits a buffered pending gate once", async () => {
test("SSE resets a cursor from an older backend generation and emits a buffered pending gate once", async () => {
// The cursor comes from a PREVIOUS process: same session, ids restarted. It must be
// treated as stale (replay from the beginning), not honored against the new ids.
const oldGeneration = new SseHub();
oldGeneration.publish("s1", "info", { type: "info", text: "old" });
const staleCursor = oldGeneration.publish("s1", "info", { type: "info", text: "older" });
const hub = new SseHub();
const descriptor = { id: "gate-fresh", widget: "select", title: "Choose" };
hub.publish("s1", "info", { type: "info", text: "fresh generation" });
hub.publish("s1", "ui_request", { type: "ui_request", ui_request: descriptor });
const infoId = hub.publish("s1", "info", { type: "info", text: "fresh generation" });
const gateId = hub.publish("s1", "ui_request", { type: "ui_request", ui_request: descriptor });
const app = buildApp(loadConfig({ THT_HARNESS_DIR: "/tmp/h" }), {
hub,
thtRunner: {} as any,
@@ -110,17 +127,17 @@ test("SSE resets a cursor from an older process and emits a buffered pending gat
try {
const response = await fetch(`http://127.0.0.1:${port}/sessions/s1/events`, {
headers: { "Last-Event-ID": "900" },
headers: { "Last-Event-ID": staleCursor },
signal: controller.signal,
});
const reader = response.body!.getReader();
const body = await readUntil(reader, (text) => text.includes('"gate-fresh"'));
await reader.cancel();
expect(body).toContain("id: 1\nevent: info\n");
expect(body).toContain(`id: ${infoId}\nevent: info\n`);
expect(body).toContain('data: {"type":"info","text":"fresh generation"}');
expect(body.match(/event: ui_request/g)).toHaveLength(1);
expect(body).toContain("id: 2\nevent: ui_request\n");
expect(body).toContain(`id: ${gateId}\nevent: ui_request\n`);
expect(body).toContain('"ui_request":{"id":"gate-fresh","widget":"select","title":"Choose"}');
} finally {
controller.abort();
@@ -145,32 +162,35 @@ test("clear ends every live SSE response and a cursor reconnect replays post-cle
)));
const readers = responses.map((response) => response.body!.getReader());
expect(hub.publish("s1", "info", { type: "info", text: "before" })).toBe(1);
const beforeId = hub.publish("s1", "info", { type: "info", text: "before" });
expect(seqOf(beforeId)).toBe(1);
const initial = await Promise.all(readers.map((reader) =>
readUntil(reader, (text) => text.includes('"before"'))));
expect(initial.every((body) => body.includes("id: 1\nevent: info\n"))).toBe(true);
expect(initial.every((body) => body.includes(`id: ${beforeId}\nevent: info\n`))).toBe(true);
hub.clear("s1");
await Promise.all(readers.map(expectStreamEnd));
expect(hub.publish("s1", "info", { type: "info", text: "after" })).toBe(2);
expect(hub.publish("s1", "ui_request", {
const afterId = hub.publish("s1", "info", { type: "info", text: "after" });
expect(seqOf(afterId)).toBe(2);
const gateId = hub.publish("s1", "ui_request", {
type: "ui_request",
ui_request: { id: "gate-1", widget: "select" },
})).toBe(3);
});
expect(seqOf(gateId)).toBe(3);
const reconnected = await fetch(
`http://127.0.0.1:${port}/sessions/s1/events`,
{
headers: { "Last-Event-ID": "1" },
headers: { "Last-Event-ID": beforeId },
signal: controllers[2].signal,
},
);
const reconnectReader = reconnected.body!.getReader();
const replay = await readUntil(reconnectReader, (text) => text.includes('"gate-1"'));
expect(replay).toContain("id: 2\nevent: info\n");
expect(replay).toContain(`id: ${afterId}\nevent: info\n`);
expect(replay).toContain('data: {"type":"info","text":"after"}');
expect(replay).toContain("id: 3\nevent: ui_request\n");
expect(replay).toContain(`id: ${gateId}\nevent: ui_request\n`);
expect(replay).toContain('"ui_request":{"id":"gate-1","widget":"select"}');
await reconnectReader.cancel();
} finally {