fix: improve event stream reconnect reliability

Recover stalled event streams without dropping the session Wait briefly for reconnection before showing connection lost errors Persist Electron server logs for easier disconnect debugging
2026-04-22 21:03:02 +03:00
parent 17dd526731
commit f24e6de21b
10 changed files with 493 additions and 139 deletions
@@ -28,7 +28,9 @@ This module contains the OpenChamber message-stream WebSocket protocol and runti
 - Browser clients connect to the WS endpoints above.
 - OpenChamber still fetches OpenCode upstream event streams over SSE.
 - Each WS connection proxies one upstream SSE stream.
- Global synthetic events such as `openchamber:session-status`, `openchamber:session-activity`, `openchamber:notification`, and `openchamber:heartbeat` are preserved on the WS path.
+- If an upstream SSE stream stalls after the browser WS is already ready, the runtime aborts that upstream fetch and reconnects upstream with `Last-Event-ID`, keeping the browser WS alive when recovery is fast.
+- Health checks are reserved for initial upstream connect failures and explicit upstream-unavailable responses, not for ordinary stall recovery on an already-established stream.
+- Global synthetic events such as `openchamber:session-status`, `openchamber:session-activity`, `openchamber:notification`, and `openchamber:heartbeat` are preserved on the WS path, but heartbeat frames are emitted only while an upstream SSE stream is actively attached.
 - Global UI broadcasts are fan-out capable across both SSE and WS clients.

 ## Notes for contributors
@@ -10,6 +10,9 @@ import {
  sendMessageStreamWsFrame,
 } from './protocol.js';

