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.
This commit is contained in:
Stefan Hildebrandt
2026-08-16 21:09:13 +02:00
parent 5ab5749306
commit 26e8c5797f
2 changed files with 60 additions and 0 deletions
@@ -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', () => {
+36
View File
@@ -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;
}
}