From 26e8c5797f516efb52b7930c697bba3d554f6e77 Mon Sep 17 00:00:00 2001 From: Stefan Hildebrandt <695494+hildebrandttk@users.noreply.github.com> Date: Thu, 13 Aug 2026 23:26:57 +0200 Subject: [PATCH] fix(jmap): keep exactly one SSE stream per client and coalesce state polls connectSSE() overwrote sseAbortController with a new controller without aborting the previous one, so re-running setupPushNotifications (which happens per connected account and on fairly frequent effect changes) orphaned SSE streams. readSSEStream only unwinds on abort/done, so each orphan held its HTTP/1.1 socket forever; with multiple accounts this starved normal JMAP POSTs, which queued and timed out (~30s). - connectSSE: abort any prior stream (+ its ping monitor / pending reconnect) before opening a new one -> exactly one live SSE per client. - startPollingFallback: bail if a polling interval already exists. - checkForStateChanges: skip while a poll is already in flight (stateCheckInFlight) so overlapping triggers don't stack POSTs. Closes #781. --- lib/__tests__/jmap-client-resilience.test.ts | 24 +++++++++++++ lib/jmap/client.ts | 36 ++++++++++++++++++++ 2 files changed, 60 insertions(+) diff --git a/lib/__tests__/jmap-client-resilience.test.ts b/lib/__tests__/jmap-client-resilience.test.ts index caf674320..18fa5370f 100644 --- a/lib/__tests__/jmap-client-resilience.test.ts +++ b/lib/__tests__/jmap-client-resilience.test.ts @@ -547,6 +547,30 @@ describe('JMAPClient resilience', () => { client.closePushNotifications(); expect(replacementSignal.aborted).toBe(true); }); + + // #781: setupPushNotifications is re-run for every connected client on + // fairly frequent effect deps changes. Without an already-active guard a + // second connect stacks a new SSE stream while the previous one keeps its + // HTTP/1.1 socket open forever (readSSEStream only unwinds on abort/done), + // starving normal JMAP POSTs. Opening a new stream must abort the old one. + it('aborts the previous SSE stream when push is set up again (no orphan)', async () => { + vi.useFakeTimers({ shouldAdvanceTime: false }); + const client = await createConnectedClient(); + const signals: AbortSignal[] = []; + fetchSpy.mockImplementation(inFlightFetch(signals)); + + client.setupPushNotifications(); + await vi.advanceTimersByTimeAsync(0); // let connect #1 register its signal + client.setupPushNotifications(); + await vi.advanceTimersByTimeAsync(0); // connect #2 registers + guard aborts #1 + + // The first stream's signal is aborted, and exactly one stream stays live + // - no accumulation of abandoned SSE connections. + expect(signals[0].aborted).toBe(true); + expect(signals.filter((s) => !s.aborted)).toHaveLength(1); + + client.closePushNotifications(); + }); }); describe('fetchBlobAsObjectUrl', () => { diff --git a/lib/jmap/client.ts b/lib/jmap/client.ts index 83c87b166..e524ae922 100644 --- a/lib/jmap/client.ts +++ b/lib/jmap/client.ts @@ -6244,6 +6244,11 @@ export class JMAPClient implements IJMAPClient { private sseReconnectTimeout: NodeJS.Timeout | null = null; private ssePingTimer: NodeJS.Timeout | null = null; private lastSSEActivity: number = 0; + // Guards a state-change poll from stacking: the 3s interval, the secondary + // poll, and the visibility/online handlers all call checkForStateChanges, so + // under socket pressure they would otherwise pile identical POSTs onto an + // already-saturated pool and amplify the queueing (#781). + private stateCheckInFlight: boolean = false; private visibilityHandler: (() => void) | null = null; private onlineHandler: (() => void) | null = null; @@ -6306,6 +6311,25 @@ export class JMAPClient implements IJMAPClient { .replace('{closeafter}', 'no') .replace('{ping}', '30'); + // Already-active guard (#781): never stack a second SSE stream on a live + // one. readSSEStream only unwinds on abort or stream-end, so a previous + // connect left open would hold its HTTP/1.1 socket indefinitely; with + // several accounts each pinning a socket, routine JMAP POSTs then queue and + // time out. Abort any prior stream (and its ping monitor / pending + // reconnect) before opening the replacement. The old read loop unwinds via + // AbortError and, because we install a fresh controller below, its + // end-of-stream reconnect guard (`sseAbortController === controller`) no + // longer matches, so it does not spawn a competing connection. + if (this.sseAbortController) { + this.sseAbortController.abort(); + this.sseAbortController = null; + } + this.stopSSEPingMonitor(); + if (this.sseReconnectTimeout) { + clearTimeout(this.sseReconnectTimeout); + this.sseReconnectTimeout = null; + } + // 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 @@ -6428,6 +6452,9 @@ export class JMAPClient implements IJMAPClient { if (this.isRateLimited()) { return; } + // Idempotent: a repeat setup (or an SSE->poll fallback that races another) + // must not leak a second 3s interval alongside the first (#781). + if (this.pollingInterval) return; this.fetchCurrentStates(); this.pollingInterval = setInterval(() => { this.checkForStateChanges(); @@ -6518,6 +6545,13 @@ export class JMAPClient implements IJMAPClient { if (this.isRateLimited()) { return; } + // Coalesce overlapping polls: if one is already awaiting a response, skip + // this call rather than issuing a duplicate POST (#781). The next tick + // picks up any change once the in-flight one settles. + if (this.stateCheckInFlight) { + return; + } + this.stateCheckInFlight = true; try { const { using, methodCalls } = this.buildStatePollingRequest(); const response = await this.authenticatedFetch(this.apiUrl, { @@ -6550,6 +6584,8 @@ export class JMAPClient implements IJMAPClient { } } catch { // Silently fail - polling will retry + } finally { + this.stateCheckInFlight = false; } }