From 7aed50ae8b67eff1ce407aafa66d632b44e4b406 Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Tue, 3 Nov 2026 17:48:14 +0100 Subject: [PATCH] fix(chord): clear stale worker lock files on macOS and the BSDs A worker killed by a signal left its lock's socket file, and every later start of its address failed closed on it, retry too. The workers component now removes a lock socket file nobody accepts on before each start and after a child's exit by a signal, in the address's queue. processLock also treats ENOTSOCK as stale instead of retrying until its context ended. --- packages/chord/CHANGELOG.md | 5 + packages/chord/docs/decisions.md | 9 ++ packages/chord/docs/design.md | 3 +- packages/chord/src/node/workers/README.md | 3 + packages/chord/src/node/workers/lock.ts | 60 ++++++++++-- packages/chord/src/node/workers/workers.ts | 21 ++++- packages/chord/test/node/workers/lock.test.ts | 91 +++++++++++++++---- packages/chord/test/node/workers/main.test.ts | 4 +- .../chord/test/node/workers/workers.test.ts | 39 +++++++- 9 files changed, 205 insertions(+), 30 deletions(-) diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index 02257c937..c3e3641fb 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -24,6 +24,11 @@ - Chord v2's `host.load` and `host.reload` throw for a plugin name that is no npm package name, before they change anything (it was listed in the report's `failed`). - Chord v2 plugins import each other by relative path within a root (copied into the importer's script) and across repositories by package name through the package's `exports`, pinned in `dependencies`; slice 10's imports by plugin name (`@app/notes/contract`) are removed, and a relative path leaving the root is a build error. +### Fixed + +- Chord v2's workers on macOS and the BSDs: a worker killed by a signal (the grace's or the start timer's SIGKILL) left its lock's socket file, and every later start of its address failed with `... is a stale lock file`, `retry` too, until the file was removed by hand. The workers component now removes a lock socket file nobody accepts on before each start and after a child's exit by a signal. +- Chord v2's `processLock` on macOS spun until its context ended when a regular file sat at the lock's path (`ENOTSOCK`); it now fails closed naming the file, as for a socket file nobody listens on. + ## [1.5.0] - 2026-11-02 ### Breaking Changes diff --git a/packages/chord/docs/decisions.md b/packages/chord/docs/decisions.md index 73d036cc6..489105045 100644 --- a/packages/chord/docs/decisions.md +++ b/packages/chord/docs/decisions.md @@ -2084,6 +2084,15 @@ validation and copies the v2 kernel added around Delta. is trusted (a plugin's own file). (2) The worker lock on macOS and the BSDs fails closed on a stale lock file, with an error naming it (no takeover; Linux and Android use abstract sockets and are unaffected). (3) A route's overhead: Opus's two trims, measured again after the `fetch` rework, and the numbers brought to the owner. +- **Stale lock files are cleared by the starter (owner, 2026-11-03: "do both"; refines point (2) of "Slice 10 fix + round, three small points"):** on macOS and the BSDs a worker killed by a signal (the grace's SIGKILL, the start + timer, a dev supervisor) runs no exit hooks and leaves its lock's socket file; every later start of that address then + failed closed on it, also `retry`, until someone removed the file by hand. Takers still fail closed (two takers + clearing could race); the workers component, the one process that starts an address's workers, removes a socket file + nobody accepts on (`clearStaleLock`) before each start's message and after a child's exit by a signal, both in the + address's queue, so never two clears of one name at once. A live holder (an older worker still exiting) accepts and + is kept. One connect per start on macOS and the BSDs, nothing on Linux, Android and Windows. A taker also treats + `ENOTSOCK` (a non-socket file at the path on macOS) as stale; before it retried until its context ended. - **Slice 10: a route's overhead accepted; no header accessor (owner, 2026-10-31: "1 yes, 2 no"):** after the `fetch` rework and the Node fast path a route costs +20-21 µs over a plain Node handler on the standard bench, +12-13 µs warmed (before the rework +4.5 µs): the request's scope (2.1 µs), URL parsing, admission and routing, the baseline's headers, diff --git a/packages/chord/docs/design.md b/packages/chord/docs/design.md index 4af2c11dd..e8de3a77e 100644 --- a/packages/chord/docs/design.md +++ b/packages/chord/docs/design.md @@ -2283,7 +2283,8 @@ connect only (Q4). `workers.state` (a JSON watchable per address key), `stop`, ` `workerMain(fn, { hostPackages, redact, onError })` is the child's side (`fn({ address, start, parent, plugins, onClose }, ctx)`, `plugins` a `WorkerPlugins` evaluating the shipped scripts, closers in reverse order). One live worker per address (`processLock`: an abstract socket on Linux, a named pipe on Windows, a socket file elsewhere that -fails closed when stale; a start waits for a draining old holder without polling). +fails closed when stale, the stale file cleared by the component before a start and after a child's exit by a signal, +decision "Stale lock files are cleared by the starter"; a start waits for a draining old holder without polling). - **Keep-alive** (D9, R3, Q5): the child reports when `host.activity`'s references, jobs and lingering objects and the app's `busy` counts flip between all zero and not (local calls do not count; the first state with `ready`, which diff --git a/packages/chord/src/node/workers/README.md b/packages/chord/src/node/workers/README.md index 15a153cf4..9ba953aa2 100644 --- a/packages/chord/src/node/workers/README.md +++ b/packages/chord/src/node/workers/README.md @@ -82,6 +82,9 @@ workerMain(async ({ address, start, parent, plugins, onClose }, ctx) => { `shutdown`, the parent's disconnect, `SIGTERM` and `SIGINT` abort its context and run the closers registered so far, also during startup. An uncaught error or a throwing function sends `fatal` with `redact(error)` and exits 1. - `processLock(name, ctx)`: the lock (an abstract socket on Linux, a named pipe on Windows, a socket file elsewhere). + A taker fails closed on a socket file nobody accepts on; the component clears one (a worker killed by a signal left + it) before each start and after a child's exit by a signal, in the address's queue (decision "Stale lock files are + cleared by the starter"). ## Examples diff --git a/packages/chord/src/node/workers/lock.ts b/packages/chord/src/node/workers/lock.ts index 3c1199dd9..e76846532 100644 --- a/packages/chord/src/node/workers/lock.ts +++ b/packages/chord/src/node/workers/lock.ts @@ -1,13 +1,14 @@ // Process locks (slice 10 §3): one live worker per address. A lock is a listening socket the // kernel owns for the holder's lifetime (an abstract socket on Linux and Android, a named pipe on Windows, a socket file -// elsewhere), so there is no race between two would-be holders; a socket file a crashed holder left fails closed (the -// error names it: a takeover would race another taker). A child finding the lock held waits +// elsewhere), so there is no race between two would-be holders; a socket file a crashed holder left fails closed for a +// taker (the error names it: a takeover would race another taker), and only the one process that starts the holders of +// a name clears it (decision "Stale lock files are cleared by the starter"). A child finding the lock held waits // without polling: it connects to the holder's socket, which the holder never answers, and tries again when that // connection closes (the holder exited). import { createHash } from "node:crypto"; import { unlinkSync } from "node:fs"; -import { lstat, mkdir } from "node:fs/promises"; +import { lstat, mkdir, unlink } from "node:fs/promises"; import { connect, createServer, type Server, type Socket } from "node:net"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -24,13 +25,55 @@ export type LockAddress = { readonly kind: "abstract" | "pipe" | "file"; readonl * Throws what listening throws besides "in use", and for a stale socket file (macOS, the BSDs) an error naming it. */ export async function processLock(name: string, context: Context): Promise { + return lockAt(await lockAddress(name), context); +} + +/** + * Removes the socket file of the lock `name` when nobody accepts on it (a holder killed without its exit hooks left + * it); resolves with whether it did. Abstract sockets and named pipes end with their holder: nothing to clear. Only one + * process clears a name, one clear at a time: two clearers could both find it stale, one remove it, a new holder take + * it, and the other remove the new holder's file. + */ +export async function clearStaleLock(name: string, context: Context): Promise { + return clearStaleLockAt(await lockAddress(name), context); +} + +/** Where the lock `name` lives on this platform. */ +async function lockAddress(name: string): Promise { const key = createHash("sha256").update(name).digest("hex").slice(0, 32); - if (process.platform === "win32") - return lockAt({ kind: "pipe", address: `\\\\.\\pipe\\chord-lock-${key}` }, context); + if (process.platform === "win32") return { kind: "pipe", address: `\\\\.\\pipe\\chord-lock-${key}` }; if (process.platform === "linux" || process.platform === "android") { - return lockAt({ kind: "abstract", address: `\0chord-lock-${key}` }, context); + return { kind: "abstract", address: `\0chord-lock-${key}` }; } - return lockAt({ kind: "file", address: await socketPath(`chord-lock-${key}.sock`) }, context); + return { kind: "file", address: await socketPath(`chord-lock-${key}.sock`) }; +} + +/** `clearStaleLock` at `lock`: a socket file nobody accepts on is removed. */ +export async function clearStaleLockAt(lock: LockAddress, context: Context): Promise { + context.abortSignal?.throwIfAborted(); + if (lock.kind !== "file" || !(await nobodyAccepts(lock.address))) return false; + try { + await unlink(lock.address); + return true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; + throw error; + } +} + +/** A connect's errors that mean nobody accepts on the path: a socket file nobody listens on, or no socket there. */ +const NOBODY_ACCEPTS = new Set(["ECONNREFUSED", "ENOTSOCK"]); + +/** Whether the path `address` exists and nobody accepts on it. */ +function nobodyAccepts(address: string): Promise { + return new Promise((resolve) => { + const socket = connect(address); + socket.once("connect", () => { + socket.destroy(); + resolve(false); + }); + socket.once("error", (error: NodeJS.ErrnoException) => resolve(NOBODY_ACCEPTS.has(error.code ?? ""))); + }); } /** @@ -115,7 +158,8 @@ function waitWhileHeld(address: string, context: Context): Promise { connected = true; }); socket.on("error", (error: NodeJS.ErrnoException) => { - if (!connected && error.code === "ECONNREFUSED") settle(false); + // A regular file at the path is ENOTSOCK on macOS (ECONNREFUSED on Linux); retrying would spin. + if (!connected && NOBODY_ACCEPTS.has(error.code ?? "")) settle(false); }); socket.once("close", () => settle(true)); }); diff --git a/packages/chord/src/node/workers/workers.ts b/packages/chord/src/node/workers/workers.ts index c1e1f7f13..66c3eef04 100644 --- a/packages/chord/src/node/workers/workers.ts +++ b/packages/chord/src/node/workers/workers.ts @@ -15,6 +15,7 @@ import type { Watchable } from "../../watch.ts"; import type { Plugins, RolloutHost, RolloutStep } from "../plugins/plugins.ts"; import type { PluginScript } from "../plugins/runtime.ts"; import { type CrossedError, type CrossedReport, type FromWorker, uncrossed, uncrossedReport } from "./control.ts"; +import { clearStaleLock } from "./lock.ts"; import { type Exit, WorkerProcess } from "./process.ts"; /** What `createWorkers` takes. `kind`, `entry`, `execArgv` and `env` are bootstrap; the functions are called per use. */ @@ -167,6 +168,8 @@ type Child = { ended: Ended | undefined; /** The grace timer of a stop or of a lost link. */ grace: ReturnType | undefined; + /** The lock its start message named; cleared after its exit when it was killed holding it. */ + lock: string | undefined; }; /** One address: its operations in a queue, its child, its failure. Kept while it has any of them. */ @@ -270,7 +273,10 @@ class WorkersComponent implements Workers { if (!this.#closing.signal.aborted) this.#closing.abort(closed()); clearInterval(this.#idleTimer); this.#idleTimer = undefined; - const stops = [...this.#addresses.values()].map((routed) => this.#enqueue(routed, () => this.#stop(routed))); + // Each stop, then what its exit queued (the clear of a killed child's lock file). + const stops = [...this.#addresses.values()].map((routed) => + this.#enqueue(routed, () => this.#stop(routed)).then(() => routed.queue), + ); await awaitWithContext(Promise.all(stops), context); } @@ -365,6 +371,12 @@ class WorkersComponent implements Workers { const scripts = await this.#options.plugins.scripts(this.#options.kind, context); child.shipped = keyOf(scripts); const lock = this.#options.lock(address); + // A worker killed without its exit hooks (SIGKILL), maybe by an earlier server, left its socket file, which fails + // its successors closed. The component is the one clearer of its addresses' locks, in the address's queue: one + // clear at a time (decision "Stale lock files are cleared by the starter"). A live holder (an old worker still + // exiting) is kept and waited for. + await clearStaleLock(lock, context); + child.lock = lock; child.process.send({ kind: "start", address, start: data, scripts, lock }); await awaitWithContext(child.started.promise, this.#closingContext); return { link: child.process.link, principal: this.#options.principal(address), trusted: true }; @@ -414,6 +426,7 @@ class WorkersComponent implements Workers { reloaded: undefined, ended: undefined, grace: undefined, + lock: undefined, }; // Nobody may wait for them. child.started.promise.catch(() => {}); @@ -484,6 +497,12 @@ class WorkersComponent implements Workers { child.started.reject(uncrossed(error)); child.ready.reject(uncrossed(error)); this.#ended(child); + const lock = child.lock; + // A child a signal ended (SIGKILL) ran no exit hooks: its socket file would stay until the address's next start. + // Cleared in the queue, like a start's clear; a failure here is the next start's to report (it clears again). + if (lock !== undefined && exit.signal !== null) { + void this.#enqueue(routed, () => clearStaleLock(lock, BACKGROUND_CONTEXT)).catch(() => {}); + } if (routed.child !== child) return; routed.child = undefined; if (!clean && !this.#closing.signal.aborted) { diff --git a/packages/chord/test/node/workers/lock.test.ts b/packages/chord/test/node/workers/lock.test.ts index eb53ebe5d..796165064 100644 --- a/packages/chord/test/node/workers/lock.test.ts +++ b/packages/chord/test/node/workers/lock.test.ts @@ -1,13 +1,15 @@ // Tests of process locks (slice 10 §3): one holder at a time, a waiter taking the lock when the holder releases it -// (told by its connection's close, no polling), a waiter's abort. +// (told by its connection's close, no polling), a waiter's abort, socket files failing closed when stale, and clearing +// a stale one. +import { spawnSync } from "node:child_process"; import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { describe, expect, it } from "vitest"; import { BACKGROUND_CONTEXT as context, withAbortSignal } from "../../../src/context/index.ts"; import { processLock } from "../../../src/node/workers/index.ts"; -import { lockAt } from "../../../src/node/workers/lock.ts"; +import { clearStaleLockAt, lockAt } from "../../../src/node/workers/lock.ts"; describe("processLock", () => { it("makes a second taker wait until the holder releases, then takes it", async () => { @@ -41,22 +43,79 @@ describe("processLock", () => { describe("socket file locks (macOS, the BSDs)", () => { it("fail closed on a stale lock file, naming it: two takers never both take it over (Sol)", async () => { - const dir = await mkdtemp(join(tmpdir(), "chord-lock-file-")); - const address = join(dir, "lock.sock"); - // What a crashed holder leaves: a path nobody accepts on. - await writeFile(address, ""); - const takers = [lockAt({ kind: "file", address }, context), lockAt({ kind: "file", address }, context)]; - for (const taker of takers) await expect(taker).rejects.toThrow(`lock: ${address} is a stale lock file`); - await rm(dir, { recursive: true, force: true }); + await inTemporaryDirectory(async (dir) => { + // What a holder killed without its exit hooks leaves (a socket nobody listens on), and a regular file (ENOTSOCK + // on macOS: it once made the taker spin until its context ended). + for (const [name, leave] of [ + ["killed.sock", leaveSocketFile], + ["regular.sock", (path: string) => writeFile(path, "")], + ] as const) { + const address = join(dir, name); + await leave(address); + const takers = [lockAt({ kind: "file", address }, context), lockAt({ kind: "file", address }, context)]; + for (const taker of takers) await expect(taker).rejects.toThrow(`lock: ${address} is a stale lock file`); + } + }); }); it("are taken, waited for and released like the others", async () => { - const dir = await mkdtemp(join(tmpdir(), "chord-lock-file-")); - const address = join(dir, "lock.sock"); - const first = await lockAt({ kind: "file", address }, context); - const second = lockAt({ kind: "file", address }, context); - await first.release(); - await (await second).release(); - await rm(dir, { recursive: true, force: true }); + await inTemporaryDirectory(async (dir) => { + const address = join(dir, "lock.sock"); + const first = await lockAt({ kind: "file", address }, context); + const second = lockAt({ kind: "file", address }, context); + await first.release(); + await (await second).release(); + }); }); }); + +describe("clearStaleLockAt", () => { + it("removes a socket file nobody accepts on: the lock can be taken again", async () => { + await inTemporaryDirectory(async (dir) => { + const address = join(dir, "lock.sock"); + await leaveSocketFile(address); + expect(await clearStaleLockAt({ kind: "file", address }, context)).toBe(true); + await (await lockAt({ kind: "file", address }, context)).release(); + }); + }); + + it("keeps a held lock's file, and does nothing without a file or for abstract sockets and pipes", async () => { + await inTemporaryDirectory(async (dir) => { + const address = join(dir, "lock.sock"); + expect(await clearStaleLockAt({ kind: "file", address }, context)).toBe(false); + const held = await lockAt({ kind: "file", address }, context); + expect(await clearStaleLockAt({ kind: "file", address }, context)).toBe(false); + // Still held: a second taker waits. + let taken = false; + const second = lockAt({ kind: "file", address }, context).then((lock) => { + taken = true; + return lock; + }); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(taken).toBe(false); + await held.release(); + await (await second).release(); + expect(await clearStaleLockAt({ kind: "abstract", address: "\0chord-test-unused" }, context)).toBe(false); + expect(await clearStaleLockAt({ kind: "pipe", address: "\\\\.\\pipe\\chord-test-unused" }, context)).toBe( + false, + ); + }); + }); +}); + +/** Runs `fn` in a fresh temporary directory, removed afterwards whatever `fn` did. */ +async function inTemporaryDirectory(fn: (dir: string) => Promise): Promise { + const dir = await mkdtemp(join(tmpdir(), "chord-lock-file-")); + try { + await fn(dir); + } finally { + await rm(dir, { recursive: true, force: true }); + } +} + +/** Leaves a socket file at `path` as a holder killed without its exit hooks does: a process listening there, SIGKILLed. */ +async function leaveSocketFile(path: string): Promise { + const script = `require("node:net").createServer().listen(${JSON.stringify(path)}, () => process.kill(process.pid, "SIGKILL"))`; + const result = spawnSync(process.execPath, ["-e", script]); + expect(result.signal).toBe("SIGKILL"); +} diff --git a/packages/chord/test/node/workers/main.test.ts b/packages/chord/test/node/workers/main.test.ts index a85a82521..b4ef4c551 100644 --- a/packages/chord/test/node/workers/main.test.ts +++ b/packages/chord/test/node/workers/main.test.ts @@ -51,7 +51,9 @@ describe("the worker's side", () => { await until(() => control.some(({ kind }) => kind === "ready")); expect(control.map(({ kind }) => kind)).toEqual(["started", "ready"]); expect(control[1]).toMatchObject({ idle: true, report: { plugins: { added: [] }, failed: [] } }); - child.kill(); + // Shut down, not killed: a SIGKILL would leave its lock's socket file on macOS and the BSDs. + child.send({ kind: "shutdown" }); + await child.exited; }); it("shuts down during its startup: aborts it, runs the closers registered so far in reverse order, exits 0", async () => { diff --git a/packages/chord/test/node/workers/workers.test.ts b/packages/chord/test/node/workers/workers.test.ts index 983d4e7a2..fec07c969 100644 --- a/packages/chord/test/node/workers/workers.test.ts +++ b/packages/chord/test/node/workers/workers.test.ts @@ -1,11 +1,14 @@ // Tests of the workers component (slice 10 §3, §4): starts on demand (the link made at the fork, `started` on // `connect`, the start timer covering only the start), failures that stay (a crash, a `fatal`, a lost link, a failed // start: no fork for a new hold, holders `failed { waiting }`) until `retry` or a rollout, idle stops and `busy`, `stop` -// and its suspension, the per-address queue, `close`, the rollout step, and nothing kept of 1,000 addresses. +// and its suspension, the per-address queue, `close`, the rollout step, stale lock files cleared, and nothing kept of +// 1,000 addresses. +import { spawnSync } from "node:child_process"; import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { fileURLToPath } from "node:url"; import { describe, expect, it, vi } from "vitest"; import { withAbortSignal } from "../../../src/context/index.ts"; import { createHost } from "../../../src/host/index.ts"; @@ -23,6 +26,9 @@ function listed(hosts: readonly RolloutHost[]) { ]); } +/** The process lock module, for a process that takes a lock and is killed holding it. */ +const LOCK_MODULE = fileURLToPath(new URL("../../../src/node/workers/lock.ts", import.meta.url)); + /** The state entry of `id`. */ const stateOf = (state: unknown, id: string) => (state as Record)[`session|${id}`]; @@ -112,7 +118,7 @@ describe("failures stay until retry or a rollout (R2)", () => { notes.release(); }); - it("a link lost while its child lives (a blocked event loop past the heartbeat): killed after the grace, failed", async () => { + it("a link lost while its child lives (a blocked event loop past the heartbeat): killed after the grace, failed; retry starts it past the lock file the kill left", async () => { const { server, workers, starts, errors } = await fleet({ heartbeat: { ping: 100, close: 300 } }); const notes = await server.get(Notes, { id: "S1" }, context); const control = await server.get(Control, { id: "S1" }, context); @@ -124,9 +130,15 @@ describe("failures stay until retry or a rollout (R2)", () => { error: { message: "worker: link lost" }, }); expect(starts).toEqual(["session|S1"]); + // SIGKILL skipped the child's exit hooks: on macOS and the BSDs its lock's socket file was left, and the component + // cleared it after the exit (else the new child would fail closed on it). + await workers.retry(Session({ id: "S1" }), context); + await until(() => notes.state.value.kind === "live", 10_000); + expect(stateOf(workers.state.value, "S1")?.status).not.toBe("failed"); + expect(starts).toEqual(["session|S1", "session|S1"]); notes.release(); control.release(); - }, 20_000); + }, 30_000); it("an exit before started fails the start; a worker that never connects is killed at the start timer", async () => { const { server, workers, behaviours, errors } = await fleet(); @@ -177,6 +189,27 @@ describe("failures stay until retry or a rollout (R2)", () => { }); }); +describe('stale lock files (macOS, the BSDs; decision "Stale lock files are cleared by the starter")', () => { + it("a start clears the lock file a killed worker of an earlier server left: the worker runs", async () => { + const { server, workers, root } = await fleet(); + // What an earlier server's worker SIGKILLed while holding the lock leaves (nothing on Linux: an abstract socket). + const name = `chord-test:${root}:${Session({ id: "S1" }).key}`; + const script = `const { processLock } = await import(${JSON.stringify(LOCK_MODULE)}); await processLock(${JSON.stringify(name)}, {}); process.kill(process.pid, "SIGKILL");`; + const killed = spawnSync(process.execPath, [ + "--experimental-strip-types", + "--no-warnings", + "--input-type=module", + "-e", + script, + ]); + expect(killed.signal).toBe("SIGKILL"); + const notes = server.hold(Notes, { id: "S1" }, context); + await until(() => notes.state.value.kind === "live"); + expect(stateOf(workers.state.value, "S1")?.status).not.toBe("failed"); + notes.release(); + }); +}); + describe("keep-alive (D9, R3, Q5)", () => { it("stops a worker idle since idleMs; a hold keeps one running; busy keeps it too", async () => { const { server, workers } = await fleet({ idleMs: 200 });