mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(chord): never keep a connection whose link was already closed
A route that returned a dead link (a child that exited before its link was wrapped) left a closed connection in the host, so every later call to that host name failed with unavailable until restart. Such a connection is not added, closed children are skipped when routing by name, and no ping timer is started for it.
This commit is contained in:
parent
d547949184
commit
7f022043f5
5 changed files with 36 additions and 2 deletions
|
|
@ -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
|
||||
|
|
|
|||
1
packages/chord/node_modules
Symbolic link
1
packages/chord/node_modules
Symbolic link
|
|
@ -0,0 +1 @@
|
|||
/home/badlogic/pi-private/packages/chord//node_modules
|
||||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -750,7 +750,8 @@ export class HostImpl implements Host, HostCore {
|
|||
|
||||
#addConnection(link: Link, config: ConstructorParameters<typeof Connection>[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,
|
||||
);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue