fix(ports): a worker redials a gateway that vanished without closing the uplink

src/ports/uplink-websocket.test.ts

@@ -14,17 +14,25 @@
  * fabricated request object.
  */
 import { createServer, type Server } from "node:http";
-import type { AddressInfo } from "node:net";
+import { createServer as createTcpServer, type AddressInfo } from "node:net";
 
 import { afterEach, describe, expect, it, vi } from "vitest";
-import { WebSocket } from "ws";
+import { WebSocket, WebSocketServer } from "ws";
+
+import { FakeClock } from "./clock.js";
+import { GatewayUplink } from "../daemon/gateway-uplink.js";
 
 import type {
   ClaimedUplinkIdentity,
   UplinkAuthOutcome,
   UplinkTransportConnection,
 } from "./uplink.js";
-import { WebSocketUplinkConnector, WebSocketUplinkListenerFactory } from "./uplink-websocket.js";
+import { UplinkError } from "./uplink.js";
+import {
+  UPLINK_PING_INTERVAL_MS,
+  WebSocketUplinkConnector,
+  WebSocketUplinkListenerFactory,
+} from "./uplink-websocket.js";
 
 let server: Server | undefined;
 const openConnections: Array<{ close(): unknown }> = [];
@@ -219,3 +227,115 @@ describe("WebSocketUplinkConnection binary messages", () => {
     expect(onSent).toHaveBeenCalledTimes(1);
   });
 });
+
+async function startGateway(answersPings: boolean): Promise<{
+  url: string;
+  peers: WebSocket[];
+  pings: () => number;
+}> {
+  const peers: WebSocket[] = [];
+  let pingCount = 0;
+  const wss = new WebSocketServer({ noServer: true, autoPong: answersPings });
+  server = createServer();
+  server.on("upgrade", (request, socket, head) => {
+    wss.handleUpgrade(request, socket, head, (peer) => {
+      peers.push(peer);
+      openConnections.push({ close: () => peer.terminate() });
+      peer.on("ping", () => {
+        pingCount += 1;
+        peer.send("after-pong");
+      });
+    });
+  });
+  await new Promise<void>((resolve) => server?.listen(0, "127.0.0.1", resolve));
+  const { port } = server.address() as AddressInfo;
+  return {
+    url: `ws://127.0.0.1:${port}`,
+    peers,
+    pings: () => pingCount,
+  };
+}
+
+describe("WebSocketUplinkConnector liveness", () => {
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("terminates an uplink whose gateway stops answering pings within two intervals, and GatewayUplink redials", async () => {
+    vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
+    const gateway = await startGateway(false);
+    const clock = new FakeClock();
+    let accepted = 0;
+    const uplink = new GatewayUplink({
+      clock,
+      connector: new WebSocketUplinkConnector(),
+      url: gateway.url,
+      token: "join",
+      workerId: "wrk_1",
+      accept: () => {
+        accepted += 1;
+      },
+      backoff: { initialMs: 1_000, maxMs: 1_000, multiplier: 1 },
+      random: () => 1,
+    });
+    uplink.start();
+    await vi.waitFor(() => expect(accepted).toBe(1));
+    const first = gateway.peers[0];
+    const firstClosed = new Promise<void>((resolve) => first?.on("close", () => resolve()));
+
+    await vi.advanceTimersByTimeAsync(UPLINK_PING_INTERVAL_MS);
+    await vi.waitFor(() => expect(gateway.pings()).toBe(1));
+    await vi.advanceTimersByTimeAsync(UPLINK_PING_INTERVAL_MS);
+    const firstOutcome = await Promise.race([
+      firstClosed.then(() => "closed"),
+      new Promise<string>((resolve) => setTimeout(() => resolve("still open"), 1_000)),
+    ]);
+    expect(firstOutcome).toBe("closed");
+
+    clock.advance(1_000);
+    await vi.waitFor(() => expect(accepted).toBe(2));
+    expect(gateway.peers).toHaveLength(2);
+    await uplink.stop();
+  });
+
+  it("keeps a healthy uplink open across several ping rounds", async () => {
+    vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
+    const gateway = await startGateway(true);
+    const worker = await new WebSocketUplinkConnector().connect(dialOptions(gateway.url));
+    openConnections.push(worker);
+    let afterPongs = 0;
+    worker.onData(() => {
+      afterPongs += 1;
+    });
+
+    for (let round = 1; round <= 4; round += 1) {
+      await vi.advanceTimersByTimeAsync(UPLINK_PING_INTERVAL_MS);
+      await vi.waitFor(() => expect(afterPongs).toBe(round));
+    }
+
+    expect(gateway.pings()).toBe(4);
+    expect(worker.closed).toBe(false);
+  });
+});
+
+describe("WebSocketUplinkConnector handshake", () => {
+  it("fails the dial with an UplinkError when the server accepts TCP but never completes the handshake", async () => {
+    const sockets: Array<{ destroy(): unknown }> = [];
+    const silent = createTcpServer((socket) => sockets.push(socket));
+    await new Promise<void>((resolve) => silent.listen(0, "127.0.0.1", resolve));
+    const { port } = silent.address() as AddressInfo;
+    try {
+      const outcome = await Promise.race([
+        new WebSocketUplinkConnector({ handshakeTimeoutMs: 100 })
+          .connect(dialOptions(`ws://127.0.0.1:${port}`))
+          .catch((error: unknown) => error),
+        new Promise<string>((resolve) => setTimeout(() => resolve("still dialling"), 2_000)),
+      ]);
+
+      expect(outcome).toBeInstanceOf(UplinkError);
+    } finally {
+      for (const socket of sockets) socket.destroy();
+      silent.close();
+    }
+  });
+});

