fix(server): close expired dashboard selection streams quietly when no frames are queued

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",

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<Uint8Array> | 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();