diff --git a/src/server/management/account-selection-stream.ts b/src/server/management/account-selection-stream.ts index e04e8a3e14..b51e5653ce 100644 --- a/src/server/management/account-selection-stream.ts +++ b/src/server/management/account-selection-stream.ts @@ -3,6 +3,7 @@ import { registerOptionalShutdownHook } from "../../lib/optional-shutdown-hooks" const MAX_SELECTION_STREAMS = 64; const HEARTBEAT_MS = 15_000; +const STREAM_QUEUE_HIGH_WATER_MARK = 16; const encoder = new TextEncoder(); const connections = new Set<() => void>(); @@ -36,9 +37,17 @@ export function accountSelectionStream(request: Request, validate: () => boolean const send = (frame: string) => { if (closed) return; if (!authorized()) { - // Error clears queued frames as well, so a revoked consumer cannot drain them. - try { controller.error(new DOMException("Management session is no longer authorized", "NotAllowedError")); } - finally { close(); } + // A revoked consumer must not drain frames queued before revocation: error() is + // what discards a non-empty queue. When nothing is queued — the common expired-session + // path, including the heartbeat — close() alone terminates quietly, so an expired + // dashboard session does not dump an expected DOMException into the server console. + // desiredSize equals the high water mark exactly when the queue is empty. + if (controller.desiredSize !== null && controller.desiredSize < STREAM_QUEUE_HIGH_WATER_MARK) { + try { controller.error(new DOMException("Management session is no longer authorized", "NotAllowedError")); } + finally { close(); } + } else { + close(); + } return; } // Reconnection sends a ready event, so a slow reader can reconcile without an @@ -61,7 +70,7 @@ export function accountSelectionStream(request: Request, validate: () => boolean heartbeat.unref?.(); }, cancel() { cleanup(); }, - }, { highWaterMark: 16 }); + }, { highWaterMark: STREAM_QUEUE_HIGH_WATER_MARK }); return new Response(body, { headers: { "Content-Type": "text/event-stream; charset=utf-8", "Cache-Control": "no-cache, no-transform", diff --git a/tests/oauth/oauth-accounts-api.test.ts b/tests/oauth/oauth-accounts-api.test.ts index 75e88d7dbb..c3c6e0c632 100644 --- a/tests/oauth/oauth-accounts-api.test.ts +++ b/tests/oauth/oauth-accounts-api.test.ts @@ -156,7 +156,9 @@ describe("multiauth accounts API", () => { expect(requireManagementAuth(ctx.req, state, ctx.config)).toBeNull(); // Deliberately memoized. const pending = reader.read(); publishAccountSelection("private-provider", "oauth"); - await expect(pending).rejects.toMatchObject({ name: "NotAllowedError" }); + // Nothing was queued before revocation, so the stream closes quietly instead of + // erroring; the pending read resolves done and the post-revocation frame is never sent. + await expect(pending).resolves.toMatchObject({ done: true }); } finally { await reader.cancel().catch(() => undefined); } }); @@ -179,7 +181,31 @@ describe("multiauth accounts API", () => { session.expiresAt = Date.now() - 1; const pending = reader.read(); tick(); - await expect(pending).rejects.toMatchObject({ name: "NotAllowedError" }); + await expect(pending).resolves.toMatchObject({ done: true }); + } finally { + await reader?.cancel().catch(() => undefined); + interval.mockRestore(); + } + }); + + test("selection stream discards frames queued before revocation instead of draining them", async () => { + const { ctx, state, token } = selectionSessionFixture(); + const interval = spyOn(globalThis, "setInterval"); + let reader: ReadableStreamDefaultReader | undefined; + try { + const response = await handleOauthAccountRoutes(ctx); + expect(response?.status).toBe(200); + reader = response!.body!.getReader(); + expect(new TextDecoder().decode((await reader.read()).value)).toContain("event: ready"); + // No pending read: this event stays queued in the controller when the session expires. + publishAccountSelection("queued-provider", "oauth"); + state.sessions.get(token)!.expiresAt = Date.now() - 1; + const tick = interval.mock.calls.find(call => call[1] === 15_000)?.[0]; + if (typeof tick !== "function") throw new Error("selection heartbeat not registered"); + tick(); + // A non-empty queue still takes the error path: the queued frame is discarded and the + // revoked consumer rejects instead of ever draining it. + await expect(reader.read()).rejects.toMatchObject({ name: "NotAllowedError" }); } finally { await reader?.cancel().catch(() => undefined); interval.mockRestore();