src/ports/uplink-websocket.ts

@@ -36,6 +36,14 @@ import {
   type UplinkTransportConnection,
 } from "./uplink.js";
 
+/** why: a ping that finds no pong since the previous one terminates the link, so a gateway that
+ * vanished without closing is noticed within two of these. */
+export const UPLINK_PING_INTERVAL_MS = 15_000;
+
+/** why: a dial into a host that drops packets would otherwise wait for the OS connect timeout
+ * (about 75s on macOS) before the next backoff step. */
+const UPLINK_HANDSHAKE_TIMEOUT_MS = 10_000;
+
 function rawBytes(data: RawData): Buffer {
   if (Array.isArray(data)) return Buffer.concat(data);
   return Buffer.isBuffer(data) ? data : Buffer.from(data as ArrayBuffer);
@@ -275,10 +283,19 @@ export class WebSocketUplinkListenerFactory implements UplinkListenerFactory {
 
 /** The worker's half: dials `gateway.url` with the join token and its instance id. */
 export class WebSocketUplinkConnector implements UplinkConnector {
+  readonly #handshakeTimeoutMs: number;
+
+  /** why: `handshakeTimeoutMs` exists so a test can see the dial give up without waiting the
+   * real `UPLINK_HANDSHAKE_TIMEOUT_MS`; production passes nothing. */
+  constructor(options: { readonly handshakeTimeoutMs?: number } = {}) {
+    this.#handshakeTimeoutMs = options.handshakeTimeoutMs ?? UPLINK_HANDSHAKE_TIMEOUT_MS;
+  }
+
   async connect(options: UplinkDialOptions): Promise<UplinkTransportConnection> {
     // `options.url` is `gateway.url`, the gateway's base URL; the endpoint is this worker's to
     // derive (ADR 0005 §4), so an operator configures one URL and never an internal path.
     const socket = new WebSocket(resolveUplinkUrl(options.url), {
+      handshakeTimeout: this.#handshakeTimeoutMs,
       headers: {
         authorization: `Bearer ${options.token}`,
         [WORKER_ID_HEADER]: options.workerId,
@@ -298,10 +315,30 @@ export class WebSocketUplinkConnector implements UplinkConnector {
       socket.terminate();
       throw error;
     }
+    watchLiveness(socket);
     return connection;
   }
 }
 
+/** why: over the uplink only the gateway writes, so a worker whose gateway vanished sees no
+ * `close` unless it pings; terminating makes the `close` that drives `GatewayUplink`'s redial. */
+function watchLiveness(socket: WebSocket): void {
+  let awaitingPong = false;
+  socket.on("pong", () => {
+    awaitingPong = false;
+  });
+  const timer = setInterval(() => {
+    if (awaitingPong) {
+      socket.terminate();
+      return;
+    }
+    awaitingPong = true;
+    socket.ping();
+  }, UPLINK_PING_INTERVAL_MS);
+  timer.unref();
+  socket.once("close", () => clearInterval(timer));
+}
+
 function waitForOpen(socket: WebSocket): Promise<void> {
   return new Promise<void>((resolve, reject) => {
     const onOpen = () => {