fix(daemon): retry a Slack socket that fails or stalls before it opens (#942)

A Slack socket that fails or stalls before it opens is retried, and stopping the gateway still completes.

packages/daemon/src/domain/gateway/slack/socket-inbound.ts

@@ -30,6 +30,8 @@ export interface SocketInboundDeps {
   inboundMaxConnects?: number;
   /** Dead-letter retry cadence WHILE the socket stays connected (default 5min). */
   retryIntervalMs?: number;
+  /** How long a new socket may stay opening before it is abandoned and retried (default 30s). */
+  openTimeoutMs?: number;
   receipts?: InboundReceiptStore;
   recovery?: { run(): Promise<void>; stop(): void };
   log?: (msg: string) => void;
@@ -58,11 +60,13 @@ export function startSocketInbound(appToken: string, router: InboundRouter, deps
   const log = deps.log ?? (() => {});
   const wsFactory = deps.wsFactory ?? ((url: string) => new (globalThis as unknown as { WebSocket: new (u: string) => WsLike }).WebSocket(url));
   const retryIntervalMs = deps.retryIntervalMs ?? 5 * 60 * 1000;
+  const openTimeoutMs = deps.openTimeoutMs ?? 30_000;
   let connects = 0;
   let backoff = 1000;
   let stopped = false;
   let liveWs: WsLike | undefined;
   let pendingTimer: ReturnType<typeof setTimeout> | undefined;
+  let openTimer: ReturnType<typeof setTimeout> | undefined;
   let retryTimer: ReturnType<typeof setInterval> | undefined;
   let finish: () => void = () => {};
   const status: SocketInboundStatus = { generation: 0, reconnects: 0, state: "disconnected" };
@@ -107,7 +111,36 @@ export function startSocketInbound(appToken: string, router: InboundRouter, deps
       }
       const ws = wsFactory(open.url);
       liveWs = ws;
+      // Until the socket opens, onclose cannot be relied on to drive the reconnect: Node's
+      // WebSocket (undici 6) fires onerror but never onclose when the handshake fails
+      // (refused, reset, or its own 300s headers timeout), and an upgrade that is never
+      // answered fires nothing at all. Either way the loop would stay "connecting" forever.
+      // So a failure or timeout before onopen detaches the socket and schedules the retry here.
+      let opened = false;
+      const abandon = (reason: "socket-open-error" | "socket-open-timeout"): void => {
+        clearTimeout(openTimer);
+        openTimer = undefined;
+        if (stopped) return;
+        ws.onopen = ws.onmessage = ws.onclose = ws.onerror = null;
+        try {
+          ws.close();
+        } catch {
+          /* ignore */
+        }
+        liveWs = undefined;
+        log(`socket did not open (${reason}); reconnect in ${backoff}ms`);
+        status.state = "disconnected";
+        status.disconnectedAt = stamp();
+        receipt({ generation: connects, status: "connect-failed", reason });
+        if (deps.inboundMaxConnects && connects >= deps.inboundMaxConnects) return resolve();
+        pendingTimer = setTimeout(connect, backoff);
+        backoff = Math.min(backoff * 2, 60000);
+      };
+      openTimer = setTimeout(() => abandon("socket-open-timeout"), openTimeoutMs);
       ws.onopen = () => {
+        opened = true;
+        clearTimeout(openTimer);
+        openTimer = undefined;
         if (stopped) return;
         backoff = 1000;
         log("socket connected");
@@ -171,6 +204,8 @@ export function startSocketInbound(appToken: string, router: InboundRouter, deps
           });
       };
       ws.onclose = () => {
+        clearTimeout(openTimer);
+        openTimer = undefined;
         if (retryTimer) clearInterval(retryTimer);
         liveWs = undefined;
         status.state = stopped ? "stopped" : "disconnected";
@@ -183,6 +218,7 @@ export function startSocketInbound(appToken: string, router: InboundRouter, deps
         backoff = Math.min(backoff * 2, 60000);
       };
       ws.onerror = () => {
+        if (!opened) return abandon("socket-open-error");
         try {
           ws.close();
         } catch {
@@ -201,6 +237,7 @@ export function startSocketInbound(appToken: string, router: InboundRouter, deps
       if (retryTimer) clearInterval(retryTimer);
       status.state = "stopped";
       if (pendingTimer) clearTimeout(pendingTimer);
+      if (openTimer) clearTimeout(openTimer);
       try { liveWs?.close(); } catch { /* best-effort */ }
       finish(); // A canceled backoff has no future connect/close callback to settle done.
     },

packages/daemon/test/slack-socket-inbound.test.ts

@@ -64,6 +64,90 @@ describe("Slice-11 INBOUND transport — real runInboundLoop / open / message /
     ]));
   });
 
+  it("abandons a socket that never opens and reconnects instead of staying connecting forever", async () => {
+    const fsx = memFs();
+    const receipts = new InboundReceiptStore("/s/inbound-receipts.jsonl", fsx, clock);
+    const router = new InboundRouter({
+      queue: { createQitem: async () => "unused" },
+      seen: new SeenStore("/s/seen.jsonl", fsx, clock),
+      deadLetter: new DeadLetterStore<SlackEvent>("/s/dead.jsonl", fsx, clock),
+      destination: "operator-agent@kernel",
+      resolveSender: () => ({ admitted: true, source: "human-founder@external" }),
+      log: () => {},
+    });
+    // The first socket hangs: no onopen, no onclose, no onerror. The second opens at once.
+    const hung = makeFakeWs();
+    let hungClosed = 0;
+    hung.ws.close = () => { hungClosed++; };
+    const healthy = makeFakeWs();
+    const sockets = [hung.ws, healthy.ws];
+    let opened = false;
+    const handle = startSocketInbound("xapp-EXAMPLE-fake", router, {
+      fetchImpl: openFetch,
+      wsFactory: () => {
+        const ws = sockets.shift()!;
+        if (ws === healthy.ws) queueMicrotask(() => { opened = true; ws.onopen!(); });
+        return ws;
+      },
+      openTimeoutMs: 20,
+      inboundMaxConnects: 2,
+      receipts,
+      log: () => {},
+    });
+
+    await expect.poll(() => receipts.readAll().some((r) => r.status === "connect-failed")).toBe(true);
+    expect(handle.status()).toMatchObject({ generation: 1, state: "disconnected" });
+    expect(hungClosed).toBe(1);
+    // A late event from the abandoned socket has no handler to reach the loop.
+    expect([hung.ws.onopen, hung.ws.onmessage, hung.ws.onclose, hung.ws.onerror]).toEqual([null, null, null, null]);
+
+    // After the first backoff the loop opens a fresh socket.
+    await expect.poll(() => opened, { timeout: 3000 }).toBe(true);
+    await new Promise((r) => setTimeout(r, 60)); // well past openTimeoutMs
+    expect(handle.status()).toMatchObject({ generation: 2, reconnects: 1, state: "connected" });
+    expect(receipts.readAll().map((r) => [r.generation, r.status, r.reason])).toEqual([
+      [1, "connect-attempt", undefined],
+      [1, "connect-failed", "socket-open-timeout"],
+      [2, "connect-attempt", undefined],
+      [2, "connected", undefined],
+    ]);
+
+    healthy.ws.onclose!();
+    await handle.done;
+  });
+
+  it("retries after an error before open even though no close follows it", async () => {
+    const fsx = memFs();
+    const receipts = new InboundReceiptStore("/s/inbound-receipts.jsonl", fsx, clock);
+    const router = new InboundRouter({
+      queue: { createQitem: async () => "unused" },
+      seen: new SeenStore("/s/seen.jsonl", fsx, clock),
+      deadLetter: new DeadLetterStore<SlackEvent>("/s/dead.jsonl", fsx, clock),
+      destination: "operator-agent@kernel",
+      resolveSender: () => ({ admitted: true, source: "human-founder@external" }),
+      log: () => {},
+    });
+    const failing = makeFakeWs();
+    const handle = startSocketInbound("xapp-EXAMPLE-fake", router, {
+      fetchImpl: openFetch,
+      wsFactory: () => {
+        queueMicrotask(() => failing.ws.onerror!()); // refused: onerror, and never onclose
+        return failing.ws;
+      },
+      openTimeoutMs: 60_000, // far beyond the test: the error alone must end the attempt
+      inboundMaxConnects: 1,
+      receipts,
+      log: () => {},
+    });
+
+    await handle.done;
+    expect(handle.status()).toMatchObject({ generation: 1, state: "disconnected" });
+    expect(receipts.readAll().map((r) => [r.status, r.reason])).toEqual([
+      ["connect-attempt", undefined],
+      ["connect-failed", "socket-open-error"],
+    ]);
+  });
+
   it("acks every envelope, lands human messages, and retries dead-letters ON CONNECT and PERIODICALLY while connected", async () => {
     const order: string[] = [];
     const fsx = memFs((path, data) => {

packages/daemon/test/slack-socket-stop-completion.test.ts

@@ -109,3 +109,93 @@ it("settles an already connected native WebSocket loop on repeated stop", async
     rmSync(home, { recursive: true, force: true });
   }
 });
+
+it("abandons a native WebSocket whose upgrade never answers and opens a new one", async () => {
+  const home = mkdtempSync(join(tmpdir(), "openrig-socket-open-timeout-"));
+  const held = new Set<import("node:stream").Duplex>();
+  let upgrades = 0;
+  const server = createServer((_req, res) => {
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("missing fixture address");
+    res.setHeader("content-type", "application/json");
+    res.end(JSON.stringify({ ok: true, url: `ws://127.0.0.1:${address.port}/socket` }));
+  });
+  // Accept the TCP connection and the upgrade request, then never answer it.
+  server.on("upgrade", (_request, socket) => {
+    upgrades++;
+    held.add(socket);
+    socket.on("close", () => held.delete(socket));
+  });
+  server.listen(0, "127.0.0.1");
+  await once(server, "listening");
+  let handle: SocketInboundHandle | undefined;
+  try {
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("missing fixture address");
+    const router = new InboundRouter({
+      queue: { createQitem: async () => { throw new Error("unexpected inbound event"); } },
+      seen: new SeenStore(join(home, "seen.jsonl")),
+      deadLetter: new DeadLetterStore<SlackEvent>(join(home, "dead.jsonl")),
+      destination: "operator-agent@kernel",
+      resolveSender: () => ({ admitted: true, source: "human-fixture@external" }),
+    });
+    handle = startSocketInbound("synthetic-app-token", router, {
+      fetchImpl: (_url, init) => fetch(`http://127.0.0.1:${address.port}`, init),
+      openTimeoutMs: 200,
+      inboundMaxConnects: 2,
+    });
+    await expect.poll(() => upgrades).toBe(1);
+    expect(handle.status().state).toBe("connecting"); // the server holds the upgrade unanswered
+    // The timeout abandons it and, after the first backoff, a second socket really dials.
+    await expect.poll(() => upgrades, { timeout: 3000 }).toBe(2);
+    await handle.done; // the second attempt times out too, ending the allowed attempts
+    expect(handle.status()).toMatchObject({ generation: 2, state: "disconnected" });
+  } finally {
+    handle?.stop();
+    for (const socket of held) socket.destroy();
+    server.closeAllConnections();
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+    rmSync(home, { recursive: true, force: true });
+  }
+});
+
+it("retries when a native WebSocket handshake is refused, though Node fires no close for it", async () => {
+  const home = mkdtempSync(join(tmpdir(), "openrig-socket-refused-"));
+  // A port with nothing listening: the handshake is refused at once.
+  const vacant = createServer();
+  vacant.listen(0, "127.0.0.1");
+  await once(vacant, "listening");
+  const vacantAddress = vacant.address();
+  if (!vacantAddress || typeof vacantAddress === "string") throw new Error("missing fixture address");
+  await new Promise<void>((resolve) => vacant.close(() => resolve()));
+  const server = createServer((_req, res) => {
+    res.setHeader("content-type", "application/json");
+    res.end(JSON.stringify({ ok: true, url: `ws://127.0.0.1:${vacantAddress.port}/socket` }));
+  });
+  server.listen(0, "127.0.0.1");
+  await once(server, "listening");
+  let handle: SocketInboundHandle | undefined;
+  try {
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("missing fixture address");
+    const router = new InboundRouter({
+      queue: { createQitem: async () => { throw new Error("unexpected inbound event"); } },
+      seen: new SeenStore(join(home, "seen.jsonl")),
+      deadLetter: new DeadLetterStore<SlackEvent>(join(home, "dead.jsonl")),
+      destination: "operator-agent@kernel",
+      resolveSender: () => ({ admitted: true, source: "human-fixture@external" }),
+    });
+    handle = startSocketInbound("synthetic-app-token", router, {
+      fetchImpl: (_url, init) => fetch(`http://127.0.0.1:${address.port}`, init),
+      openTimeoutMs: 60_000, // far beyond the test: the refusal alone must end the attempt
+      inboundMaxConnects: 2,
+    });
+    await handle.done; // both attempts end on the refusal, without the timeout
+    expect(handle.status()).toMatchObject({ generation: 2, state: "disconnected" });
+  } finally {
+    handle?.stop();
+    server.closeAllConnections();
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+    rmSync(home, { recursive: true, force: true });
+  }
+});