fix: cap live SSE streams per tab so many logins can't starve JMAP requests #702

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));

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<string, string> | 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<Response>((_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<JMAPClient[]> {
+      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();

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<JMAPClient>();
+  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<string, string> = {
     '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;