diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 6705867e6..32af9a067 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- A link that is already closed when a host accepts or routes to it is no longer kept as a connection: before, a route that returned a dead link (a child process that exited before its link was wrapped) made every later call to that host name fail with `unavailable` until restart, and leaked the connection's ping timer. + ## [1.2.0] - 2026-10-12 ### Breaking Changes diff --git a/packages/chord/node_modules b/packages/chord/node_modules new file mode 120000 index 000000000..126c7e284 --- /dev/null +++ b/packages/chord/node_modules @@ -0,0 +1 @@ +/home/badlogic/pi-private/packages/chord//node_modules \ No newline at end of file diff --git a/packages/chord/src/connection.ts b/packages/chord/src/connection.ts index 5a9907fdd..594f9357f 100644 --- a/packages/chord/src/connection.ts +++ b/packages/chord/src/connection.ts @@ -254,6 +254,8 @@ export class Connection implements RemoteChannel { this.send({ t: "hello", name: host.name, kind: host.kind, epoch: host.current.id }); const interval = config.pingInterval ?? 15_000; const timeout = config.pingTimeout ?? 45_000; + // A link that was already closed closed this connection above; it needs no keepalive. + if (link.closed) return; let n = 0; this.#pinger = setInterval(() => { if (Date.now() - this.#lastSeen > timeout) link.close("ping timeout"); diff --git a/packages/chord/src/host.ts b/packages/chord/src/host.ts index 93d753298..fb0ade01b 100644 --- a/packages/chord/src/host.ts +++ b/packages/chord/src/host.ts @@ -750,7 +750,8 @@ export class HostImpl implements Host, HostCore { #addConnection(link: Link, config: ConstructorParameters[2]): Connection { const connection = new Connection(this, link, config); - this.connections.add(connection); + // A link that was already closed closed the connection in its constructor: keeping it would route to it forever. + if (connection.open) this.connections.add(connection); return connection; } @@ -800,7 +801,10 @@ export class HostImpl implements Host, HostCore { #childNamed(name: string): Connection | undefined { return [...this.connections].find( (connection) => - connection.role === "downstream" && connection.peer.kind === "host" && connection.peer.name === name, + connection.open && + connection.role === "downstream" && + connection.peer.kind === "host" && + connection.peer.name === name, ); } diff --git a/packages/chord/test/link.test.ts b/packages/chord/test/link.test.ts index 6b4c09170..709c62c0b 100644 --- a/packages/chord/test/link.test.ts +++ b/packages/chord/test/link.test.ts @@ -248,6 +248,29 @@ describe("links: relay, stamping, lifetime", () => { assert.deepEqual(childResult?.t === "result" ? childResult.value : undefined, { worker: "w9" }); }); + it("a route that returns an already closed link does not poison the host name", async () => { + const server = createHost({ name: "server", onError: () => {} }); + let attempt = 0; + server.route("session:", { + authorize: () => true, + resolve: async (name) => { + attempt++; + const [down, up] = memoryLinkPair(); + const worker = createHost({ name, upstream: up, onError: () => {} }); + serveCounter(worker); + // The first worker dies before the server wraps its link. + if (attempt === 1) down.close("worker exited"); + return down; + }, + }); + const first = server.get(Counter, { name: "a" }, { at: "session:S1" }); + await assert.rejects(first.add(1), (error: Error) => /closed/.test(error.message)); + // The next call resolves the route again and reaches the new worker. + const second = server.get(Counter, { name: "a" }, { at: "session:S1" }); + assert.equal(await second.add(2), 2); + assert.equal(attempt, 2); + }); + it("authorizes before resolving and again after", async () => { const c = cluster(); c.allow = (principal) => (principal as { guest: string }).guest !== "blocked";