bin/daemon/lifetime.ts
Rama main · réplica de lectura del repositorio original.
/**
* The daemon's lifetime lock, and the socket generation it correlates with.
*
* `serve` holds an exclusive kernel lock on `<state>/daemon.lock` for as long as it runs: `flock(2)`
* on Linux and macOS, `LockFileEx` on one byte past the record on Windows. The kernel owns that
* lock's lifetime: it is released when the last descriptor referring to it closes, which includes
* every way the process can die. A daemon killed with SIGKILL, or taken by
* the OOM killer, therefore leaves no lock behind, and the next `serve` learns its predecessor is
* gone from the kernel's own refusal to hold the lock twice, not from a recorded pid it has to
* judge alive or dead.
*
* That is what makes stale-socket recovery safe on Linux and macOS. (A Windows pipe name needs no
* recovery: it exists only while a process holds an instance of it.) A unix socket's inode outlives its listener, so
* a crash leaves a published socket nobody is bound to, and a probe-then-unlink of it races with
* another starter publishing a live one. Holding the lock removes the race: publishing requires the
* lock, so while this process holds it nobody else can put a socket at the path. The one socket it
* may then remove is the one its dead predecessor published, which the lock record names by
* `(dev, ino)` and which nothing listens on any more. Any other socket, or an unreadable record, is
* refused and left alone.
*
* The record is written through the locked descriptor, so it is always the record of the file this
* process locked. It carries no secret: a pid, its start token, a start time and an inode number.
* The start token is the kernel's own start time for that pid ({@link processBirth}), so a reused
* pid never matches it.
*
* Two different questions use the two different facts. "May I serve?" is a liveness question, and
* the kernel answers it by holding the lock for at most one process. "Is the process answering
* this socket the one that holds the lock?" is an identity question, and the record answers it with
* the holder's pid and process-start token, which a client checks without touching the lock: a
* client that took the lock to test it, even for an instant, would make a legitimate start refuse.
*/
import { closeSync, fstatSync, ftruncateSync, lstatSync, openSync, readFileSync, writeSync } from "node:fs";
import { constants } from "node:fs";
import { createConnection } from "node:net";
import { processStartToken } from "@cotal-ai/core";
import { lockExclusiveNb, processStartTime } from "@cotal-ai/seat";
import { codeSpan, shellPath } from "@cotal-ai/workspace";
import { EndpointRefusal } from "./endpoint.js";
/** The socket a holder published, identified by the inode it is, not by its path alone. */
export interface SocketGeneration {
readonly path: string;
readonly dev: number;
readonly ino: number;
}
/**
* The kernel's start token for process `pid`, so a pid reused by a later process never matches:
* `/proc/<pid>/stat` starttime on Linux, the process table's start time on macOS, the creation
* time of the process object on Windows (the same value the native pipe open reads for its peer).
* Undefined when there is no such process.
*/
export function processBirth(pid: number): string | undefined {
return process.platform === "linux" ? processStartToken(pid) : processStartTime(pid);
}
/** What a holder writes through the locked descriptor. `socket` is absent until it publishes one. */
export interface LockRecord {
readonly v: 1;
readonly pid: number;
/** The holder's {@link processBirth} token, so a client can tell its pid from a reused one. */
readonly start?: string;
readonly startedAt: string;
readonly socket?: SocketGeneration;
}
export interface HeldLifetime {
/** The record the previous holder left, when it is readable and well-formed. */
readonly previous: LockRecord | undefined;
/** Stop preserving a predecessor whose socket was rejected, removed or already absent. */
discardPrevious(): void;
/** Record the socket this daemon is about to publish, before it publishes it. */
publishing(generation: SocketGeneration): void;
/** Clear the record and release the lock. A failed start may restore its unconsumed predecessor
* before closing the descriptor. Idempotent. */
release(opts?: { preservePrevious?: boolean }): void;
}
function parse(raw: string): LockRecord | undefined {
let o: unknown;
try { o = JSON.parse(raw); } catch { return undefined; }
if (!o || typeof o !== "object") return undefined;
const r = o as Record<string, unknown>;
if (r.v !== 1 || typeof r.pid !== "number" || !Number.isInteger(r.pid) || r.pid <= 0 || typeof r.startedAt !== "string") return undefined;
const start = typeof r.start === "string" ? { start: r.start } : {};
if (r.socket === undefined) return { v: 1, pid: r.pid, ...start, startedAt: r.startedAt };
const s = r.socket as Record<string, unknown>;
if (!s || typeof s !== "object" || typeof s.path !== "string" || typeof s.dev !== "number" || typeof s.ino !== "number") return undefined;
return { v: 1, pid: r.pid, ...start, startedAt: r.startedAt, socket: { path: s.path, dev: s.dev, ino: s.ino } };
}
/** The record at `path`, for a client deciding whether the process answering a socket is its
* holder. Undefined when there is none, when it is unreadable, or when it is not one of ours. */
export function readLifetimeRecord(path: string): LockRecord | undefined {
try {
const raw = readFileSync(path, "utf8");
return raw.trim() ? parse(raw) : undefined;
} catch { return undefined; }
}
/** Exclude new starters during uninstall without writing or clearing the holder's record. */
export function excludeLifetime(path: string): { release(): void } {
const fd = openSync(path, constants.O_RDWR | constants.O_CREAT, 0o600);
try {
if (!lockExclusiveNb(fd)) throw new EndpointRefusal("LOCK_HELD", "the daemon lifetime is still held; uninstall retained its source and receipt");
} catch (error) { closeSync(fd); throw error; }
let released = false;
return { release() { if (!released) { released = true; closeSync(fd); } } };
}
/**
* Take the lifetime lock at `path`, or refuse `LOCK_HELD`. The file is created 0600 if absent and
* is never replaced, so the descriptor every holder locks is the same inode.
*/
export function holdLifetime(path: string): HeldLifetime {
const fd = openSync(path, constants.O_RDWR | constants.O_CREAT, 0o600);
let held: boolean;
try {
held = lockExclusiveNb(fd);
} catch (e) {
closeSync(fd);
throw e;
}
if (!held) {
// Read the live holder's record only to name it. The lock, not the record, is the proof.
let holder: LockRecord | undefined;
try { holder = parse(readFileSync(path, "utf8")); } catch { /* naming is best effort */ }
closeSync(fd);
throw new EndpointRefusal(
"LOCK_HELD",
`${path} is held by a live process${holder ? ` (pid ${holder.pid}, started ${holder.startedAt})` : ""}: this daemon is already being served. Nothing was changed`,
);
}
// The lock is ours, so whatever the file holds was left by a holder the kernel has already
// released: a predecessor that exited without clearing it, which is exactly a crash.
let previous: LockRecord | undefined;
try {
const size = fstatSync(fd).size;
if (size > 0 && size <= 4096) previous = parse(readFileSync(path, "utf8"));
} catch { /* an unreadable record names no generation, so nothing is recovered */ }
const startedAt = new Date().toISOString();
const token = processBirth(process.pid);
const identity = { v: 1 as const, pid: process.pid, ...(token !== undefined ? { start: token } : {}), startedAt };
let released = false;
let recoverable = previous;
const write = (record: LockRecord): void => {
const bytes = `${JSON.stringify(record)}\n`;
ftruncateSync(fd, 0);
writeSync(fd, bytes, 0);
};
write(identity);
return {
previous,
discardPrevious() {
recoverable = undefined;
},
publishing(generation) {
recoverable = undefined;
write({ ...identity, socket: generation });
},
release(opts = {}) {
if (released) return;
released = true;
// A startup preflight failure has not consumed the dead socket. Keep its evidence under
// the lock so another attempt can recover it. Clean shutdown still leaves no generation.
try {
if (opts.preservePrevious && recoverable?.socket) write(recoverable);
else {
try { ftruncateSync(fd, 0); } catch { /* closing still releases the lock */ }
}
} finally { closeSync(fd); }
},
};
}
/** The generation of the socket already bound at `path`, read from the file itself. */
export function generationOf(path: string): SocketGeneration {
const st = lstatSync(path);
return { path, dev: st.dev, ino: st.ino };
}
/** How a connect to the unix socket at `path` ends: `ECONNREFUSED` when nothing listens on it,
* `answered` when a process accepts it, or the error code the kernel gave. Nothing is written. */
function connectOutcome(path: string): Promise<string> {
return new Promise((resolve) => {
const s = createConnection(path);
s.once("connect", () => { s.destroy(); resolve("answered"); });
s.once("error", (e: NodeJS.ErrnoException) => resolve(e.code ?? e.message));
});
}
/**
* Decide what to do with whatever sits at the public socket path, holding the lifetime lock.
*
* `clear` means nothing is there. `recover` means the socket there is the exact inode the dead
* predecessor's record names and nothing listens on it, so this daemon may remove it. Anything else
* refuses: a non-socket entry, a socket this home's last daemon did not publish, or one a process
* answers on, is reported and left in place.
*/
export async function claimSocketPath(path: string, previous: SocketGeneration | undefined): Promise<"clear" | "recover"> {
let st;
try {
st = lstatSync(path);
} catch (e) {
if ((e as NodeJS.ErrnoException).code === "ENOENT") return "clear";
throw e;
}
if (!st.isSocket()) throw new EndpointRefusal("SOCKET_NOT_SOCKET", `${path} exists and is not a socket; nothing was removed`);
if (previous && previous.path === path && previous.dev === st.dev && previous.ino === st.ino) {
// A filesystem can give a new file the inode number of one it freed, so equal numbers alone do
// not prove this is the dead predecessor's socket. Its listener died with it, so the kernel
// refuses a connect; a socket anything still answers on belongs to a live process.
const outcome = await connectOutcome(path);
if (outcome === "ECONNREFUSED") return "recover";
if (outcome === "answered")
throw new EndpointRefusal(
"ALREADY_RUNNING",
`${path} has the device and inode numbers this daemon home's last daemon published, but a connect to it was answered: another process is listening at that path, and serve does not start until that process releases it; nothing was removed`,
`find that process with \`ss -xlp\` or ${codeSpan(`fuser ${shellPath(path)}`)}, and stop it; then run \`cotal daemon serve\` again`,
);
throw new EndpointRefusal(
"ALREADY_RUNNING",
`${path} has the device and inode numbers this daemon home's last daemon published, but a connect to it failed with ${outcome}, so it is not that daemon's dead socket; nothing was removed`,
);
}
// A rejected socket claim clears the record as a clean stop does, so the next attempt does
// not reuse evidence that failed this check.
throw new EndpointRefusal(
"ALREADY_RUNNING",
previous
? `${path} already exists (uid ${st.uid}) and is not the socket this daemon home's last daemon published; nothing was removed`
: `${path} already exists (uid ${st.uid}), and this daemon home's lock record names no socket to compare it with: a clean stop or a refused socket check clears that record; nothing was removed`,
);
}