diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fe845c..37225b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,16 @@ when metadata is missing or malformed, and revalidates it while holding the per-name creation lock before removal. +### Generation-guarded gc cleanup and respawn + +- Residual sweep and permanent respawn now compare the generation observed by + gc with the current metadata while holding the per-name creation lock. + Cleanup is skipped when another process replaced or edited the record after + gc's snapshot, preventing a stale pass from unlinking or respawning over the + replacement. Legacy generation-less records use exact metadata equality as a + conservative fallback. Bundled CLI respawns inherit the already-held + creation lock until the replacement publishes its socket. + ### Attach-only CLI policy - `pty attach --no-restart ` attaches only to a currently running diff --git a/src/cli.ts b/src/cli.ts index df6d311..ae6684a 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -20,6 +20,7 @@ import { validateName, validateDisplayName, acquireLock, + isLockOwnedByPid, releaseLock, updateTags, setDisplayName, @@ -1546,7 +1547,13 @@ async function cmdRun( process.exit(1); } - if (!acquireLock(name)) { + const delegatedOwner = Number(process.env.PTY_CREATION_LOCK_OWNER_PID); + const inheritedCreationLock = + Number.isSafeInteger(delegatedOwner) && + isLockOwnedByPid(name, delegatedOwner); + // This is a one-hop control value for the CLI process, not session env. + delete process.env.PTY_CREATION_LOCK_OWNER_PID; + if (!inheritedCreationLock && !acquireLock(name)) { console.error( `Session "${name}" is being created by another process. Try again.` ); @@ -1578,7 +1585,7 @@ async function cmdRun( ...(extraEnvOpt && Object.keys(extraEnvOpt).length > 0 ? { extraEnv: extraEnvOpt } : {}), }); } finally { - releaseLock(name); + if (!inheritedCreationLock) releaseLock(name); } console.log(`Session "${name}" created.`); diff --git a/src/sessions.ts b/src/sessions.ts index ed1708c..9d2bdb2 100644 --- a/src/sessions.ts +++ b/src/sessions.ts @@ -449,6 +449,50 @@ export async function cleanupRawCandidateGuarded( } } +function metadataMatchesObservation( + observed: SessionMetadata, + current: SessionMetadata, +): boolean { + if (observed.generation !== undefined || current.generation !== undefined) { + return observed.generation !== undefined && + observed.generation === current.generation; + } + // Legacy records have no generation token. Exact structural equality is a + // conservative fallback: any intervening tag/launch/exit update makes the + // observation stale and suppresses cleanup. + return JSON.stringify(observed) === JSON.stringify(current); +} + +function cleanupAllWhileLocked(name: string): void { + cleanupSocket(name); + try { + fs.unlinkSync(getMetadataPath(name)); + } catch {} + try { + fs.unlinkSync(getEventsPath(name)); + } catch {} +} + +/** @internal Generation-CAS cleanup primitive; not part of client-api. */ +export async function cleanupObservedSession( + session: SessionInfo, +): Promise { + if (!session.metadata || !acquireLock(session.name)) return false; + try { + const current = readMetadata(session.name); + if ( + !current || + !metadataMatchesObservation(session.metadata, current) + ) { + return false; + } + cleanupAllWhileLocked(session.name); + return true; + } finally { + releaseLock(session.name); + } +} + /** Return one bounded, read-only observation of the session registry. * * This function deliberately performs no lifecycle work: it does not create @@ -981,8 +1025,9 @@ export async function gc( } try { - await respawnPermanent(s.name, s.metadata!, decision.newBookkeeping); - respawned.push({ name: s.name, ptyfileReread }); + if (await respawnPermanent(s.name, s.metadata!, decision.newBookkeeping)) { + respawned.push({ name: s.name, ptyfileReread }); + } } catch (err: any) { respawnFailed.push({ name: s.name, error: err?.message ?? String(err) }); } @@ -1007,8 +1052,11 @@ export async function gc( kept.push(s.name); continue; } - if (!dryRun) cleanupAll(s.name); - removed.push(s.name); + if (dryRun) { + removed.push(s.name); + } else if (await cleanupObservedSession(s)) { + removed.push(s.name); + } } return { @@ -1208,12 +1256,14 @@ function classifyFlapping( * * Lazy-imports `spawn.ts` so the `sessions.ts ↔ spawn.ts` cycle doesn't * bite at module-init time. After spawn, appends a `session_respawn` - * event to the session's event log so consumers see the restart. */ -async function respawnPermanent( + * event to the session's event log so consumers see the restart. + * + * @internal Generation-CAS primitive; not part of client-api. */ +export async function respawnPermanent( name: string, metadata: SessionMetadata, bookkeepingOverlay: Record = {}, -): Promise { +): Promise { let command = metadata.command; let args = metadata.args; let displayCommand = metadata.displayCommand; @@ -1262,22 +1312,36 @@ async function respawnPermanent( delete tags["strategy.status"]; } - // Wipe stale socket/pid/events before respawn so spawnDaemon doesn't - // trip over leftovers from the dead daemon. Metadata is recreated by - // spawnDaemon. - cleanupAll(name); - - const { spawnDaemon } = await import("./spawn.ts"); - await spawnDaemon({ - name, command, args, displayCommand, cwd, tags, - ...(displayName ? { displayName } : {}), - ...(metadata.rows !== undefined ? { rows: metadata.rows } : {}), - ...(metadata.cols !== undefined ? { cols: metadata.cols } : {}), - ...(metadata.ephemeral !== undefined ? { ephemeral: metadata.ephemeral } : {}), - ...(metadata.isolateEnv ? { isolateEnv: true } : {}), - ...(extraEnv && Object.keys(extraEnv).length > 0 ? { extraEnv } : {}), - ...(exactEnv ? { env: exactEnv } : {}), - }); + // Serialize compare-and-swap cleanup + replacement creation under the same + // per-name lock. If the observed generation changed while gc was planning, + // this tick is stale and must not touch the replacement. + if (!acquireLock(name)) return false; + try { + const current = readMetadata(name); + if (!current || !metadataMatchesObservation(metadata, current)) { + return false; + } + + // Wipe stale socket/pid/events before respawn so spawnDaemon doesn't trip + // over leftovers from the dead daemon. Keep the creation lock held until + // the replacement has published its socket. + cleanupAllWhileLocked(name); + + const { spawnDaemon } = await import("./spawn.ts"); + await spawnDaemon({ + name, command, args, displayCommand, cwd, tags, + creationLockOwnerPid: process.pid, + ...(displayName ? { displayName } : {}), + ...(metadata.rows !== undefined ? { rows: metadata.rows } : {}), + ...(metadata.cols !== undefined ? { cols: metadata.cols } : {}), + ...(metadata.ephemeral !== undefined ? { ephemeral: metadata.ephemeral } : {}), + ...(metadata.isolateEnv ? { isolateEnv: true } : {}), + ...(extraEnv && Object.keys(extraEnv).length > 0 ? { extraEnv } : {}), + ...(exactEnv ? { env: exactEnv } : {}), + }); + } finally { + releaseLock(name); + } // Best-effort event; respawn already succeeded if we got here. try { @@ -1287,6 +1351,7 @@ async function respawnPermanent( ts: new Date().toISOString(), }); } catch {} + return true; } /** @@ -1519,6 +1584,17 @@ function getLockPath(name: string): string { return path.join(getSessionDir(), `${name}.lock`); } +/** @internal Verify an explicitly delegated creation lock without acquiring it. */ +export function isLockOwnedByPid(name: string, ownerPid: number): boolean { + if (!Number.isSafeInteger(ownerPid) || ownerPid <= 0) return false; + try { + return parseInt(fs.readFileSync(getLockPath(name), "utf-8").trim(), 10) === ownerPid && + isProcessAlive(ownerPid); + } catch { + return false; + } +} + /** * Acquire an exclusive lock for a session name. Prevents concurrent * `pty run` calls from racing to create the same session. diff --git a/src/spawn.ts b/src/spawn.ts index d7538ba..96b9d10 100644 --- a/src/spawn.ts +++ b/src/spawn.ts @@ -93,6 +93,9 @@ export interface SpawnDaemonOptions { * path can't express it (a bundled consumer that hits the fallback also * isn't the operator-restart context this guards). */ scrubEnv?: string[]; + /** @internal PID owning an already-held per-name creation lock. Used only + * when a lifecycle operation must keep its CAS lock across CLI fallback. */ + creationLockOwnerPid?: number; } /** Default time we wait for a daemon's Unix socket to appear after @@ -236,9 +239,17 @@ function spawnViaCli(options: SpawnDaemonOptions): Promise { } cliArgs.push("--", options.command, ...options.args); + const env = { ...process.env }; + if (options.creationLockOwnerPid !== undefined) { + env.PTY_CREATION_LOCK_OWNER_PID = String(options.creationLockOwnerPid); + } else { + delete env.PTY_CREATION_LOCK_OWNER_PID; + } + const result = spawnSync("pty", cliArgs, { stdio: ["ignore", "pipe", "pipe"], encoding: "utf-8", + env, }); if (result.error !== undefined) { const err = result.error as NodeJS.ErrnoException; diff --git a/tests/gc-generation-guard.test.ts b/tests/gc-generation-guard.test.ts new file mode 100644 index 0000000..b0a79a4 --- /dev/null +++ b/tests/gc-generation-guard.test.ts @@ -0,0 +1,160 @@ +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { + cleanupObservedSession, + respawnPermanent, + type SessionInfo, + type SessionMetadata, +} from "../src/sessions.ts"; +import { terminateAndWait } from "./setup/processes.ts"; + +const roots: string[] = []; +const daemonPids: number[] = []; + +const makeRoot = () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "pty-gc-generation-")); + roots.push(root); + process.env.PTY_ROOT = root; + return root; +}; + +const metadata = ( + root: string, + generation?: string, + tags?: Record, +): SessionMetadata => ({ + ...(generation !== undefined ? { generation } : {}), + daemonPid: 2147483647, + command: "true", + args: [], + displayCommand: "true", + cwd: root, + createdAt: "2026-01-01T00:00:00.000Z", + exitedAt: "2026-01-01T00:00:01.000Z", + exitCode: 0, + tags, +}); + +afterEach(async () => { + await terminateAndWait(daemonPids.splice(0)); + delete process.env.PTY_ROOT; + for (const root of roots.splice(0)) { + fs.rmSync(root, { recursive: true, force: true }); + } +}); + +describe("gc generation compare-and-swap", () => { + it("does not residual-sweep a replacement generation", async () => { + const root = makeRoot(); + const name = "residual"; + const observed = metadata(root, "old"); + fs.writeFileSync( + path.join(root, `${name}.json`), + JSON.stringify(metadata(root, "replacement")), + ); + const session: SessionInfo = { + name, + socketPath: path.join(root, `${name}.sock`), + pid: null, + status: "exited", + metadata: observed, + }; + + const removed = await cleanupObservedSession(session); + + expect(removed).toBe(false); + expect(JSON.parse( + fs.readFileSync(path.join(root, `${name}.json`), "utf-8"), + ).generation).toBe("replacement"); + }); + + it("cleans an unchanged legacy observation without a generation", async () => { + const root = makeRoot(); + const name = "legacy-exact"; + const observed = metadata(root); + fs.writeFileSync( + path.join(root, `${name}.json`), + JSON.stringify(observed), + ); + const session: SessionInfo = { + name, + socketPath: path.join(root, `${name}.sock`), + pid: null, + status: "exited", + metadata: observed, + }; + + expect(await cleanupObservedSession(session)).toBe(true); + expect(fs.existsSync(path.join(root, `${name}.json`))).toBe(false); + }); + + it("preserves changed legacy metadata without a generation", async () => { + const root = makeRoot(); + const name = "legacy-stale"; + const observed = metadata(root, undefined, { revision: "old" }); + fs.writeFileSync( + path.join(root, `${name}.json`), + JSON.stringify(metadata(root, undefined, { revision: "replacement" })), + ); + const session: SessionInfo = { + name, + socketPath: path.join(root, `${name}.sock`), + pid: null, + status: "exited", + metadata: observed, + }; + + expect(await cleanupObservedSession(session)).toBe(false); + expect(JSON.parse( + fs.readFileSync(path.join(root, `${name}.json`), "utf-8"), + ).tags).toEqual({ revision: "replacement" }); + }); + + it("does not respawn over a replacement generation", async () => { + const root = makeRoot(); + const name = "permanent"; + const tags = { strategy: "permanent" }; + const observed = metadata(root, "old", tags); + fs.writeFileSync( + path.join(root, `${name}.json`), + JSON.stringify(metadata(root, "replacement", tags)), + ); + + const respawned = await respawnPermanent(name, observed); + + expect(respawned).toBe(false); + expect(JSON.parse( + fs.readFileSync(path.join(root, `${name}.json`), "utf-8"), + ).generation).toBe("replacement"); + }); + + it("keeps the CAS lock across bundled CLI fallback respawn", async () => { + const root = makeRoot(); + const name = "fallback"; + const observed = metadata(root, "same", { strategy: "permanent" }); + observed.command = "/bin/sh"; + observed.args = ["-c", "sleep 30"]; + observed.displayCommand = "sleep 30"; + fs.writeFileSync( + path.join(root, `${name}.json`), + JSON.stringify(observed), + ); + + const oldPath = process.env.PATH ?? ""; + process.env.PATH = `${path.resolve(import.meta.dirname, "../bin")}:${oldPath}`; + try { + expect(await respawnPermanent(name, observed)).toBe(true); + } finally { + process.env.PATH = oldPath; + } + + const pid = parseInt( + fs.readFileSync(path.join(root, `${name}.pid`), "utf-8").trim(), + 10, + ); + daemonPids.push(pid); + expect(pid).toBeGreaterThan(0); + }, 15_000); +});