diff --git a/lib/__tests__/jmap-client-resilience.test.ts b/lib/__tests__/jmap-client-resilience.test.ts index cc3eaa17..a213a049 100644 --- a/lib/__tests__/jmap-client-resilience.test.ts +++ b/lib/__tests__/jmap-client-resilience.test.ts @@ -349,6 +349,68 @@ describe('JMAPClient resilience', () => { }); }); + // Every account switch tears down and re-creates push for all connected + // clients. Aborting an SSE connect that is still in flight must read as an + // intentional close: treated as a network failure it spawns an unsupervised + // 3s polling interval per client, and its late rejection nulls the abort + // controller of the connection set up right after - which then can never be + // closed and reconnects itself in parallel. Rapid switching multiplies both + // until the server's concurrency limit stalls the app. + describe('SSE connect aborted mid-flight (account-switch churn)', () => { + function inFlightFetch(signals: AbortSignal[]) { + return (_url: RequestInfo | URL, init?: RequestInit) => { + if (init?.signal) signals.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); + }); + }; + } + + it('does not fall back to polling when the in-flight connect was aborted', async () => { + vi.useFakeTimers({ shouldAdvanceTime: false }); + const client = await createConnectedClient(); + fetchSpy.mockImplementation(inFlightFetch([])); + + client.setupPushNotifications(); + client.closePushNotifications(); + + // Flush the 1s network-error retry inside authenticatedFetch and a few + // would-be polling ticks (3s each). + await vi.advanceTimersByTimeAsync(10_000); + + // The polling fallback is recognizable by its state-poll body; a + // keep-alive Core/echo that slips in must not fail the assertion. + const statePolls = fetchSpy.mock.calls.filter((call: unknown[]) => { + const body = (call[1] as RequestInit | undefined)?.body; + return typeof body === 'string' && body.includes('Mailbox/get'); + }); + expect(statePolls).toHaveLength(0); + }); + + it('keeps the replacement connection abortable when the aborted connect settles late', async () => { + vi.useFakeTimers({ shouldAdvanceTime: false }); + const client = await createConnectedClient(); + const signals: AbortSignal[] = []; + fetchSpy.mockImplementation(inFlightFetch(signals)); + + client.setupPushNotifications(); + client.closePushNotifications(); + client.setupPushNotifications(); + // The retry inside authenticatedFetch re-sends the aborted first + // attempt later, so grab the replacement's signal now. + const replacementSignal = signals[signals.length - 1]; + + // Let the first attempt run through its retry and reject - after the + // replacement connect is already up. + await vi.advanceTimersByTimeAsync(2_000); + + client.closePushNotifications(); + expect(replacementSignal.aborted).toBe(true); + }); + }); + 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 770d28bd..9f64e1ff 100644 --- a/lib/jmap/client.ts +++ b/lib/jmap/client.ts @@ -5914,18 +5914,29 @@ export class JMAPClient implements IJMAPClient { .replace('{closeafter}', 'no') .replace('{ping}', '30'); - this.sseAbortController = new AbortController(); + // Each attempt tracks its own controller. When closePushNotifications + // aborts a connect that is still in flight (every account switch tears + // down and re-creates push for all connected clients), the rejection + // lands in the catch below AFTER the next attempt has already been set + // up - treating it as a network failure there would spawn an + // unsupervised polling interval and, via fallbackToPolling nulling + // sseAbortController, orphan the replacement connection. + const controller = new AbortController(); + this.sseAbortController = controller; this.authenticatedFetch(url, { headers: { 'Accept': 'text/event-stream' }, - signal: this.sseAbortController.signal, + signal: controller.signal, }).then(response => { + if (controller.signal.aborted) return; if (!response.ok || !response.body) { this.fallbackToPolling(); return; } - this.readSSEStream(response.body); + this.readSSEStream(response.body, controller); }).catch((error) => { + // Intentional close, not a failure - no polling fallback. + if (controller.signal.aborted) return; if (error instanceof RateLimitError) { this.sseAbortController = null; this.scheduleSSEReconnect(); @@ -5935,7 +5946,7 @@ export class JMAPClient implements IJMAPClient { }); } - private async readSSEStream(body: ReadableStream): Promise { + private async readSSEStream(body: ReadableStream, controller: AbortController): Promise { const reader = body.getReader(); const decoder = new TextDecoder(); let buffer = ''; @@ -5964,8 +5975,10 @@ export class JMAPClient implements IJMAPClient { this.stopSSEPingMonitor(); - // Stream ended - reconnect unless we were intentionally closed - if (this.sseAbortController && !this.sseAbortController.signal.aborted) { + // Stream ended - reconnect only if this stream is still the current one + // and was not intentionally closed. A superseded stream must not spawn a + // second connection next to its replacement. + if (this.sseAbortController === controller && !controller.signal.aborted) { this.scheduleSSEReconnect(); } }