From 7aecc95dd94e8926a6c700a6ca23bd2087df3e90 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 20:00:40 +0200 Subject: [PATCH 1/6] test: a worker terminates an uplink whose gateway stops answering pings and redials (#941) --- src/ports/uplink-websocket.test.ts | 81 +++++++++++++++++++++++++++++- src/ports/uplink-websocket.ts | 4 ++ 2 files changed, 83 insertions(+), 2 deletions(-) diff --git a/src/ports/uplink-websocket.test.ts b/src/ports/uplink-websocket.test.ts index b8d10976a..9047d8578 100644 --- a/src/ports/uplink-websocket.test.ts +++ b/src/ports/uplink-websocket.test.ts @@ -17,14 +17,21 @@ import { createServer, type Server } from "node:http"; import 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 { + UPLINK_PING_INTERVAL_MS, + WebSocketUplinkConnector, + WebSocketUplinkListenerFactory, +} from "./uplink-websocket.js"; let server: Server | undefined; const openConnections: Array<{ close(): unknown }> = []; @@ -219,3 +226,73 @@ describe("WebSocketUplinkConnection binary messages", () => { expect(onSent).toHaveBeenCalledTimes(1); }); }); + +async function startMuteGateway(): Promise<{ + url: string; + peers: WebSocket[]; + pings: () => number; + nextPeer: (count: number) => Promise; +}> { + const peers: WebSocket[] = []; + let pingCount = 0; + const waiting: Array<{ count: number; resolve: () => void }> = []; + const wss = new WebSocketServer({ noServer: true, autoPong: false }); + 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; + }); + for (const w of waiting) if (peers.length >= w.count) w.resolve(); + }); + }); + await new Promise((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, + nextPeer: (count) => + peers.length >= count + ? Promise.resolve() + : new Promise((resolve) => waiting.push({ count, resolve })), + }; +} + +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 startMuteGateway(); + const clock = new FakeClock(); + const uplink = new GatewayUplink({ + clock, + connector: new WebSocketUplinkConnector(), + url: gateway.url, + token: "join", + workerId: "wrk_1", + accept: () => {}, + backoff: { initialMs: 1_000, maxMs: 1_000, multiplier: 1 }, + random: () => 1, + }); + uplink.start(); + await gateway.nextPeer(1); + const first = gateway.peers[0]; + const firstClosed = new Promise((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); + await firstClosed; + + clock.advance(1_000); + await gateway.nextPeer(2); + expect(gateway.peers).toHaveLength(2); + await uplink.stop(); + }); +}); diff --git a/src/ports/uplink-websocket.ts b/src/ports/uplink-websocket.ts index 40d772842..4d688b16a 100644 --- a/src/ports/uplink-websocket.ts +++ b/src/ports/uplink-websocket.ts @@ -36,6 +36,10 @@ 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; + function rawBytes(data: RawData): Buffer { if (Array.isArray(data)) return Buffer.concat(data); return Buffer.isBuffer(data) ? data : Buffer.from(data as ArrayBuffer); From 135e4f6557d355d1871bd0f0ee503f640b62e0cf Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 20:02:40 +0200 Subject: [PATCH 2/6] fix(ports): a worker terminates an uplink whose gateway stops answering pings (#941) --- src/ports/uplink-websocket.test.ts | 16 ++++++---------- src/ports/uplink-websocket.ts | 20 ++++++++++++++++++++ 2 files changed, 26 insertions(+), 10 deletions(-) diff --git a/src/ports/uplink-websocket.test.ts b/src/ports/uplink-websocket.test.ts index 9047d8578..c9b719701 100644 --- a/src/ports/uplink-websocket.test.ts +++ b/src/ports/uplink-websocket.test.ts @@ -231,11 +231,9 @@ async function startMuteGateway(): Promise<{ url: string; peers: WebSocket[]; pings: () => number; - nextPeer: (count: number) => Promise; }> { const peers: WebSocket[] = []; let pingCount = 0; - const waiting: Array<{ count: number; resolve: () => void }> = []; const wss = new WebSocketServer({ noServer: true, autoPong: false }); server = createServer(); server.on("upgrade", (request, socket, head) => { @@ -245,7 +243,6 @@ async function startMuteGateway(): Promise<{ peer.on("ping", () => { pingCount += 1; }); - for (const w of waiting) if (peers.length >= w.count) w.resolve(); }); }); await new Promise((resolve) => server?.listen(0, "127.0.0.1", resolve)); @@ -254,10 +251,6 @@ async function startMuteGateway(): Promise<{ url: `ws://127.0.0.1:${port}`, peers, pings: () => pingCount, - nextPeer: (count) => - peers.length >= count - ? Promise.resolve() - : new Promise((resolve) => waiting.push({ count, resolve })), }; } @@ -270,18 +263,21 @@ describe("WebSocketUplinkConnector liveness", () => { vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] }); const gateway = await startMuteGateway(); const clock = new FakeClock(); + let accepted = 0; const uplink = new GatewayUplink({ clock, connector: new WebSocketUplinkConnector(), url: gateway.url, token: "join", workerId: "wrk_1", - accept: () => {}, + accept: () => { + accepted += 1; + }, backoff: { initialMs: 1_000, maxMs: 1_000, multiplier: 1 }, random: () => 1, }); uplink.start(); - await gateway.nextPeer(1); + await vi.waitFor(() => expect(accepted).toBe(1)); const first = gateway.peers[0]; const firstClosed = new Promise((resolve) => first?.on("close", () => resolve())); @@ -291,7 +287,7 @@ describe("WebSocketUplinkConnector liveness", () => { await firstClosed; clock.advance(1_000); - await gateway.nextPeer(2); + await vi.waitFor(() => expect(accepted).toBe(2)); expect(gateway.peers).toHaveLength(2); await uplink.stop(); }); diff --git a/src/ports/uplink-websocket.ts b/src/ports/uplink-websocket.ts index 4d688b16a..f39ec1eb4 100644 --- a/src/ports/uplink-websocket.ts +++ b/src/ports/uplink-websocket.ts @@ -302,10 +302,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 { return new Promise((resolve, reject) => { const onOpen = () => { From 48a34a7c83616a3cb3835ba2fbabe9b65fd87af1 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 20:03:21 +0200 Subject: [PATCH 3/6] test: a healthy uplink stays open across several ping rounds (#941) --- src/ports/uplink-websocket.test.ts | 26 +++++++++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) diff --git a/src/ports/uplink-websocket.test.ts b/src/ports/uplink-websocket.test.ts index c9b719701..e0a7ce363 100644 --- a/src/ports/uplink-websocket.test.ts +++ b/src/ports/uplink-websocket.test.ts @@ -227,14 +227,14 @@ describe("WebSocketUplinkConnection binary messages", () => { }); }); -async function startMuteGateway(): Promise<{ +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: false }); + const wss = new WebSocketServer({ noServer: true, autoPong: answersPings }); server = createServer(); server.on("upgrade", (request, socket, head) => { wss.handleUpgrade(request, socket, head, (peer) => { @@ -242,6 +242,7 @@ async function startMuteGateway(): Promise<{ openConnections.push({ close: () => peer.terminate() }); peer.on("ping", () => { pingCount += 1; + peer.send("after-pong"); }); }); }); @@ -261,7 +262,7 @@ describe("WebSocketUplinkConnector liveness", () => { it("terminates an uplink whose gateway stops answering pings within two intervals, and GatewayUplink redials", async () => { vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] }); - const gateway = await startMuteGateway(); + const gateway = await startGateway(false); const clock = new FakeClock(); let accepted = 0; const uplink = new GatewayUplink({ @@ -291,4 +292,23 @@ describe("WebSocketUplinkConnector liveness", () => { 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); + }); }); From 0ce2fd5e7883a8d773497524f1300aab73ad2cd2 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 20:04:20 +0200 Subject: [PATCH 4/6] fix(ports): a stalled uplink handshake fails after 10s instead of the OS connect timeout (#941) --- src/ports/uplink-websocket.test.ts | 25 ++++++++++++++++++++++++- src/ports/uplink-websocket.ts | 13 +++++++++++++ 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/src/ports/uplink-websocket.test.ts b/src/ports/uplink-websocket.test.ts index e0a7ce363..5e1ebeb1f 100644 --- a/src/ports/uplink-websocket.test.ts +++ b/src/ports/uplink-websocket.test.ts @@ -14,7 +14,7 @@ * 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, WebSocketServer } from "ws"; @@ -27,6 +27,7 @@ import type { UplinkAuthOutcome, UplinkTransportConnection, } from "./uplink.js"; +import { UplinkError } from "./uplink.js"; import { UPLINK_PING_INTERVAL_MS, WebSocketUplinkConnector, @@ -312,3 +313,25 @@ describe("WebSocketUplinkConnector liveness", () => { 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((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((resolve) => setTimeout(() => resolve("still dialling"), 2_000)), + ]); + + expect(outcome).toBeInstanceOf(UplinkError); + } finally { + for (const socket of sockets) socket.destroy(); + silent.close(); + } + }); +}); diff --git a/src/ports/uplink-websocket.ts b/src/ports/uplink-websocket.ts index f39ec1eb4..9f9aefdf1 100644 --- a/src/ports/uplink-websocket.ts +++ b/src/ports/uplink-websocket.ts @@ -40,6 +40,10 @@ import { * 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. */ +export 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); @@ -279,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 { // `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, From bf87c71719e1a319533fda4f808860782e140c59 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 20:05:22 +0200 Subject: [PATCH 5/6] chore(ports): keep the handshake timeout constant private (#941) --- src/ports/uplink-websocket.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ports/uplink-websocket.ts b/src/ports/uplink-websocket.ts index 9f9aefdf1..acaeaa2c5 100644 --- a/src/ports/uplink-websocket.ts +++ b/src/ports/uplink-websocket.ts @@ -42,7 +42,7 @@ 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. */ -export const UPLINK_HANDSHAKE_TIMEOUT_MS = 10_000; +const UPLINK_HANDSHAKE_TIMEOUT_MS = 10_000; function rawBytes(data: RawData): Buffer { if (Array.isArray(data)) return Buffer.concat(data); From df01811a1125feb7652da055627754863b090659 Mon Sep 17 00:00:00 2001 From: Szymon Chmal Date: Fri, 23 Oct 2026 20:07:55 +0200 Subject: [PATCH 6/6] test: a silent gateway link fails on the closed assertion, not a timeout (#941) --- src/ports/uplink-websocket.test.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/ports/uplink-websocket.test.ts b/src/ports/uplink-websocket.test.ts index 5e1ebeb1f..a899ded35 100644 --- a/src/ports/uplink-websocket.test.ts +++ b/src/ports/uplink-websocket.test.ts @@ -286,7 +286,11 @@ describe("WebSocketUplinkConnector liveness", () => { await vi.advanceTimersByTimeAsync(UPLINK_PING_INTERVAL_MS); await vi.waitFor(() => expect(gateway.pings()).toBe(1)); await vi.advanceTimersByTimeAsync(UPLINK_PING_INTERVAL_MS); - await firstClosed; + const firstOutcome = await Promise.race([ + firstClosed.then(() => "closed"), + new Promise((resolve) => setTimeout(() => resolve("still open"), 1_000)), + ]); + expect(firstOutcome).toBe("closed"); clock.advance(1_000); await vi.waitFor(() => expect(accepted).toBe(2));