From f38c82ed0cf41ff2e11cae698e60437c96890bfc Mon Sep 17 00:00:00 2001 From: duyetbot <101855044+duyetbot@users.noreply.github.com> Date: Thu, 17 Sep 2026 19:44:43 +0700 Subject: [PATCH] fix(api): unblock SSE watch backlog replay Return the watch response before awaiting backpressured backlog writes and run replay under ctx.waitUntil. Clean up cancelled streams and cover replay, reconnect cursors, live events, and cancellation with Workers tests. Closes #360 Co-Authored-By: Duyet Le Co-Authored-By: duyetbot --- PLAN.md | 5 + packages/api/src/state-stream-hub.ts | 40 +++-- packages/api/test/state-stream-hub.test.ts | 163 +++++++++++++++++++++ 3 files changed, 197 insertions(+), 11 deletions(-) create mode 100644 packages/api/test/state-stream-hub.test.ts diff --git a/PLAN.md b/PLAN.md index 7dc792fd..dfadb5dc 100644 --- a/PLAN.md +++ b/PLAN.md @@ -2,6 +2,11 @@ This file is the single source of truth for autonomous maintenance. Read by Claude on every `/loop` iteration. +## Targeted Streaming Follow-up + +- [x] #360: return the DO `/watch` response before awaiting backpressured backlog writes; run replay in background and clean up failed/cancelled streams. Real Workers regression tests cover non-empty replay, reconnect cursors, live delivery, and cancellation. +- #366 concurrent replay/live ordering remains separate; draft #361 is not reused. PR CI must pass; do not merge or deploy this fix or touch release-please. + ## Targeted Security Follow-up - [x] #389: persist verified user/org identity bindings, backfill recognized existing tenants without moving data, resolve dashboard tenant before reads/writes, and restore explicit Personal selection. Shared/unknown legacy ownership requires operator audit; see `docs/knowledge/organization-identity.md`. diff --git a/packages/api/src/state-stream-hub.ts b/packages/api/src/state-stream-hub.ts index 391b8c36..c9d10372 100644 --- a/packages/api/src/state-stream-hub.ts +++ b/packages/api/src/state-stream-hub.ts @@ -42,23 +42,41 @@ export class StateStreamHub extends DurableObject { } private async watch(projectId: string, after: number, signal: AbortSignal): Promise { - const stream = new TransformStream(); + let controller!: TransformStreamDefaultController; + const stream = new TransformStream({ + start(streamController) { + controller = streamController; + }, + }); const writer = stream.writable.getWriter(); this.writers.add(writer); - const heartbeat = setInterval(() => { - writer.write(this.encoder.encode("event: ping\ndata: {}\n\n")).catch(() => { - this.writers.delete(writer); - }); - }, 15_000); - - signal.addEventListener("abort", () => { + const cleanup = () => { clearInterval(heartbeat); this.writers.delete(writer); - writer.close().catch(() => {}); - }); + signal.removeEventListener("abort", abort); + }; + const abort = () => { + // Error the readable too, releasing any backpressured replay write. + controller.error(signal.reason); + cleanup(); + }; + const heartbeat = setInterval(() => { + writer.write(this.encoder.encode("event: ping\ndata: {}\n\n")).catch(cleanup); + }, 15_000); - await this.writeBacklog(writer, projectId, after); + signal.addEventListener("abort", abort, { once: true }); + void writer.closed.then(cleanup, cleanup); + if (signal.aborted) abort(); + + // Return the readable before awaiting writes: replay is backpressured until + // the caller can attach a reader to the response (#360). + this.ctx.waitUntil( + this.writeBacklog(writer, projectId, after).catch((error) => { + controller.error(error); + cleanup(); + }), + ); return new Response(stream.readable, { headers: { diff --git a/packages/api/test/state-stream-hub.test.ts b/packages/api/test/state-stream-hub.test.ts new file mode 100644 index 00000000..025d9b76 --- /dev/null +++ b/packages/api/test/state-stream-hub.test.ts @@ -0,0 +1,163 @@ +import { env, SELF } from "cloudflare:test"; +import { beforeAll, beforeEach, describe, expect, it } from "vitest"; +import type { StateEventResponse } from "../src/services/states"; +import { applyMigrations, authHeaders, seedProject, TEST_PROJECT_ID } from "./setup"; + +async function within(promise: Promise): Promise { + let timer: ReturnType; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error("Watch stalled for 3 seconds")), 3_000); + }), + ]); + } finally { + clearTimeout(timer!); + } +} + +async function seedEvent(index: number): Promise { + const event = { + id: crypto.randomUUID(), + state_key: `watch-${index}`, + agent_id: "watch-agent", + event_type: "upsert" as const, + data: { index }, + metadata: null, + tags: ["watch"], + idempotency_key: null, + created_at: Date.now(), + }; + // Seed directly so no delayed /notify races with backlog-only assertions (#366). + const row = await env.DB.prepare( + `INSERT INTO state_events + (id, project_id, state_key, agent_id, event_type, data, tags, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) RETURNING sequence`, + ) + .bind( + event.id, + TEST_PROJECT_ID, + event.state_key, + event.agent_id, + event.event_type, + JSON.stringify(event.data), + JSON.stringify(event.tags), + event.created_at, + ) + .first<{ sequence: number }>(); + return { ...event, sequence: row!.sequence }; +} + +async function openWatch(after = 0) { + // No once=true: this must exercise the real StateStreamHub DO via the API. + const pending = SELF.fetch(`http://localhost/api/v1/states/watch?after=${after}`, { + headers: authHeaders(), + }); + let response: Response; + try { + response = await within(pending); + } catch (error) { + void pending.then((late) => late.body?.cancel()).catch(() => {}); + throw error; + } + const reader = response.body!.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + return { + response, + cancel: () => within(reader.cancel()), + read: () => + within( + (async () => { + while (!buffer.includes("\n\n")) { + const chunk = await reader.read(); + if (chunk.done) throw new Error("Watch closed before the next event"); + buffer += decoder.decode(chunk.value, { stream: true }); + } + const boundary = buffer.indexOf("\n\n"); + const frame = buffer.slice(0, boundary); + buffer = buffer.slice(boundary + 2); + return frame; + })(), + ), + }; +} + +function expectFrame(frameText: string, event: StateEventResponse) { + const lines = frameText.split("\n"); + expect(lines[0]).toBe(`id: ${event.sequence}`); + expect(lines[1]).toBe(`event: state.${event.event_type}`); + expect(lines[2].startsWith("data: ")).toBe(true); + // Compare parsed JSON: key order differs between the replay mapper and a + // broadcast payload, but clients consume JSON semantics. + expect(JSON.parse(lines[2].slice("data: ".length))).toEqual(event); +} + +async function notify(event: StateEventResponse) { + const hub = env.STATE_STREAM_HUB.getByName(TEST_PROJECT_ID); + const response = await within( + hub.fetch("https://state-stream.local/notify", { + method: "POST", + body: JSON.stringify(event), + }), + ); + expect(response.status).toBe(204); +} + +describe("StateStreamHub watch", () => { + beforeAll(applyMigrations); + beforeEach(seedProject); + + it.each([1, 3])("opens and replays a %i-row backlog without hanging", async (count) => { + const events = []; + for (let index = 0; index < count; index++) events.push(await seedEvent(index)); + const watch = await openWatch(); + try { + expect(watch.response.status).toBe(200); + expect(watch.response.headers.get("Content-Type")).toBe("text/event-stream"); + expect(watch.response.headers.get("Cache-Control")).toBe("no-cache"); + for (const event of events) expectFrame(await watch.read(), event); + const live = await seedEvent(count); + await notify(live); + expectFrame(await watch.read(), live); + } finally { + await watch.cancel(); + } + }); + + it("replays only events after the reconnect cursor", async () => { + const first = await seedEvent(0); + const second = await seedEvent(1); + const watch = await openWatch(first.sequence); + try { + expectFrame(await watch.read(), second); + } finally { + await watch.cancel(); + } + }); + + it("opens an empty backlog and receives live notifications", async () => { + const watch = await openWatch(); + try { + expect(watch.response.status).toBe(200); + const live = await seedEvent(0); + await notify(live); + expectFrame(await watch.read(), live); + } finally { + await watch.cancel(); + } + }); + + it("can cancel without consuming the backlog and reconnect", async () => { + const event = await seedEvent(0); + const unread = await openWatch(); + await unread.cancel(); + const watch = await openWatch(); + try { + expectFrame(await watch.read(), event); + } finally { + await watch.cancel(); + } + }); +});