+const MESSAGE_STREAM_UPSTREAM_STALL_TIMEOUT_MS = 20_000;
+const MESSAGE_STREAM_UPSTREAM_RECONNECT_DELAY_MS = 250;
+
 function shouldTriggerUpstreamHealthCheck(upstream) {
  if (!upstream) {
    return true;
@@ -67,6 +70,9 @@ export function createMessageStreamWsRuntime({
  processForwardedEventPayload,
  wsClients,
  triggerHealthCheck,
+  heartbeatIntervalMs = MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS,
+  upstreamStallTimeoutMs = MESSAGE_STREAM_UPSTREAM_STALL_TIMEOUT_MS,
+  upstreamReconnectDelayMs = MESSAGE_STREAM_UPSTREAM_RECONNECT_DELAY_MS,
  fetchImpl = fetch,
 }) {
  const wsServer = new WebSocketServer({
@@ -82,13 +88,22 @@ export function createMessageStreamWsRuntime({
    const requestedDirectory = requestUrl.searchParams.get('directory')?.trim() || '';

    const controller = new AbortController();
+    let currentUpstreamAbortController = null;
+    let upstreamConnected = false;
+    let streamReady = false;
+    let lastEventId = requestedLastEventId;
    const cleanup = () => {
      if (!controller.signal.aborted) {
        controller.abort();
      }
+      if (currentUpstreamAbortController && !currentUpstreamAbortController.signal.aborted) {
+        currentUpstreamAbortController.abort();
+      }
      wsClients.delete(socket);
    };

+    const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
+
    const pingInterval = setInterval(() => {
      if (socket.readyState !== 1) {
        return;
@@ -98,15 +113,20 @@ export function createMessageStreamWsRuntime({
        socket.ping();
      } catch {
      }
-    }, MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS);
+    }, heartbeatIntervalMs);

    const heartbeatInterval = setInterval(() => {
+      if (!upstreamConnected) {
+        return;
+      }
+
      sendMessageStreamWsEvent(socket, { type: 'openchamber:heartbeat', timestamp: Date.now() }, { directory: 'global' });
-    }, MESSAGE_STREAM_WS_HEARTBEAT_INTERVAL_MS);
+    }, heartbeatIntervalMs);

    socket.on('close', () => {
      clearInterval(pingInterval);
      clearInterval(heartbeatInterval);
+      upstreamConnected = false;
      cleanup();
    });

@@ -115,73 +135,6 @@ export function createMessageStreamWsRuntime({
    });

    const run = async () => {
-      let targetUrl;
-      try {
-        targetUrl = new URL(buildOpenCodeUrl(isGlobalStream ? '/global/event' : '/event', ''));
-      } catch {
-        sendMessageStreamWsFrame(socket, { type: 'error', message: 'OpenCode service unavailable' });
-        socket.close(1011, 'OpenCode service unavailable');
-        return;
-      }
-
-      if (!isGlobalStream && requestedDirectory) {
-        targetUrl.searchParams.set('directory', requestedDirectory);
-      }
-
-      const headers = {
-        Accept: 'text/event-stream',
-        'Cache-Control': 'no-cache',
-        Connection: 'keep-alive',
-        ...getOpenCodeAuthHeaders(),
-      };
-
-      if (requestedLastEventId) {
-        headers['Last-Event-ID'] = requestedLastEventId;
-      }
-
-      let upstream;
-      try {
-        upstream = await fetchImpl(targetUrl.toString(), {
-          headers,
-          signal: controller.signal,
-        });
-      } catch {
-        if (!controller.signal.aborted) {
-          sendMessageStreamWsFrame(socket, { type: 'error', message: 'Failed to connect to OpenCode event stream' });
-          socket.close(1011, 'Failed to connect to OpenCode event stream');
-          // Trigger immediate health check so the server detects and
-          // restarts a dead OpenCode process without waiting for the next
-          // periodic interval (up to 15 s).
-          triggerHealthCheck?.();
-        }
-        return;
-      }
-
-      if (!upstream.ok || !upstream.body) {
-        sendMessageStreamWsFrame(socket, {
-          type: 'error',
-          message: `OpenCode event stream unavailable (${upstream.status})`,
-        });
-        socket.close(1011, 'OpenCode event stream unavailable');
-        if (shouldTriggerUpstreamHealthCheck(upstream)) {
-          triggerHealthCheck?.();
-        }
-        return;
-      }
-
-      sendMessageStreamWsFrame(socket, {
-        type: 'ready',
-        scope: isGlobalStream ? 'global' : 'directory',
-      });
-
-      if (isGlobalStream) {
-        wsClients.add(socket);
-      }
-
-      const decoder = new TextDecoder();
-      const reader = upstream.body.getReader();
-      let buffer = '';
-
      const forwardBlock = (block) => {
        if (!block) {
          return;
@@ -197,6 +150,10 @@ export function createMessageStreamWsRuntime({
          ? (typeof envelope?.directory === 'string' && envelope.directory.length > 0 ? envelope.directory : 'global')
          : (requestedDirectory || envelope?.directory || 'global');

+        if (typeof envelope?.eventId === 'string' && envelope.eventId.length > 0) {
+          lastEventId = envelope.eventId;
+        }
+
        sendMessageStreamWsEvent(socket, payload, {
          directory,
          eventId: typeof envelope?.eventId === 'string' && envelope.eventId.length > 0 ? envelope.eventId : undefined,
@@ -208,25 +165,148 @@ export function createMessageStreamWsRuntime({
      };

      try {
-        while (true) {
-          const { value, done } = await reader.read();
-          if (done) {
-            break;
+        while (!controller.signal.aborted) {
+          let targetUrl;
+          try {
+            targetUrl = new URL(buildOpenCodeUrl(isGlobalStream ? '/global/event' : '/event', ''));
+          } catch {
+            sendMessageStreamWsFrame(socket, { type: 'error', message: 'OpenCode service unavailable' });
+            socket.close(1011, 'OpenCode service unavailable');
+            return;
          }

-          buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n');
-
-          let separatorIndex = buffer.indexOf('\n\n');
-          while (separatorIndex !== -1) {
-            const block = buffer.slice(0, separatorIndex);
-            buffer = buffer.slice(separatorIndex + 2);
-            forwardBlock(block);
-            separatorIndex = buffer.indexOf('\n\n');
+          if (!isGlobalStream && requestedDirectory) {
+            targetUrl.searchParams.set('directory', requestedDirectory);
          }
-        }

-        if (buffer.trim().length > 0) {
-          forwardBlock(buffer.trim());
+          const headers = {
+            Accept: 'text/event-stream',
+            'Cache-Control': 'no-cache',
+            Connection: 'keep-alive',
+            ...getOpenCodeAuthHeaders(),
+          };
+          if (lastEventId) {
+            headers['Last-Event-ID'] = lastEventId;
+          }
+
+          const upstreamController = new AbortController();
+          currentUpstreamAbortController = upstreamController;
+          const abortUpstream = () => upstreamController.abort();
+          controller.signal.addEventListener('abort', abortUpstream, { once: true });
+
+          let upstream;
+          try {
+            upstream = await fetchImpl(targetUrl.toString(), {
+              headers,
+              signal: upstreamController.signal,
+            });
+          } catch {
+            controller.signal.removeEventListener('abort', abortUpstream);
+            currentUpstreamAbortController = null;
+            upstreamConnected = false;
+            if (controller.signal.aborted) {
+              return;
+            }
+            if (!streamReady) {
+              sendMessageStreamWsFrame(socket, { type: 'error', message: 'Failed to connect to OpenCode event stream' });
+              socket.close(1011, 'Failed to connect to OpenCode event stream');
+              triggerHealthCheck?.();
+              return;
+            }
+            await wait(upstreamReconnectDelayMs);
+            continue;
+          }
+
+          if (!upstream.ok || !upstream.body) {
+            controller.signal.removeEventListener('abort', abortUpstream);
+            currentUpstreamAbortController = null;
+            upstreamConnected = false;
+            if (!streamReady) {
+              sendMessageStreamWsFrame(socket, {
+                type: 'error',
+                message: `OpenCode event stream unavailable (${upstream.status})`,
+              });
+              socket.close(1011, 'OpenCode event stream unavailable');
+              if (shouldTriggerUpstreamHealthCheck(upstream)) {
+                triggerHealthCheck?.();
+              }
+              return;
+            }
+            await wait(upstreamReconnectDelayMs);
+            continue;
+          }
+
+          if (!streamReady) {
+            sendMessageStreamWsFrame(socket, {
+              type: 'ready',
+              scope: isGlobalStream ? 'global' : 'directory',
+            });
+            streamReady = true;
+
+            if (isGlobalStream) {
+              wsClients.add(socket);
+            }
+          }
+
+          upstreamConnected = true;
+          const decoder = new TextDecoder();
+          const reader = upstream.body.getReader();
+          let buffer = '';
+          let upstreamAbortReason = null;
+          let stallTimer = null;
+          const resetStallTimer = () => {
+            if (stallTimer) {
+              clearTimeout(stallTimer);
+            }
+            stallTimer = setTimeout(() => {
+              upstreamAbortReason = 'upstream_stalled';
+              upstreamConnected = false;
+              upstreamController.abort();
+            }, upstreamStallTimeoutMs);
+          };
+
+          resetStallTimer();
+
+          try {
+            while (true) {
+              const { value, done } = await reader.read();
+              if (done) {
+                break;
+              }
+
+              resetStallTimer();
+              buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n');
+
+              let separatorIndex = buffer.indexOf('\n\n');
+              while (separatorIndex !== -1) {
+                const block = buffer.slice(0, separatorIndex);
+                buffer = buffer.slice(separatorIndex + 2);
+                forwardBlock(block);
+                separatorIndex = buffer.indexOf('\n\n');
+              }
+            }
+
+            if (buffer.trim().length > 0) {
+              forwardBlock(buffer.trim());
+            }
+          } catch (error) {
+            if (!controller.signal.aborted && upstreamAbortReason !== 'upstream_stalled') {
+              console.warn('Message stream WS proxy error:', error);
+            }
+          } finally {
+            if (stallTimer) {
+              clearTimeout(stallTimer);
+            }
+            upstreamConnected = false;
+            currentUpstreamAbortController = null;
+            controller.signal.removeEventListener('abort', abortUpstream);
+          }
+
+          if (controller.signal.aborted) {
+            return;
+          }
+
+          await wait(upstreamReconnectDelayMs);
        }
      } catch (error) {
        if (!controller.signal.aborted) {
@@ -1,6 +1,68 @@
+import { EventEmitter } from 'node:events';
 import { describe, expect, it } from 'vitest';

-import { createGlobalUiEventBroadcaster } from './runtime.js';
+import { createGlobalUiEventBroadcaster, createMessageStreamWsRuntime } from './runtime.js';
+
+class FakeSocket extends EventEmitter {
+  constructor() {
+    super();
+    this.readyState = 1;
+    this.sent = [];
+    this.closeCalls = [];
+  }
+
+  send(payload) {
+    this.sent.push(JSON.parse(payload));
+  }
+
+  ping() {
+    void 0;
+  }
+
+  close(code, reason) {
+    if (this.readyState === 3) {
+      return;
+    }
+    this.readyState = 3;
+    this.closeCalls.push({ code, reason });
+    this.emit('close');
+  }
+}
+
+function createSseResponse({ blocks = [], signal, holdOpen = false }) {
+  const encoder = new TextEncoder();
+  let index = 0;
+
+  return {
+    ok: true,
+    body: {
+      getReader() {
+        return {
+          async read() {
+            if (index < blocks.length) {
+              const next = blocks[index++];
+              return { value: encoder.encode(next), done: false };
+            }
+
+            if (!holdOpen) {
+              return { value: undefined, done: true };
+            }
+
+            return new Promise((resolve, reject) => {
+              const onAbort = () => {
+                signal.removeEventListener('abort', onAbort);
+                const error = new Error('Aborted');
+                error.name = 'AbortError';
+                reject(error);
+              };
+              signal.addEventListener('abort', onAbort, { once: true });
+            });
+          },
+        };
+      },
+    },
+  };
+}

 describe('event stream broadcaster', () => {
  it('fans out synthetic events to SSE and WS clients', () => {
@@ -63,3 +125,71 @@ describe('event stream broadcaster', () => {
    expect(wsClients.size).toBe(0);
  });
 });
+
+describe('message stream websocket runtime', () => {
+  it('reconnects a stalled upstream SSE stream and resumes from the last event id', async () => {
+    const server = new EventEmitter();
+    const wsClients = new Set();
+    let triggerHealthCheckCalls = 0;
+    const fetchCalls = [];
+    let upstreamAttempt = 0;
+
+    const runtime = createMessageStreamWsRuntime({
+      server,
+      uiAuthController: null,
+      isRequestOriginAllowed: async () => true,
+      rejectWebSocketUpgrade() {
+        throw new Error('upgrade should not be used in this test');
+      },
+      buildOpenCodeUrl: (path) => `http://127.0.0.1:4096${path}`,
+      getOpenCodeAuthHeaders: () => ({}),
+      processForwardedEventPayload() {},
+      wsClients,
+      triggerHealthCheck: () => {
+        triggerHealthCheckCalls += 1;
+      },
+      heartbeatIntervalMs: 50,
+      upstreamStallTimeoutMs: 20,
+      upstreamReconnectDelayMs: 0,
+      fetchImpl: async (_url, options) => {
+        const lastEventId = options?.headers?.['Last-Event-ID'] ?? null;
+        fetchCalls.push(lastEventId);
+        upstreamAttempt += 1;
+
+        if (upstreamAttempt === 1) {
+          return createSseResponse({
+            signal: options.signal,
+            holdOpen: true,
+            blocks: [
+              'id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n',
+            ],
+          });
+        }
+
+        return createSseResponse({
+          signal: options.signal,
+          holdOpen: false,
+          blocks: [
+            'id: evt-2\ndata: {"type":"server.connected","properties":{}}\n\n',
+          ],
+        });
+      },
+    });
+
+    const socket = new FakeSocket();
+    runtime.wsServer.emit('connection', socket, { url: '/api/global/event/ws' });
+
+    await new Promise((resolve) => setTimeout(resolve, 35));
+
+    const readyFrames = socket.sent.filter((frame) => frame.type === 'ready');
+    const eventFrames = socket.sent.filter((frame) => frame.type === 'event' && frame.payload?.type === 'server.connected');
+
+    expect(readyFrames).toHaveLength(1);
+    expect(eventFrames.length).toBeGreaterThanOrEqual(2);
+    expect(fetchCalls.slice(0, 2)).toEqual([null, 'evt-1']);
+    expect(triggerHealthCheckCalls).toBe(0);
+
+    socket.close();
+    await runtime.close();
+  });
+});