Coalesce the transcript cache write instead of re-serializing it per event
Writing the whole transcript to sessionStorage on every event made ingesting a session quadratic: 1440 records cost 2880 writes and 4.5 GB of string traffic, and the view took 21.7s to fill. A write now waits for the stream to pause (or CACHE_MAX_WAIT_MS, whichever comes first) and for the page going away, and the cursor is written by that same flush, after the records it covers, so a reload can never resume from a cursor the cached records do not hold. 1440 records: 21.7s -> 4.0s, 2880 writes -> 2. 4800 records: 120.3s -> 34.3s, 50 GB -> 21 MB.
46 unmodified lines4748495051525354555657545859606162636465666768697071727374757677789 unmodified lines88899091929377949596979899526 unmodified linesExpand all6266276286296306316326336346356363 unmodified lines640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688201 unmodified linesExpand all890891892893832894895896897898899900143 unmodified linesExpand all46 unmodified lines
(globalThis as any).WebSocket = MockWebSocket;
const mockStorage = new Map<string, string>();/** Writes per key, so a test can assert how OFTEN the cache is re-serialized, not just that it is. */const mockWrites = new Map<string, number>();const pagehideListeners: Array<() => void> = [];
(globalThis as any).window = { sessionStorage: { getItem: (k: string) => mockStorage.get(k) ?? null, setItem: (k: string, v: string) => mockStorage.set(k, v), setItem: (k: string, v: string) => { mockWrites.set(k, (mockWrites.get(k) ?? 0) + 1); mockStorage.set(k, v); }, removeItem: (k: string) => mockStorage.delete(k), clear: () => mockStorage.clear(), }, addEventListener: (type: string, listener: () => void) => { if (type === "pagehide") pagehideListeners.push(listener); },};
/** The page going away is the one cache write that cannot wait for a timer. */function pagehide() { for (const listener of pagehideListeners) listener();}
describe("StreamingClientV2 sequence cursor & lane algebra", () => { let client: StreamingClientV2; let ws: MockWebSocket;
9 unmodified lines test("subscribes on open with initial cursor", () => { expect(ws.sent.length).toBeGreaterThanOrEqual(1); const subMsg = JSON.parse(ws.sent[0]); expect(subMsg.type).toBe("subscribe"); // With nothing cached, the client asks for the durable history from its start rather than for a // live-only subscription. A session with records cached resumes from the newest of them instead. expect(subMsg.after_seq).toBe(null); expect(subMsg.after_seq).toBe(0); expect(get(client.timelineStore).length).toBe(0); });
test("in-order event ingestion updates cursor and stores", () => { ws.emitMessage({526 unmodified linesExpand all created_ms: 1000, }, });
// The cache is written when the stream pauses or the page goes away, not per event. pagehide();
// Simulate quota error on timeline key while cursor key was set to 2 if (typeof window !== "undefined" && window.sessionStorage) { window.sessionStorage.setItem(`${serverId}:${sessionId}:cursor`, "2"); }3 unmodified lines expect(restoredSeqs).toEqual([1]); expect(get(restored.connectionStore).lastSeenSeq).toBe(1); });
test("a burst of events re-serializes the transcript once, not once per event", () => { const sessionId = "coalesce-" + Math.random().toString(36).slice(2, 7); const serverId = "coalesce-server"; const timelineKey = `${serverId}:${sessionId}:timeline`; const cursorKey = `${serverId}:${sessionId}:cursor`; const client = new StreamingClientV2("ws://test", sessionId, serverId); const events = 40;
for (let seq = 1; seq <= events; seq++) { client.handleServerMsg({ type: "event", record: { seq, session_id: sessionId, lane: "input", payload: { source: "operator", text: `burst ${seq}` }, created_ms: 1000 + seq, }, }); }
// Every record is in the view, and the cache has not been re-serialized once per record: ingesting // a session must not cost a full serialize of the session per record. expect(get(client.timelineStore).length).toBe(events); expect(mockWrites.get(timelineKey) ?? 0).toBe(0);
pagehide(); expect(mockWrites.get(timelineKey)).toBe(1); const cached = JSON.parse(mockStorage.get(timelineKey)!); expect(cached.length).toBe(events); expect(cached.map((t: { seq: number }) => t.seq)).toEqual( Array.from({ length: events }, (_, i) => i + 1) ); // Records and the cursor they cover are one write, so the cursor cannot claim more than the cache holds. expect(mockStorage.get(cursorKey)).toBe(String(events));
const reloaded = new StreamingClientV2("ws://test", sessionId, serverId); expect(get(reloaded.timelineStore).length).toBe(events); expect(get(reloaded.connectionStore).lastSeenSeq).toBe(events); });
test("lifecycle diagnostics update healthStore, workItemsStore, and candidateStore while strictly preserving lane separation", () => { const sessionId = "diag-test-" + Math.random().toString(36).slice(2, 7); const client = new StreamingClientV2("ws://test", sessionId);
201 unmodified linesExpand all expect(get(client1.candidateStore)?.candidate_id).toBe("c1"); expect(get(client1.connectionStore).lastSeenSeq).toBe(10); expect(get(client1.speechStore).length).toBe(0);
// Simulate page reload: instantiate new StreamingClientV2 for same sessionId // Simulate page reload: the page goes away (which is what makes the cache durable) and // a new client is instantiated for the same sessionId pagehide(); const reloaded = new StreamingClientV2("ws://test", sessionId);
// Reloaded client must recover diagnostics AND cursor const reloadHealth = get(reloaded.healthStore);143 unmodified linesExpand all137 unmodified linesExpand all13813914014114214314414514614714814915036 unmodified lines187188189190191192193194195196197198199200201153 unmodified linesExpand all355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423271 unmodified linesExpand all695696697698699700701702703152 unmodified linesExpand all856857858859786860861862863864555 unmodified linesExpand all1420142114221423142414251426135214271354135514281429143013571358143114321433143414351436143714381365136614391440144114421443144414451446144717 unmodified lines137 unmodified linesExpand all rejectedCount: number; lastRejection?: { seq: number; reason: string };}
/** How long the stream must be quiet before the transcript is re-serialized into its cache. */const CACHE_IDLE_MS = 400;/** A stream that never goes quiet still gets a cache write this often. */const CACHE_MAX_WAIT_MS = 3000;
export class StreamingClientV2 { private ws: WebSocket | null = null; private url: string; public readonly serverId: string;36 unmodified lines private activeSpeechId: string | null = null; private pendingRequests = new Map<string, { text: string; source: string; timestamp: number }>(); private timelineSaveFailed = false;
// The cache write is the one cost in this client that grows with the session, so it is coalesced // rather than issued per event. See `scheduleCacheWrite`. private pendingCache: { items: TimelineItem[]; seq: number } | null = null; private pendingCacheSince = 0; private cacheTimer: ReturnType<typeof setTimeout> | null = null; private cacheFlushBound = false;
constructor( url: string = "ws://127.0.0.1:8765", sessionId: string = "default", serverId?: string153 unmodified linesExpand all } return false; }
/** * Serializing the transcript is the one cost in this client that grows with the session, and writing it * per event made ingesting a session quadratic: a replay of a few thousand records spent most of its * time re-serializing records it had already cached. So a write is coalesced — one lands once the * stream goes quiet, and a stream that never goes quiet still gets one every `CACHE_MAX_WAIT_MS`. * * The records and the cursor they cover are one record, written together in `flushCacheWrite`. The * cursor is therefore never persisted ahead of the records it claims, which is the ordering the reload * path depends on: `rehydrateFromCache` resumes from `min(savedCursor, newest cached record)`. */ private scheduleCacheWrite(items: TimelineItem[], seq: number) { this.pendingCache = { items, seq }; if (this.pendingCacheSince === 0) this.pendingCacheSince = Date.now(); const waited = Date.now() - this.pendingCacheSince; const delay = Math.max(0, Math.min(CACHE_IDLE_MS, CACHE_MAX_WAIT_MS - waited)); if (this.cacheTimer !== null) clearTimeout(this.cacheTimer); this.cacheTimer = setTimeout(() => this.flushCacheWrite(), delay); this.bindCacheFlush(); }
/** Write what is pending now, records before cursor, and drop it whatever the outcome. */ private flushCacheWrite() { if (this.cacheTimer !== null) { clearTimeout(this.cacheTimer); this.cacheTimer = null; } this.pendingCacheSince = 0; const pending = this.pendingCache; this.pendingCache = null; if (!pending || pending.items.length === 0) return; if (this.saveTimelineCache(pending.items)) { this.saveCursor(pending.seq); } }
/** Drop a pending write without making it: the caller is about to invalidate what it holds. */ private cancelCacheWrite() { if (this.cacheTimer !== null) { clearTimeout(this.cacheTimer); this.cacheTimer = null; } this.pendingCache = null; this.pendingCacheSince = 0; }
/** * A coalesced write has one deadline it cannot miss: the page is going away and there is no next timer. * Bound once, on the first write that is actually pending. */ private bindCacheFlush() { if (this.cacheFlushBound || typeof window === "undefined" || typeof window.addEventListener !== "function") { return; } this.cacheFlushBound = true; const flush = () => this.flushCacheWrite(); window.addEventListener("pagehide", flush); window.addEventListener("visibilitychange", () => { if (typeof document !== "undefined" && document.visibilityState === "hidden") flush(); }); }
private rehydrateFromCache(): boolean { try { if (typeof window !== "undefined" && window.sessionStorage) { const raw = window.sessionStorage.getItem(this.timelineKey);271 unmodified linesExpand all this.saveDiagnosticsSnapshot(); }
public clearSessionState() { this.cancelCacheWrite(); this.lastSeenSeq = 0; this.saveCursor(0); try { if (typeof window !== "undefined" && window.sessionStorage) {152 unmodified linesExpand all }
private setLastSeenSeq(seq: number) { this.lastSeenSeq = seq; this.saveCursor(seq); this.scheduleCacheWrite(get(this.timelineStore), seq); this.updateConnection({ lastSeenSeq: seq }); }
private finishActiveScratchpad() {555 unmodified linesExpand all } }
private updateTimelineItem(item: TimelineItem, prevId?: string) { let next: TimelineItem[] = []; this.timelineStore.update((timeline) => { const idx = timeline.findIndex((t) => t.id === item.id || (prevId !== undefined && t.id === prevId)); let newTimeline: TimelineItem[]; if (idx >= 0) { newTimeline = [...timeline]; newTimeline[idx] = item; next = [...timeline]; next[idx] = item; } else { newTimeline = [...timeline, item]; newTimeline.sort((a, b) => { next = [...timeline, item]; next.sort((a, b) => { if (a.seq !== undefined && b.seq !== undefined) { return a.seq - b.seq; } return a.created_ms - b.created_ms; }); } this.saveTimelineCache(newTimeline); return newTimeline; return next; }); // An append is what makes the cache stale, and it must not be what makes the ingest slow: the // array is handed to the coalescing writer, which serializes it once the stream pauses. this.scheduleCacheWrite(next, this.lastSeenSeq); }
private updateConnection(patch: Partial<ConnectionStatus>) { this.connectionStore.update((prev) => ({ ...prev, ...patch }));17 unmodified lines