From 55ef7803424fbc85f0d426d57ff8e5dac7b4627d Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Tue, 25 Aug 2026 20:03:29 +0200 Subject: [PATCH] fix: cap live SSE streams per tab so many logins can't starve JMAP requests #702 --- components/mail/mail-app.tsx | 9 +- lib/__tests__/jmap-client-resilience.test.ts | 124 +++++++++++++++++++ lib/jmap/client.ts | 96 ++++++++++++++ 3 files changed, 228 insertions(+), 1 deletion(-) diff --git a/components/mail/mail-app.tsx b/components/mail/mail-app.tsx index 11ba738eb..da2069dc4 100644 --- a/components/mail/mail-app.tsx +++ b/components/mail/mail-app.tsx @@ -1201,7 +1201,14 @@ export function MailApp({ linkSegments }: MailAppProps = {}) { const clients = useAuthStore.getState().getAllConnectedClients(); const cleanups: Array<() => void> = []; - for (const [accId, c] of clients) { + // Active login first: the client caps live SSE streams per tab (#702) and + // hands out slots in setup order, so the account the user is looking at + // must claim one before the background logins do. + const ordered = [...clients.entries()].sort(([a], [b]) => + (a === activeAccountId ? 0 : 1) - (b === activeAccountId ? 0 : 1), + ); + + for (const [accId, c] of ordered) { try { if (accId === activeAccountId) { c.onStateChange((change) => handleStateChange(change, c)); diff --git a/lib/__tests__/jmap-client-resilience.test.ts b/lib/__tests__/jmap-client-resilience.test.ts index 18fa5370f..9daf7da9c 100644 --- a/lib/__tests__/jmap-client-resilience.test.ts +++ b/lib/__tests__/jmap-client-resilience.test.ts @@ -573,6 +573,130 @@ describe('JMAPClient resilience', () => { }); }); + // #702: one logged-in account = one JMAPClient = one SSE stream held open as + // a long-lived fetch, i.e. one pinned HTTP/1.1 socket. Browsers allow 6 per + // host, so from the sixth login on no socket is left for ordinary JMAP + // POSTs - sends grey out and never finish, loads and moves take minutes. + // The client must cap concurrent streams per tab and slow-poll the rest. + describe('SSE stream budget across logins (#702)', () => { + const isSSE = (init?: RequestInit) => + (init?.headers as Record | undefined)?.['Accept'] === 'text/event-stream'; + + function trackingFetch(sse: AbortSignal[]) { + return (_url: RequestInfo | URL, init?: RequestInit) => { + if (isSSE(init)) { + if (init?.signal) sse.push(init.signal); + return new Promise((_resolve, reject) => { + const abort = () => reject(new DOMException('The operation was aborted.', 'AbortError')); + if (init?.signal?.aborted) return abort(); + init?.signal?.addEventListener('abort', abort); + }); + } + return Promise.resolve(mockFetchResponse(200, { + methodResponses: [ + ['Mailbox/get', { state: 'm1', list: [] }, 'mbx:acct-1'], + ['Email/get', { state: 'e1', list: [] }, 'eml:acct-1'], + ], + })); + }; + } + + const statePollCount = () => + fetchSpy.mock.calls.filter((call: unknown[]) => { + const body = (call[1] as RequestInit | undefined)?.body; + return typeof body === 'string' && body.includes('Mailbox/get'); + }).length; + + async function createClients(n: number): Promise { + const clients: JMAPClient[] = []; + for (let i = 0; i < n; i++) clients.push(await createConnectedClient()); + return clients; + } + + it('never holds more than MAX_SSE_STREAMS streams open, however many logins there are', async () => { + vi.useFakeTimers({ shouldAdvanceTime: false }); + const clients = await createClients(7); + const sse: AbortSignal[] = []; + fetchSpy.mockImplementation(trackingFetch(sse)); + + for (const c of clients) c.setupPushNotifications(); + await vi.advanceTimersByTimeAsync(0); + + expect(sse.filter((s) => !s.aborted)).toHaveLength(JMAPClient.MAX_SSE_STREAMS); + expect(JMAPClient.activeSSEStreamCount()).toBe(JMAPClient.MAX_SSE_STREAMS); + // Setup order decides who gets a slot - the caller puts the active login first. + expect(clients[0].hasSSEStream()).toBe(true); + expect(clients[1].hasSSEStream()).toBe(true); + expect(clients[6].hasSSEStream()).toBe(false); + + for (const c of clients) c.closePushNotifications(); + await vi.advanceTimersByTimeAsync(0); + expect(JMAPClient.activeSSEStreamCount()).toBe(0); + }); + + it('slow-polls the logins that did not get a stream so their counters still refresh', async () => { + vi.useFakeTimers({ shouldAdvanceTime: false }); + const clients = await createClients(3); + const sse: AbortSignal[] = []; + fetchSpy.mockImplementation(trackingFetch(sse)); + + for (const c of clients) c.setupPushNotifications(); + await vi.advanceTimersByTimeAsync(0); + // The budget-denied client primes its baseline immediately... + expect(statePollCount()).toBe(1); + + fetchSpy.mockClear(); + await vi.advanceTimersByTimeAsync(20_000); + // ...and polls on the slow 20s cadence, not the 3s outage fallback. + expect(statePollCount()).toBe(1); + + for (const c of clients) c.closePushNotifications(); + }); + + it('promotes a waiting login into the slot a closed stream releases', async () => { + vi.useFakeTimers({ shouldAdvanceTime: false }); + const clients = await createClients(3); + const sse: AbortSignal[] = []; + fetchSpy.mockImplementation(trackingFetch(sse)); + + for (const c of clients) c.setupPushNotifications(); + await vi.advanceTimersByTimeAsync(0); + expect(clients[2].hasSSEStream()).toBe(false); + + clients[0].closePushNotifications(); + await vi.advanceTimersByTimeAsync(0); + + expect(clients[0].hasSSEStream()).toBe(false); + expect(clients[2].hasSSEStream()).toBe(true); + expect(sse.filter((s) => !s.aborted)).toHaveLength(JMAPClient.MAX_SSE_STREAMS); + + for (const c of clients) c.closePushNotifications(); + }); + + it('does not promote a waiter that was closed in the same teardown (account-switch churn)', async () => { + vi.useFakeTimers({ shouldAdvanceTime: false }); + const clients = await createClients(3); + const sse: AbortSignal[] = []; + fetchSpy.mockImplementation(trackingFetch(sse)); + + for (const c of clients) c.setupPushNotifications(); + await vi.advanceTimersByTimeAsync(0); + + // The push effect closes every client and re-runs setup synchronously. + for (const c of clients) c.closePushNotifications(); + for (const c of [clients[2], clients[1], clients[0]]) c.setupPushNotifications(); + await vi.advanceTimersByTimeAsync(0); + + expect(JMAPClient.activeSSEStreamCount()).toBe(JMAPClient.MAX_SSE_STREAMS); + expect(clients[2].hasSSEStream()).toBe(true); + expect(clients[1].hasSSEStream()).toBe(true); + expect(clients[0].hasSSEStream()).toBe(false); + expect(sse.filter((s) => !s.aborted)).toHaveLength(JMAPClient.MAX_SSE_STREAMS); + + for (const c of clients) c.closePushNotifications(); + }); + }); + describe('fetchBlobAsObjectUrl', () => { it('fetches blob with authentication and returns an object URL', async () => { const client = await createConnectedClient(); diff --git a/lib/jmap/client.ts b/lib/jmap/client.ts index 58c190b79..35e693f37 100644 --- a/lib/jmap/client.ts +++ b/lib/jmap/client.ts @@ -6675,6 +6675,78 @@ export class JMAPClient implements IJMAPClient { private visibilityHandler: (() => void) | null = null; private onlineHandler: (() => void) | null = null; + // Per-tab SSE budget (#702). Every logged-in account is its own JMAPClient + // and each one holds its push stream open as a long-lived fetch - one pinned + // socket per account. Browsers allow only 6 HTTP/1.1 connections per host, + // so from the sixth login on there is no socket left for an ordinary JMAP + // POST: sends grey out and never complete, moves and loads queue for + // minutes, all until one of the streams happens to drop. Cap the number of + // concurrent streams per tab well below that limit (two tabs of the app + // share the same pool) and let the remaining logins fall back to the slow + // state poll, which borrows a socket for a moment instead of pinning it. + // Waiters are promoted in order as holders release their slot. + static readonly MAX_SSE_STREAMS = 2; + private static sseHolders = new Set(); + private static sseWaiters: JMAPClient[] = []; + private static ssePromotionQueued = false; + // True between connectSSE and closePushNotifications: this client wants a + // live stream and may be promoted into one when a slot frees up. + private sseWanted = false; + + private static acquireSSESlot(client: JMAPClient): boolean { + if (JMAPClient.sseHolders.has(client)) return true; + if (JMAPClient.sseHolders.size >= JMAPClient.MAX_SSE_STREAMS) return false; + JMAPClient.sseHolders.add(client); + JMAPClient.sseWaiters = JMAPClient.sseWaiters.filter((c) => c !== client); + return true; + } + + private static releaseSSESlot(client: JMAPClient): void { + JMAPClient.sseWaiters = JMAPClient.sseWaiters.filter((c) => c !== client); + if (!JMAPClient.sseHolders.delete(client)) return; + // Promote on a microtask, not inline: the push effect tears down every + // client and immediately sets them all up again (account switch), and a + // synchronous promotion would open a stream for a waiter that is about to + // be closed a few statements later. + if (JMAPClient.ssePromotionQueued) return; + JMAPClient.ssePromotionQueued = true; + queueMicrotask(() => { + JMAPClient.ssePromotionQueued = false; + JMAPClient.promoteSSEWaiters(); + }); + } + + private static enqueueSSEWaiter(client: JMAPClient): void { + if (!JMAPClient.sseWaiters.includes(client)) JMAPClient.sseWaiters.push(client); + } + + private static promoteSSEWaiters(): void { + while (JMAPClient.sseHolders.size < JMAPClient.MAX_SSE_STREAMS && JMAPClient.sseWaiters.length > 0) { + const next = JMAPClient.sseWaiters.shift()!; + if (!next.sseWanted || next.intentionallyDisconnected) continue; + const url = next.getEventSourceUrl(); + if (!url) continue; + // Leave the slow poll running only if it is still needed for shared + // accounts; the SSE branch of setup re-creates it in that case. + if (next.secondaryPollInterval) { + clearInterval(next.secondaryPollInterval); + next.secondaryPollInterval = null; + } + next.connectSSE(url); + next.startSecondaryAccountPoll(); + } + } + + /** Number of SSE streams currently held open by clients in this tab. */ + static activeSSEStreamCount(): number { + return JMAPClient.sseHolders.size; + } + + /** Whether this client holds one of the tab's SSE slots (vs. slow-polling). */ + hasSSEStream(): boolean { + return JMAPClient.sseHolders.has(this); + } + private static readonly STATE_TYPE_MAP: Record = { 'Mailbox/get': 'Mailbox', 'Email/get': 'Email', @@ -6715,6 +6787,16 @@ export class JMAPClient implements IJMAPClient { if (this.secondaryPollInterval) return; const hasSecondary = this.pollAccountIds().some((id) => id !== this.accountId); if (!hasSecondary) return; + this.startSlowPoll(); + } + + /** + * The slow (20s) state poll over every account this login can see. Used + * alongside SSE for shared accounts, and on its own for logins that did not + * get one of the tab's SSE slots (#702) - it covers the primary account too. + */ + private startSlowPoll(): void { + if (this.secondaryPollInterval) return; // Prime the per-account baseline so the first tick doesn't false-fire. void this.fetchCurrentStates(); this.secondaryPollInterval = setInterval(() => { @@ -6724,11 +6806,21 @@ export class JMAPClient implements IJMAPClient { } private connectSSE(templateUrl: string): void { + this.sseWanted = true; if (this.isRateLimited()) { this.scheduleSSEReconnect(); return; } + // Budget check (#702): with no slot left in this tab, keep this login on + // the slow poll and queue it for promotion once a holder releases. It + // must not pin a sixth socket and starve every other request. + if (!JMAPClient.acquireSSESlot(this)) { + JMAPClient.enqueueSSEWaiter(this); + this.startSlowPoll(); + return; + } + const url = templateUrl .replace('{types}', '*') .replace('{closeafter}', 'no') @@ -6865,6 +6957,8 @@ export class JMAPClient implements IJMAPClient { private fallbackToPolling(): void { this.sseAbortController = null; + // No stream is held any more - hand the slot to a waiting login. + JMAPClient.releaseSSESlot(this); if (!this.pollingInterval) { this.startPollingFallback(); } @@ -7017,6 +7111,8 @@ export class JMAPClient implements IJMAPClient { } closePushNotifications(): void { + this.sseWanted = false; + JMAPClient.releaseSSESlot(this); if (this.pollingInterval) { clearInterval(this.pollingInterval); this.pollingInterval = null;