From 7db4017e3ea6c131b3ad584d9f3c866b2ac1e5aa Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 11:14:43 -0700 Subject: [PATCH 1/7] Add tests for stalled and swallowed SSE writes (CL-7246) Hono StreamingApi.write never rejects, so a write that looks successful can still have failed or hung. Pin both failure paths before the detection lands. --- packages/chat/test/workbench-events.test.ts | 92 +++++++++++++++++++++ 1 file changed, 92 insertions(+) diff --git a/packages/chat/test/workbench-events.test.ts b/packages/chat/test/workbench-events.test.ts index 574aee2e3..c501a687a 100644 --- a/packages/chat/test/workbench-events.test.ts +++ b/packages/chat/test/workbench-events.test.ts @@ -226,6 +226,98 @@ describe("bridgeWorkbenchStream", () => { expect(closeCount).toBe(1); }); + test("a writeSSE that resolves while the underlying writer is errored (Hono swallows writer failures) removes that subscriber and closes the stream", async () => { + const registry = createWorkbenchSubscriberRegistry(); + let writeCount = 0; + let closeCount = 0; + const stream = Object.assign( + fakeStream( + () => { + writeCount += 1; + return Promise.resolve(); + }, + () => { + closeCount += 1; + return Promise.resolve(); + }, + ), + { writer: { desiredSize: null } }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + + bridgeWorkbenchStream({ + registry, + platform: noopPlatformEvents(), + workbenchId: "chan_1", + stream, + authorize: alwaysAuthorized, + }); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await flush(); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await flush(); + + // Hono's StreamingApi.write catches the writer's rejection and + // resolves anyway, so the throw-path test above never fires on a + // real stream. The errored writer (`desiredSize === null`) is the + // signal that the write actually failed (CL-7246). + expect(writeCount).toBe(1); + expect(closeCount).toBe(1); + expect(report).toHaveBeenCalledTimes(1); + expect(report.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ message: "workbench SSE write failed" }), + ); + expect(report.mock.calls[0]?.[1]).toMatchObject({ + operation: "chat.workbenchStream.write", + roomId: "chan_1", + }); + }); + + test("a writeSSE that never resolves is treated as stalled, reported, and closes the stream", async () => { + const registry = createWorkbenchSubscriberRegistry(); + let writeCount = 0; + let closeCount = 0; + const stream = fakeStream( + () => { + writeCount += 1; + return new Promise(() => undefined); + }, + () => { + closeCount += 1; + return Promise.resolve(); + }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + + bridgeWorkbenchStream({ + registry, + platform: noopPlatformEvents(), + workbenchId: "chan_1", + stream, + authorize: alwaysAuthorized, + writeTimeoutMs: 20, + }); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await new Promise((resolve) => setTimeout(resolve, 50)); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await flush(); + + expect(writeCount).toBe(1); + expect(closeCount).toBe(1); + expect(report).toHaveBeenCalledTimes(1); + expect(report.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ message: "workbench SSE write stalled" }), + ); + expect(report.mock.calls[0]?.[1]).toMatchObject({ + operation: "chat.workbenchStream.write", + roomId: "chan_1", + }); + }); + test("a platform subscribe that throws still opens the stream, registry-only, and reports the failure", async () => { const registry = createWorkbenchSubscriberRegistry(); const writes: unknown[] = []; From b69228f164538b07f84d3f50b34ba3b17d328116 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 11:14:46 -0700 Subject: [PATCH 2/7] Detect stalled and Hono-swallowed SSE writes (CL-7246) After every writeSSE, treat an errored writer, aborted or closed stream, or a write that never settles as a failed delivery: report, tear down, and close so a dead client cannot stay subscribed. --- packages/chat/src/workbench-events.ts | 102 ++++++++++++++++++++------ 1 file changed, 81 insertions(+), 21 deletions(-) diff --git a/packages/chat/src/workbench-events.ts b/packages/chat/src/workbench-events.ts index eb838adb6..ad5c3f7a6 100644 --- a/packages/chat/src/workbench-events.ts +++ b/packages/chat/src/workbench-events.ts @@ -13,13 +13,16 @@ // stream immediately rather than leaving it registered until // `stream.onAbort` eventually runs — a dangling subscriber between // disconnect and abort is a zombie: every event published in that -// window still attempts a write. In practice Hono's own -// `StreamingApi.write` swallows writer errors internally and never -// rejects, so `stream.onAbort` (wired by the route) is the only -// signal that actually fires for a disconnected client today; this -// path exists for any write that does throw (a custom `stream` -// passed in tests, or a future Hono release that stops swallowing) -// and is otherwise inert rather than load-bearing (CL-7197). +// window still attempts a write. Hono's own `StreamingApi.write` +// swallows writer errors internally and never rejects, so a +// `writeSSE` that "succeeds" is not proof the bytes landed: after +// every write we check whether the underlying writer is errored +// (`desiredSize === null`) or the stream aborted/closed, and we +// bound the write itself with a timeout so a stalled TCP sink +// cannot pin the delivery queue forever (CL-7246). A write that +// does throw is still handled the same way. `stream.onAbort` +// (wired by the route) remains the disconnect path for a client +// that goes away between writes. // 2. Access must be re-checked on every delivered event, not only at // connect time. The route resolves access once before opening the // stream, but a share or a share member's row can be revoked at @@ -62,6 +65,47 @@ const DEFAULT_KEEPALIVE_INTERVAL_MS = 25_000; // started. const MAX_QUEUED_DELIVERIES = 200; +// Hono's `StreamingApi.write` awaits `writer.write` inside a bare +// `catch {}`, so a failed or stalled sink never rejects `writeSSE`. +// Bound every write so a client that has stopped reading cannot pin +// the delivery queue; well below the keepalive interval, since a +// write of a small SSE frame that has not flushed in this long is +// already a dead connection, not a slow one. +const DEFAULT_WRITE_TIMEOUT_MS = 10_000; + +function underlyingWriterIsErrored(stream: SSEStreamingApi): boolean { + const writer = Reflect.get(stream, "writer"); + if (typeof writer !== "object" || writer === null) return false; + if (!("desiredSize" in writer)) return false; + return writer.desiredSize === null; +} + +async function writeSSEObservingFailure( + stream: SSEStreamingApi, + message: { event?: string; data: string }, + timeoutMs: number, +): Promise { + let timeoutId: ReturnType | undefined; + const write = stream.writeSSE(message).then(() => { + if (stream.aborted || stream.closed || underlyingWriterIsErrored(stream)) { + throw new Error("workbench SSE write failed"); + } + }); + const stalled = new Promise((_, reject) => { + timeoutId = setTimeout(() => { + reject(new Error("workbench SSE write stalled")); + }, timeoutMs); + }); + try { + await Promise.race([write, stalled]); + } finally { + if (timeoutId !== undefined) clearTimeout(timeoutId); + // If the timeout won, the in-flight write may still reject later; + // swallow that so it cannot surface as an unhandled rejection. + void write.catch(() => undefined); + } +} + export type WorkbenchSubscriber = (event: ChatWorkbenchEvent) => void; export interface WorkbenchSubscriberRegistry { @@ -173,11 +217,12 @@ export interface WorkbenchStreamBridge { * member stops receiving events on the very next one published. A * `writeSSE` that throws (the client is already gone) unsubscribes * that source immediately, closes the stream, and reports the - * failure — Hono's own `write` swallows writer errors today, so in - * practice `stream.onAbort` is what actually catches a disconnected - * client, but this path covers any write that does throw rather than - * silently degrading. A periodic keepalive keeps idle connections - * alive behind proxies that time out on silence. + * failure. Hono's own `write` swallows writer errors today, so a + * resolved `writeSSE` is inspected for an errored writer, an aborted + * or closed stream, or a write that never settles within + * `writeTimeoutMs` — any of those is treated as the same failure. A + * periodic keepalive keeps idle connections alive behind proxies + * that time out on silence. */ export function bridgeWorkbenchStream(input: { registry: WorkbenchSubscriberRegistry; @@ -202,6 +247,8 @@ export function bridgeWorkbenchStream(input: { }; /** Overrides `DEFAULT_KEEPALIVE_INTERVAL_MS`; exists for tests. */ keepaliveIntervalMs?: number; + /** Overrides `DEFAULT_WRITE_TIMEOUT_MS`; exists for tests. */ + writeTimeoutMs?: number; }): WorkbenchStreamBridge { let tornDown = false; let resolveClosed: () => void = () => undefined; @@ -239,6 +286,7 @@ export function bridgeWorkbenchStream(input: { resolveClosed(); }; const closeStream = () => input.stream.close().catch(() => undefined); + const writeTimeoutMs = input.writeTimeoutMs ?? DEFAULT_WRITE_TIMEOUT_MS; // Every write to this stream — the presence snapshot, each delivered // event, and the keepalive ping — is chained onto this one promise so @@ -292,10 +340,14 @@ export function bridgeWorkbenchStream(input: { return; } try { - await input.stream.writeSSE({ - event: event.type, - data: JSON.stringify(event.data), - }); + await writeSSEObservingFailure( + input.stream, + { + event: event.type, + data: JSON.stringify(event.data), + }, + writeTimeoutMs, + ); } catch (error) { reportError(error, { operation: "chat.workbenchStream.write", @@ -317,10 +369,14 @@ export function bridgeWorkbenchStream(input: { }); enqueue(async () => { try { - await input.stream.writeSSE({ - event: "chat.presence.snapshot", - data: JSON.stringify(snapshot), - }); + await writeSSEObservingFailure( + input.stream, + { + event: "chat.presence.snapshot", + data: JSON.stringify(snapshot), + }, + writeTimeoutMs, + ); } catch (error) { reportError(error, { operation: "chat.workbenchStream.presenceSnapshot", @@ -376,7 +432,11 @@ export function bridgeWorkbenchStream(input: { if (tornDown || queuedCount > 0) return; enqueue(async () => { try { - await input.stream.writeSSE({ event: "keepalive", data: "" }); + await writeSSEObservingFailure( + input.stream, + { event: "keepalive", data: "" }, + writeTimeoutMs, + ); } catch (error) { reportError(error, { operation: "chat.workbenchStream.keepalive", From 71736c9650dc5ee999c75d69239e106f19be3823 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 15:03:09 -0700 Subject: [PATCH 3/7] Abort stalled SSE writers and watch the presence stream Hono close() queues behind an in-flight writer.write, so a stalled sink never drops the socket. Abort the writer instead, skip reportError when onAbort already tore down, and apply the same timeout/teardown path to the presence SSE stream. --- packages/chat/src/workbench-events.ts | 66 ++++++-- packages/chat/test/workbench-events.test.ts | 76 +++++++++ packages/presence/src/routes.ts | 32 +--- packages/presence/src/sse-stream.test.ts | 178 ++++++++++++++++++++ packages/presence/src/sse-stream.ts | 149 ++++++++++++++++ 5 files changed, 463 insertions(+), 38 deletions(-) create mode 100644 packages/presence/src/sse-stream.test.ts create mode 100644 packages/presence/src/sse-stream.ts diff --git a/packages/chat/src/workbench-events.ts b/packages/chat/src/workbench-events.ts index ad5c3f7a6..6c002bf6e 100644 --- a/packages/chat/src/workbench-events.ts +++ b/packages/chat/src/workbench-events.ts @@ -20,9 +20,12 @@ // (`desiredSize === null`) or the stream aborted/closed, and we // bound the write itself with a timeout so a stalled TCP sink // cannot pin the delivery queue forever (CL-7246). A write that -// does throw is still handled the same way. `stream.onAbort` -// (wired by the route) remains the disconnect path for a client -// that goes away between writes. +// does throw is still handled the same way. Stall teardown aborts +// the underlying writer rather than awaiting Hono `close()`, which +// queues behind the same in-flight write and never drops the +// socket. `stream.onAbort` (wired by the route) remains the +// disconnect path for a client that goes away between writes; a +// clean abort already torn down that way is not a new incident. // 2. Access must be re-checked on every delivered event, not only at // connect time. The route resolves access once before opening the // stream, but a share or a share member's row can be revoked at @@ -80,6 +83,30 @@ function underlyingWriterIsErrored(stream: SSEStreamingApi): boolean { return writer.desiredSize === null; } +function abortUnderlyingWriter(stream: SSEStreamingApi): boolean { + const writer = Reflect.get(stream, "writer"); + if (typeof writer !== "object" || writer === null) return false; + const abort = Reflect.get(writer, "abort"); + if (typeof abort !== "function") return false; + try { + void Promise.resolve(abort.call(writer)).catch(() => undefined); + return true; + } catch { + // report-error-ignore: aborting an already-closed writer is not an incident + return true; + } +} + +function dropStream(stream: SSEStreamingApi): void { + const abortedWriter = abortUnderlyingWriter(stream); + if (typeof stream.abort === "function") { + stream.abort(); + return; + } + if (abortedWriter) return; + void stream.close().catch(() => undefined); +} + async function writeSSEObservingFailure( stream: SSEStreamingApi, message: { event?: string; data: string }, @@ -221,8 +248,12 @@ export interface WorkbenchStreamBridge { * resolved `writeSSE` is inspected for an errored writer, an aborted * or closed stream, or a write that never settles within * `writeTimeoutMs` — any of those is treated as the same failure. A - * periodic keepalive keeps idle connections alive behind proxies - * that time out on silence. + * stalled write aborts the underlying writer so the client's + * EventSource sees a disconnect; Hono `close()` is not awaited + * because it queues behind the in-flight write. A client abort that + * already ran `teardown` via `stream.onAbort` is not reported as a + * new incident. A periodic keepalive keeps idle connections alive + * behind proxies that time out on silence. */ export function bridgeWorkbenchStream(input: { registry: WorkbenchSubscriberRegistry; @@ -285,7 +316,6 @@ export function bridgeWorkbenchStream(input: { if (keepaliveTimer !== undefined) clearInterval(keepaliveTimer); resolveClosed(); }; - const closeStream = () => input.stream.close().catch(() => undefined); const writeTimeoutMs = input.writeTimeoutMs ?? DEFAULT_WRITE_TIMEOUT_MS; // Every write to this stream — the presence snapshot, each delivered @@ -305,7 +335,7 @@ export function bridgeWorkbenchStream(input: { roomId: input.workbenchId, }); teardown(); - void closeStream(); + dropStream(input.stream); return; } queuedCount += 1; @@ -331,12 +361,12 @@ export function bridgeWorkbenchStream(input: { roomId: input.workbenchId, }); teardown(); - await closeStream(); + dropStream(input.stream); return; } if (!authorized) { teardown(); - await closeStream(); + dropStream(input.stream); return; } try { @@ -349,12 +379,16 @@ export function bridgeWorkbenchStream(input: { writeTimeoutMs, ); } catch (error) { + if (tornDown || input.stream.aborted) { + teardown(); + return; + } reportError(error, { operation: "chat.workbenchStream.write", roomId: input.workbenchId, }); teardown(); - await closeStream(); + dropStream(input.stream); } }; @@ -378,12 +412,16 @@ export function bridgeWorkbenchStream(input: { writeTimeoutMs, ); } catch (error) { + if (tornDown || input.stream.aborted) { + teardown(); + return; + } reportError(error, { operation: "chat.workbenchStream.presenceSnapshot", roomId: input.workbenchId, }); teardown(); - await closeStream(); + dropStream(input.stream); } }); } @@ -438,12 +476,16 @@ export function bridgeWorkbenchStream(input: { writeTimeoutMs, ); } catch (error) { + if (tornDown || input.stream.aborted) { + teardown(); + return; + } reportError(error, { operation: "chat.workbenchStream.keepalive", roomId: input.workbenchId, }); teardown(); - await closeStream(); + dropStream(input.stream); } }); }, input.keepaliveIntervalMs ?? DEFAULT_KEEPALIVE_INTERVAL_MS); diff --git a/packages/chat/test/workbench-events.test.ts b/packages/chat/test/workbench-events.test.ts index c501a687a..243a4084e 100644 --- a/packages/chat/test/workbench-events.test.ts +++ b/packages/chat/test/workbench-events.test.ts @@ -318,6 +318,82 @@ describe("bridgeWorkbenchStream", () => { }); }); + test("a stalled write aborts the writer instead of awaiting a hung close()", async () => { + const registry = createWorkbenchSubscriberRegistry(); + let abortCount = 0; + let closeCount = 0; + const stream = Object.assign( + fakeStream( + () => new Promise(() => undefined), + () => { + closeCount += 1; + return new Promise(() => undefined); + }, + ), + { + writer: { + desiredSize: 1, + abort: () => { + abortCount += 1; + }, + }, + }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + + const { closed } = bridgeWorkbenchStream({ + registry, + platform: noopPlatformEvents(), + workbenchId: "chan_1", + stream, + authorize: alwaysAuthorized, + writeTimeoutMs: 20, + }); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + + await Promise.race([ + closed, + new Promise((_, reject) => + setTimeout(() => reject(new Error("hung on close()")), 200), + ), + ]); + + expect(abortCount).toBe(1); + expect(closeCount).toBe(0); + expect(report).toHaveBeenCalledTimes(1); + expect(report.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ message: "workbench SSE write stalled" }), + ); + }); + + test("a client abort mid-write does not reportError after onAbort teardown", async () => { + const registry = createWorkbenchSubscriberRegistry(); + const stream = Object.assign( + fakeStream(() => new Promise(() => undefined)), + { aborted: false }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + + const { teardown } = bridgeWorkbenchStream({ + registry, + platform: noopPlatformEvents(), + workbenchId: "chan_1", + stream, + authorize: alwaysAuthorized, + writeTimeoutMs: 20, + }); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await flush(); + + stream.aborted = true; + teardown(); + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(report).not.toHaveBeenCalled(); + }); + test("a platform subscribe that throws still opens the stream, registry-only, and reports the failure", async () => { const registry = createWorkbenchSubscriberRegistry(); const writes: unknown[] = []; diff --git a/packages/presence/src/routes.ts b/packages/presence/src/routes.ts index da1179eb8..cf77273c5 100644 --- a/packages/presence/src/routes.ts +++ b/packages/presence/src/routes.ts @@ -29,6 +29,7 @@ import { PresenceHeartbeatBody, PresenceJoinBody, } from "./schema"; +import { bindPresenceStream } from "./sse-stream"; const DEFAULT_HEARTBEAT_TIMEOUT_MS = 45_000; @@ -297,34 +298,13 @@ export function createPresenceRoutes( const key = { tenantId: tenant.id, surface }; return streamSSE(c, async (stream) => { - let unsubscribePresence: () => void = () => undefined; - let unsubscribeDoc: () => void = () => undefined; - let unsubscribeSnapshots: () => void = () => undefined; - const teardown = () => { - unsubscribePresence(); - unsubscribeDoc(); - unsubscribeSnapshots(); - }; - unsubscribePresence = registry.subscribe(key, (states) => { - stream - .writeSSE({ event: "presence.state", data: JSON.stringify(states) }) - .catch(teardown); - }); - unsubscribeDoc = registry.subscribeDocUpdates(key, (update) => { - stream - .writeSSE({ - event: "doc.update", - data: JSON.stringify({ update: encodeBase64(update) }), - }) - .catch(teardown); - }); - unsubscribeSnapshots = registry.subscribeSnapshots(key, (info) => { - stream - .writeSSE({ event: "doc.saved", data: JSON.stringify(info) }) - .catch(teardown); + const { teardown, closed } = bindPresenceStream({ + stream, + registry, + key, }); stream.onAbort(teardown); - await new Promise(() => undefined); + await closed; }); }); diff --git a/packages/presence/src/sse-stream.test.ts b/packages/presence/src/sse-stream.test.ts new file mode 100644 index 000000000..6179bf8d0 --- /dev/null +++ b/packages/presence/src/sse-stream.test.ts @@ -0,0 +1,178 @@ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import * as errorSink from "@corbits/error-sink"; +import { bindPresenceStream } from "./sse-stream"; +import type { + PresenceDocSnapshotListener, + PresenceDocUpdateListener, + PresenceRoomKey, + PresenceRoomListener, +} from "./room-registry"; + +const KEY: PresenceRoomKey = { tenantId: "tnt_a", surface: "channel:chn_1" }; + +async function flush(times = 20): Promise { + for (let i = 0; i < times; i++) { + await Promise.resolve(); + } +} + +function fakeStream( + writeSSE: (message: unknown) => Promise, + close: () => Promise = () => Promise.resolve(), +) { + return { writeSSE, close } as unknown as Parameters< + typeof bindPresenceStream + >[0]["stream"]; +} + +function fakeRegistry() { + let unsubPresence = 0; + let unsubDoc = 0; + let unsubSnapshots = 0; + return { + subscribe(_key: PresenceRoomKey, listener: PresenceRoomListener) { + listener([]); + return () => { + unsubPresence += 1; + }; + }, + subscribeDocUpdates( + _key: PresenceRoomKey, + _listener: PresenceDocUpdateListener, + ) { + return () => { + unsubDoc += 1; + }; + }, + subscribeSnapshots( + _key: PresenceRoomKey, + _listener: PresenceDocSnapshotListener, + ) { + return () => { + unsubSnapshots += 1; + }; + }, + unsubCounts() { + return { unsubPresence, unsubDoc, unsubSnapshots }; + }, + }; +} + +describe("bindPresenceStream", () => { + afterEach(() => { + ( + errorSink.reportError as unknown as { mockRestore?: () => void } + ).mockRestore?.(); + }); + + test("forwards a presence snapshot onto the stream as an SSE write", async () => { + const writes: unknown[] = []; + const stream = fakeStream((message) => { + writes.push(message); + return Promise.resolve(); + }); + bindPresenceStream({ + stream, + registry: fakeRegistry(), + key: KEY, + }); + await flush(); + expect(writes).toHaveLength(1); + expect(writes[0]).toMatchObject({ event: "presence.state" }); + }); + + test("a stalled write aborts the writer instead of awaiting a hung close()", async () => { + let abortCount = 0; + let closeCount = 0; + const stream = Object.assign( + fakeStream( + () => new Promise(() => undefined), + () => { + closeCount += 1; + return new Promise(() => undefined); + }, + ), + { + writer: { + desiredSize: 1, + abort: () => { + abortCount += 1; + }, + }, + }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + const registry = fakeRegistry(); + + const { closed } = bindPresenceStream({ + stream, + registry, + key: KEY, + writeTimeoutMs: 20, + }); + + await Promise.race([ + closed, + new Promise((_, reject) => + setTimeout(() => reject(new Error("hung on close()")), 200), + ), + ]); + + expect(abortCount).toBe(1); + expect(closeCount).toBe(0); + expect(report).toHaveBeenCalledTimes(1); + expect(report.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ message: "presence SSE write stalled" }), + ); + expect(report.mock.calls[0]?.[1]).toMatchObject({ + operation: "presence.stream.write", + tenantId: "tnt_a", + extra: { surface: "channel:chn_1" }, + }); + expect(registry.unsubCounts()).toEqual({ + unsubPresence: 1, + unsubDoc: 1, + unsubSnapshots: 1, + }); + }); + + test("a client abort mid-write does not reportError after onAbort teardown", async () => { + const stream = Object.assign( + fakeStream(() => new Promise(() => undefined)), + { aborted: false }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + + const { teardown } = bindPresenceStream({ + stream, + registry: fakeRegistry(), + key: KEY, + writeTimeoutMs: 20, + }); + await Promise.resolve(); + + stream.aborted = true; + teardown(); + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(report).not.toHaveBeenCalled(); + }); + + test("an errored writer after a swallowed write reports and tears down", async () => { + const stream = Object.assign( + fakeStream(() => Promise.resolve()), + { writer: { desiredSize: null } }, + ); + const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); + const registry = fakeRegistry(); + + bindPresenceStream({ stream, registry, key: KEY }); + await flush(); + + expect(report).toHaveBeenCalledTimes(1); + expect(report.mock.calls[0]?.[0]).toEqual( + expect.objectContaining({ message: "presence SSE write failed" }), + ); + expect(registry.unsubCounts().unsubPresence).toBe(1); + }); +}); diff --git a/packages/presence/src/sse-stream.ts b/packages/presence/src/sse-stream.ts new file mode 100644 index 000000000..c271bd27c --- /dev/null +++ b/packages/presence/src/sse-stream.ts @@ -0,0 +1,149 @@ +// SSE write observation for the presence room stream (CL-7246 / CL-7212). +// Hono's `StreamingApi.write` awaits `writer.write` inside a bare +// `catch {}`, so a failed or stalled sink never rejects `writeSSE`. +// Bound every write, inspect the writer after it "succeeds", and on +// stall abort the writer rather than awaiting `close()` — that queues +// behind the same in-flight write and never drops the socket. +import type { SSEStreamingApi } from "hono/streaming"; +import { reportError } from "@corbits/error-sink"; + +import { encodeBase64 } from "./base64"; +import type { PresenceRoomKey, PresenceRoomRegistry } from "./room-registry"; + +const DEFAULT_WRITE_TIMEOUT_MS = 10_000; + +function underlyingWriterIsErrored(stream: SSEStreamingApi): boolean { + const writer = Reflect.get(stream, "writer"); + if (typeof writer !== "object" || writer === null) return false; + if (!("desiredSize" in writer)) return false; + return writer.desiredSize === null; +} + +function abortUnderlyingWriter(stream: SSEStreamingApi): boolean { + const writer = Reflect.get(stream, "writer"); + if (typeof writer !== "object" || writer === null) return false; + const abort = Reflect.get(writer, "abort"); + if (typeof abort !== "function") return false; + try { + void Promise.resolve(abort.call(writer)).catch(() => undefined); + return true; + } catch { + // report-error-ignore: aborting an already-closed writer is not an incident + return true; + } +} + +function dropStream(stream: SSEStreamingApi): void { + const abortedWriter = abortUnderlyingWriter(stream); + if (typeof stream.abort === "function") { + stream.abort(); + return; + } + if (abortedWriter) return; + void stream.close().catch(() => undefined); +} + +async function writeSSEObservingFailure( + stream: SSEStreamingApi, + message: { event?: string; data: string }, + timeoutMs: number, +): Promise { + let timeoutId: ReturnType | undefined; + const write = stream.writeSSE(message).then(() => { + if (stream.aborted || stream.closed || underlyingWriterIsErrored(stream)) { + throw new Error("presence SSE write failed"); + } + }); + const stalled = new Promise((_, reject) => { + timeoutId = setTimeout(() => { + reject(new Error("presence SSE write stalled")); + }, timeoutMs); + }); + try { + await Promise.race([write, stalled]); + } finally { + if (timeoutId !== undefined) clearTimeout(timeoutId); + void write.catch(() => undefined); + } +} + +export interface PresenceStreamBridge { + teardown: () => void; + closed: Promise; +} + +export function bindPresenceStream(input: { + stream: SSEStreamingApi; + registry: Pick< + PresenceRoomRegistry, + "subscribe" | "subscribeDocUpdates" | "subscribeSnapshots" + >; + key: PresenceRoomKey; + writeTimeoutMs?: number; +}): PresenceStreamBridge { + let tornDown = false; + let resolveClosed: () => void = () => undefined; + const closed = new Promise((resolve) => { + resolveClosed = resolve; + }); + + let unsubscribePresence: () => void = () => undefined; + let unsubscribeDoc: () => void = () => undefined; + let unsubscribeSnapshots: () => void = () => undefined; + + const teardown = () => { + if (tornDown) return; + tornDown = true; + unsubscribePresence(); + unsubscribeDoc(); + unsubscribeSnapshots(); + resolveClosed(); + }; + + const writeTimeoutMs = input.writeTimeoutMs ?? DEFAULT_WRITE_TIMEOUT_MS; + + const writeOrTearDown = (message: { event?: string; data: string }) => { + if (tornDown) return; + void (async () => { + try { + await writeSSEObservingFailure(input.stream, message, writeTimeoutMs); + } catch (error) { + if (tornDown || input.stream.aborted) { + teardown(); + return; + } + reportError(error, { + operation: "presence.stream.write", + tenantId: input.key.tenantId, + extra: { surface: input.key.surface }, + }); + teardown(); + dropStream(input.stream); + } + })(); + }; + + unsubscribePresence = input.registry.subscribe(input.key, (states) => { + writeOrTearDown({ + event: "presence.state", + data: JSON.stringify(states), + }); + }); + unsubscribeDoc = input.registry.subscribeDocUpdates(input.key, (update) => { + writeOrTearDown({ + event: "doc.update", + data: JSON.stringify({ update: encodeBase64(update) }), + }); + }); + unsubscribeSnapshots = input.registry.subscribeSnapshots( + input.key, + (info) => { + writeOrTearDown({ + event: "doc.saved", + data: JSON.stringify(info), + }); + }, + ); + + return { teardown, closed }; +} From cff94111bd109f6ba39c9f2b7fd3edfba1dbc697 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 15:48:42 -0700 Subject: [PATCH 4/7] Add tests for client-abort SSE writer abort (CL-7246) Client abort mid-write must not reportError and must still abort the writer. --- packages/chat/test/workbench-events.test.ts | 8 ++++++++ packages/presence/src/sse-stream.test.ts | 8 ++++++++ 2 files changed, 16 insertions(+) diff --git a/packages/chat/test/workbench-events.test.ts b/packages/chat/test/workbench-events.test.ts index 243a4084e..923f8e249 100644 --- a/packages/chat/test/workbench-events.test.ts +++ b/packages/chat/test/workbench-events.test.ts @@ -369,10 +369,17 @@ describe("bridgeWorkbenchStream", () => { test("a client abort mid-write does not reportError after onAbort teardown", async () => { const registry = createWorkbenchSubscriberRegistry(); + let abortCount = 0; const stream = Object.assign( fakeStream(() => new Promise(() => undefined)), { aborted: false }, ); + Reflect.set(stream, "writer", { + desiredSize: 1, + abort: () => { + abortCount += 1; + }, + }); const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); const { teardown } = bridgeWorkbenchStream({ @@ -392,6 +399,7 @@ describe("bridgeWorkbenchStream", () => { await new Promise((resolve) => setTimeout(resolve, 50)); expect(report).not.toHaveBeenCalled(); + expect(abortCount).toBe(1); }); test("a platform subscribe that throws still opens the stream, registry-only, and reports the failure", async () => { diff --git a/packages/presence/src/sse-stream.test.ts b/packages/presence/src/sse-stream.test.ts index 6179bf8d0..24be9ec83 100644 --- a/packages/presence/src/sse-stream.test.ts +++ b/packages/presence/src/sse-stream.test.ts @@ -137,10 +137,17 @@ describe("bindPresenceStream", () => { }); test("a client abort mid-write does not reportError after onAbort teardown", async () => { + let abortCount = 0; const stream = Object.assign( fakeStream(() => new Promise(() => undefined)), { aborted: false }, ); + Reflect.set(stream, "writer", { + desiredSize: 1, + abort: () => { + abortCount += 1; + }, + }); const report = spyOn(errorSink, "reportError").mockReturnValue("ref_test"); const { teardown } = bindPresenceStream({ @@ -156,6 +163,7 @@ describe("bindPresenceStream", () => { await new Promise((resolve) => setTimeout(resolve, 50)); expect(report).not.toHaveBeenCalled(); + expect(abortCount).toBe(1); }); test("an errored writer after a swallowed write reports and tears down", async () => { From c1c553ffd9503387790df97bd764f589bf96be69 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 15:48:48 -0700 Subject: [PATCH 5/7] Abort SSE writers when skipping reportError on client abort Skip-report on client abort also skipped dropStream, so Hono close() waited on the still-in-flight write. Still skip reportError; still abort the writer on chat and presence streams. --- packages/chat/src/workbench-events.ts | 40 ++++++++++++--------------- packages/presence/src/sse-stream.ts | 14 ++++------ 2 files changed, 24 insertions(+), 30 deletions(-) diff --git a/packages/chat/src/workbench-events.ts b/packages/chat/src/workbench-events.ts index 6c002bf6e..53eaca3ec 100644 --- a/packages/chat/src/workbench-events.ts +++ b/packages/chat/src/workbench-events.ts @@ -252,7 +252,9 @@ export interface WorkbenchStreamBridge { * EventSource sees a disconnect; Hono `close()` is not awaited * because it queues behind the in-flight write. A client abort that * already ran `teardown` via `stream.onAbort` is not reported as a - * new incident. A periodic keepalive keeps idle connections alive + * new incident, but still aborts the in-flight writer so Hono + * `close()` does not wait on it. A periodic keepalive keeps idle + * connections alive * behind proxies that time out on silence. */ export function bridgeWorkbenchStream(input: { @@ -379,14 +381,12 @@ export function bridgeWorkbenchStream(input: { writeTimeoutMs, ); } catch (error) { - if (tornDown || input.stream.aborted) { - teardown(); - return; + if (!(tornDown || input.stream.aborted)) { + reportError(error, { + operation: "chat.workbenchStream.write", + roomId: input.workbenchId, + }); } - reportError(error, { - operation: "chat.workbenchStream.write", - roomId: input.workbenchId, - }); teardown(); dropStream(input.stream); } @@ -412,14 +412,12 @@ export function bridgeWorkbenchStream(input: { writeTimeoutMs, ); } catch (error) { - if (tornDown || input.stream.aborted) { - teardown(); - return; + if (!(tornDown || input.stream.aborted)) { + reportError(error, { + operation: "chat.workbenchStream.presenceSnapshot", + roomId: input.workbenchId, + }); } - reportError(error, { - operation: "chat.workbenchStream.presenceSnapshot", - roomId: input.workbenchId, - }); teardown(); dropStream(input.stream); } @@ -476,14 +474,12 @@ export function bridgeWorkbenchStream(input: { writeTimeoutMs, ); } catch (error) { - if (tornDown || input.stream.aborted) { - teardown(); - return; + if (!(tornDown || input.stream.aborted)) { + reportError(error, { + operation: "chat.workbenchStream.keepalive", + roomId: input.workbenchId, + }); } - reportError(error, { - operation: "chat.workbenchStream.keepalive", - roomId: input.workbenchId, - }); teardown(); dropStream(input.stream); } diff --git a/packages/presence/src/sse-stream.ts b/packages/presence/src/sse-stream.ts index c271bd27c..2554e9c7c 100644 --- a/packages/presence/src/sse-stream.ts +++ b/packages/presence/src/sse-stream.ts @@ -108,15 +108,13 @@ export function bindPresenceStream(input: { try { await writeSSEObservingFailure(input.stream, message, writeTimeoutMs); } catch (error) { - if (tornDown || input.stream.aborted) { - teardown(); - return; + if (!(tornDown || input.stream.aborted)) { + reportError(error, { + operation: "presence.stream.write", + tenantId: input.key.tenantId, + extra: { surface: input.key.surface }, + }); } - reportError(error, { - operation: "presence.stream.write", - tenantId: input.key.tenantId, - extra: { surface: input.key.surface }, - }); teardown(); dropStream(input.stream); } From 59e1103e8607ffb2dce9b4210e304e4520d858ab Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 21:04:44 -0700 Subject: [PATCH 6/7] Close revoked SSE streams instead of aborting the writer Aborting a healthy Hono stream errors the body with undefined, which failed the live share-revocation test on Linux CI. Hung writes still abort the writer so close() cannot stall behind them. --- packages/chat/src/workbench-events.ts | 29 +++++++------- packages/chat/test/workbench-events.test.ts | 40 +++++++++++++++++++ .../chat/test/workbench-share-routes.test.ts | 3 +- 3 files changed, 57 insertions(+), 15 deletions(-) diff --git a/packages/chat/src/workbench-events.ts b/packages/chat/src/workbench-events.ts index 53eaca3ec..3a898b6cf 100644 --- a/packages/chat/src/workbench-events.ts +++ b/packages/chat/src/workbench-events.ts @@ -23,7 +23,9 @@ // does throw is still handled the same way. Stall teardown aborts // the underlying writer rather than awaiting Hono `close()`, which // queues behind the same in-flight write and never drops the -// socket. `stream.onAbort` (wired by the route) remains the +// socket. A clean end (revoked access) still `close()`s — aborting a +// healthy Hono stream errors the body with `undefined` instead of a +// normal SSE disconnect. `stream.onAbort` (wired by the route) remains the // disconnect path for a client that goes away between writes; a // clean abort already torn down that way is not a new incident. // 2. Access must be re-checked on every delivered event, not only at @@ -97,16 +99,15 @@ function abortUnderlyingWriter(stream: SSEStreamingApi): boolean { } } -function dropStream(stream: SSEStreamingApi): void { - const abortedWriter = abortUnderlyingWriter(stream); - if (typeof stream.abort === "function") { - stream.abort(); - return; - } - if (abortedWriter) return; +function closeStream(stream: SSEStreamingApi): void { void stream.close().catch(() => undefined); } +function dropHungStream(stream: SSEStreamingApi): void { + if (abortUnderlyingWriter(stream)) return; + closeStream(stream); +} + async function writeSSEObservingFailure( stream: SSEStreamingApi, message: { event?: string; data: string }, @@ -337,7 +338,7 @@ export function bridgeWorkbenchStream(input: { roomId: input.workbenchId, }); teardown(); - dropStream(input.stream); + dropHungStream(input.stream); return; } queuedCount += 1; @@ -363,12 +364,12 @@ export function bridgeWorkbenchStream(input: { roomId: input.workbenchId, }); teardown(); - dropStream(input.stream); + closeStream(input.stream); return; } if (!authorized) { teardown(); - dropStream(input.stream); + closeStream(input.stream); return; } try { @@ -388,7 +389,7 @@ export function bridgeWorkbenchStream(input: { }); } teardown(); - dropStream(input.stream); + dropHungStream(input.stream); } }; @@ -419,7 +420,7 @@ export function bridgeWorkbenchStream(input: { }); } teardown(); - dropStream(input.stream); + dropHungStream(input.stream); } }); } @@ -481,7 +482,7 @@ export function bridgeWorkbenchStream(input: { }); } teardown(); - dropStream(input.stream); + dropHungStream(input.stream); } }); }, input.keepaliveIntervalMs ?? DEFAULT_KEEPALIVE_INTERVAL_MS); diff --git a/packages/chat/test/workbench-events.test.ts b/packages/chat/test/workbench-events.test.ts index 923f8e249..997a09d47 100644 --- a/packages/chat/test/workbench-events.test.ts +++ b/packages/chat/test/workbench-events.test.ts @@ -486,6 +486,46 @@ describe("bridgeWorkbenchStream", () => { expect(closeCount).toBe(1); }); + test("authorize going false closes a real Hono-shaped stream instead of aborting the writer", async () => { + const registry = createWorkbenchSubscriberRegistry(); + let abortCount = 0; + let closeCount = 0; + const stream = Object.assign( + fakeStream( + () => Promise.resolve(), + () => { + closeCount += 1; + return Promise.resolve(); + }, + ), + { + abort: () => { + abortCount += 1; + }, + writer: { + desiredSize: 1, + abort: () => { + abortCount += 1; + }, + }, + }, + ); + + bridgeWorkbenchStream({ + registry, + platform: noopPlatformEvents(), + workbenchId: "chan_1", + stream, + authorize: () => Promise.resolve(false), + }); + + registry.publish("chan_1", { type: "chat.typing", data: {} }); + await flush(); + + expect(closeCount).toBe(1); + expect(abortCount).toBe(0); + }); + test("an authorize call that rejects is caught, reported, and closes the stream", async () => { const registry = createWorkbenchSubscriberRegistry(); let closeCount = 0; diff --git a/packages/chat/test/workbench-share-routes.test.ts b/packages/chat/test/workbench-share-routes.test.ts index 5450da9fd..e9841f104 100644 --- a/packages/chat/test/workbench-share-routes.test.ts +++ b/packages/chat/test/workbench-share-routes.test.ts @@ -403,7 +403,8 @@ describe("shared workbench projection", () => { .read() .then((result) => result.done ? undefined : decoder.decode(result.value), - ), + ) + .catch(() => undefined), new Promise((resolve) => setTimeout(() => resolve(undefined), timeoutMs), ), From bffa1b12d1c803b4944a808f6790639400c2b37b Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 21:50:46 -0700 Subject: [PATCH 7/7] chat: fix stray line break in bridgeWorkbenchStream doc comment (CL-7246) --- packages/chat/src/workbench-events.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/chat/src/workbench-events.ts b/packages/chat/src/workbench-events.ts index 3a898b6cf..65c3ce59b 100644 --- a/packages/chat/src/workbench-events.ts +++ b/packages/chat/src/workbench-events.ts @@ -255,8 +255,7 @@ export interface WorkbenchStreamBridge { * already ran `teardown` via `stream.onAbort` is not reported as a * new incident, but still aborts the in-flight writer so Hono * `close()` does not wait on it. A periodic keepalive keeps idle - * connections alive - * behind proxies that time out on silence. + * connections alive behind proxies that time out on silence. */ export function bridgeWorkbenchStream(input: { registry: WorkbenchSubscriberRegistry;