Fix: treat an aborted SSE connect as a close, not a failure
Since the per-account push setup (#281), every account switch tears down and re-creates push notifications for all connected clients. Aborting an SSE connect that is still in flight lands in the fetch rejection handler, which treated it as a network failure: - fallbackToPolling() started an unsupervised 3s state poll on a client whose push had just been intentionally closed, after the cleanup that would have removed it had already run. - The late rejection also nulled sseAbortController, orphaning the replacement connection set up right after: it could never be aborted again and reconnected itself in parallel once the server closed it. Rapid switching multiplied both effects until the server's concurrency limit stalled the app entirely. Each connect attempt now tracks its own AbortController: an aborted attempt returns silently instead of falling back to polling, and the end-of-stream reconnect only fires if the stream is still the current one. Regression tests simulate the switch churn both ways.
This commit is contained in:
+19
-6
@@ -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<Uint8Array>): Promise<void> {
|
||||
private async readSSEStream(body: ReadableStream<Uint8Array>, controller: AbortController): Promise<void> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user