diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 73595d5cfa6..19a62ca4262 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -247,7 +247,7 @@ export interface AdapterComputerExecutionTarget extends Omit; retainWarm(idleTimeoutMs: number): Promise; /** False means admission superseded this owner; callers must not close its process. */ - retire(): Promise; + retire(options?: { beforeStop?: () => Promise }): Promise; computerTool: { command: string; args: string[] }; } diff --git a/server/src/modules/computers/adapters/boat.test.ts b/server/src/modules/computers/adapters/boat.test.ts index 9c5a82f37d8..2c2535116b0 100644 --- a/server/src/modules/computers/adapters/boat.test.ts +++ b/server/src/modules/computers/adapters/boat.test.ts @@ -574,3 +574,29 @@ describe("Boat SSH admission", () => { expect(runner.execute).toHaveBeenCalledTimes(12); }); }); + + +describe("remote graceful retirement fence", () => { + it("records the exact generation deadline and blocks late launch or generation reuse", () => { + const root = realpathSync(mkdtempSync(join(tmpdir(), "computer-grace-"))); + roots.push(root); + const owners = join(root, "owners"); + const boot = join(root, "boot"); + writeFileSync(boot, "boot-id"); + const code = processProgram.replaceAll("/home/user/.paperclip-owners", owners) + .replaceAll("/proc/sys/kernel/random/boot_id", boot); + const invoke = (action: string, owner: Record) => { + const result = spawnSync("python3", ["-c", code], { input: JSON.stringify({ action, owner }), encoding: "utf8" }); + expect(result.status, result.stderr).toBe(0); + return JSON.parse(result.stdout); + }; + const owner = { id: "owned", generation: 7, phase: "retiring", retirementDeadline: "2099-01-01T00:00:00Z" }; + expect(invoke("advance", owner)).toEqual({}); + const marker = readFileSync(join(owners, "owned", "retiring.json"), "utf8"); + expect(JSON.parse(marker).deadline).toBe(Date.parse(owner.retirementDeadline)); + expect(invoke("launch", owner)).toEqual({ error: "conflict" }); + expect(invoke("advance", { ...owner, phase: "active", generation: 8 })).toEqual({ error: "conflict" }); + expect(invoke("advance", { ...owner, generation: 6 })).toEqual({ error: "conflict" }); + expect(readFileSync(join(owners, "owned", "retiring.json"), "utf8")).toBe(marker); + }); +}); diff --git a/server/src/modules/computers/adapters/boat.ts b/server/src/modules/computers/adapters/boat.ts index bf35376117d..9944f1d222b 100644 --- a/server/src/modules/computers/adapters/boat.ts +++ b/server/src/modules/computers/adapters/boat.ts @@ -112,7 +112,7 @@ with os.fdopen(fd,'a') as lock: `; export const processProgram = String.raw` -import os,sys,json,fcntl,subprocess,shutil,re +import os,sys,json,fcntl,subprocess,shutil,re,datetime p=json.load(sys.stdin);base='/home/user/.paperclip-owners';os.makedirs(base,exist_ok=True) def retire_owner(root,oid): open(os.path.join(root,'retired'),'a').close() @@ -151,6 +151,11 @@ with open(os.path.join(root,'lock'),'a') as lock: if owner['generation'] { expect(f.backend.stop).toHaveBeenCalledOnce(); }, ); - it("preserves recoverable running owners and gives completed warm owners their idle interval", async () => { + it("fences warm work at its deadline and bounds crashed-controller shutdown grace", async () => { const f = fixture(); await f.attach(); const binding = await f.admit(); + await binding.launch({ command: "runner" }); f.advance(180_000); await f.service.reconcile(); expect(f.backend.retire).not.toHaveBeenCalled(); @@ -343,6 +344,13 @@ describe("computer ownership", () => { await f.service.reconcile(); expect(f.backend.retire).not.toHaveBeenCalled(); f.advance(60_001); + await expect(f.admit()).rejects.toMatchObject({ code: "conflict" }); + await expect(binding.runner.execute({ command: "new work" })).rejects.toMatchObject({ code: "conflict" }); + await binding.process.runner.execute({ command: "close provider" }); + await f.service.reconcile(); + expect(f.backend.retire).not.toHaveBeenCalled(); + f.advance(30_000); + await expect(binding.process.ingress()).rejects.toMatchObject({ code: "conflict" }); await f.service.reconcile(); expect(f.backend.retire).toHaveBeenCalledOnce(); }); @@ -698,3 +706,107 @@ describe("computer retirement proof", () => { expect(f.backend.inspect).not.toHaveBeenCalled(); }); }); + + +describe("computer graceful retirement", () => { + it("fences admission and ordinary execution while allowing the exact pinned process to close", async () => { + const f = fixture(); + await f.attach(); + const binding = await f.admit(); + await binding.launch({ command: "runner" }); + let finish!: () => void; + let entered!: () => void; + const closing = new Promise((resolve) => { finish = resolve; }); + const started = new Promise((resolve) => { entered = resolve; }); + const beforeStop = vi.fn(async () => { entered(); await closing; }); + const retiring = f.service.retire({ ...f.scope, owner: binding.owner, beforeStop }); + await started; + expect((await f.repository.get(f.scope)).ledger.owners[0]).toMatchObject({ phase: "retiring" }); + await expect(f.admit()).rejects.toMatchObject({ code: "conflict" }); + await expect(binding.runner.execute({ command: "forbidden" })).rejects.toMatchObject({ code: "conflict" }); + await expect(binding.launch({ command: "forbidden" })).rejects.toMatchObject({ code: "conflict" }); + await binding.process.runner.execute({ command: "close" }); + await binding.process.ingress(); + expect(f.backend.retire).not.toHaveBeenCalled(); + const repeated = f.service.retire({ ...f.scope, owner: binding.owner, beforeStop }); + finish(); + expect(await retiring).toEqual({ retired: true }); + expect(await repeated).toEqual({ retired: true }); + expect(beforeStop).toHaveBeenCalledOnce(); + expect(f.backend.retire).toHaveBeenCalledOnce(); + await expect(binding.process.ingress()).rejects.toMatchObject({ code: "conflict" }); + expect((await f.admit()).owner.ownerId).not.toBe(binding.owner.ownerId); + }); + it("never invokes a stale generation callback and always stops after callback failure", async () => { + const f = fixture(); + await f.attach(); + const old = await f.admit(); + await old.launch({ command: "runner" }); + await f.service.retainWarm({ ...f.scope, owner: old.owner, idleTimeoutMs: 60_000 }); + const current = await f.admit(); + const stale = vi.fn(async () => {}); + expect(await f.service.retire({ ...f.scope, owner: old.owner, beforeStop: stale })).toEqual({ retired: false }); + expect(stale).not.toHaveBeenCalled(); + expect(await f.service.retire({ ...f.scope, owner: current.owner, beforeStop: async () => { throw new Error("provider close failed"); } })).toEqual({ retired: true }); + expect(f.backend.retire).toHaveBeenCalledOnce(); + }); + it("bounds a stalled graceful close and denies late process control", async () => { + vi.useFakeTimers(); + try { + const f = fixture(); + await f.attach(); + const binding = await f.admit(); + await binding.launch({ command: "runner" }); + let entered!: () => void; + const started = new Promise((resolve) => { entered = resolve; }); + const retiring = f.service.retire({ ...f.scope, owner: binding.owner, beforeStop: async () => { + entered(); await new Promise(() => {}); + } }); + await started; + f.advance(30_001); + await expect(binding.process.runner.execute({ command: "late" })).rejects.toMatchObject({ code: "conflict" }); + await vi.advanceTimersByTimeAsync(30_000); + expect(await retiring).toEqual({ retired: true }); + expect(f.backend.retire).toHaveBeenCalledOnce(); + } finally { vi.useRealTimers(); } + }); + it("does not extend the original warm shutdown deadline when retirement starts late", async () => { + vi.useFakeTimers(); + try { + const f = fixture(); + await f.attach(); + const binding = await f.admit(); + await binding.launch({ command: "runner" }); + await f.service.retainWarm({ ...f.scope, owner: binding.owner, idleTimeoutMs: 60_000 }); + f.advance(80_000); + let entered!: () => void; + const started = new Promise((resolve) => { entered = resolve; }); + const retiring = f.service.retire({ ...f.scope, owner: binding.owner, beforeStop: async () => { + entered(); await new Promise(() => {}); + } }); + await started; + expect((await f.repository.get(f.scope)).ledger.owners[0]!.retirementDeadline).toBe("2026-10-10T12:01:30.000Z"); + f.advance(10_000); + await vi.advanceTimersByTimeAsync(10_000); + expect(await retiring).toEqual({ retired: true }); + expect(f.backend.retire).toHaveBeenCalledOnce(); + } finally { vi.useRealTimers(); } + }); + it("reconciles a crashed controller reservation only after its durable deadline", async () => { + const f = fixture(); + await f.attach(); + const binding = await f.admit(); + await binding.launch({ command: "runner" }); + await f.repository.update(f.scope, (record) => { + record.ledger.owners[0]!.phase = "retiring"; + record.ledger.owners[0]!.retirementDeadline = "2026-10-10T12:00:30.000Z"; + }); + await f.service.reconcile(); + expect(f.backend.retire).not.toHaveBeenCalled(); + await expect(f.service.retire({ ...f.scope, owner: binding.owner })).rejects.toMatchObject({ code: "conflict" }); + f.advance(30_001); + await f.service.reconcile(); + expect(f.backend.retire).toHaveBeenCalledOnce(); + expect((await f.repository.get(f.scope)).ledger.owners[0]!.phase).toBe("retired"); + }); +}); diff --git a/server/src/modules/computers/application/service.ts b/server/src/modules/computers/application/service.ts index 3fb06c0a1c1..5ad160d8370 100644 --- a/server/src/modules/computers/application/service.ts +++ b/server/src/modules/computers/application/service.ts @@ -23,6 +23,8 @@ class ComputerSessionBusyError extends ComputerError { } } +// This is shutdown time only; admission and ordinary work stop at the warm deadline. +const RETIREMENT_GRACE_MS = 30_000; type Scope = { companyId: string; environmentId: string }; export function createComputerService( repository: ComputerRepository, @@ -38,6 +40,15 @@ export function createComputerService( ownerId: owner.id, generation: owner.generation, }); + const retirements = new Map>(); + const retirementGraceOpen = (owner: Owner) => owner.phase === "retiring" && + !!owner.retirementDeadline && Date.parse(owner.retirementDeadline) > now().getTime(); + const warmShutdownGraceOpen = (owner: Owner) => owner.kind === "runner" && owner.phase === "warm" && + !!owner.deadline && Date.parse(owner.deadline) + RETIREMENT_GRACE_MS > now().getTime(); + const processControlAllowed = (owner: Owner) => + ((owner.phase === "active" || owner.phase === "starting") && !expired(owner, now())) || + (owner.phase === "warm" && (!expired(owner, now()) || warmShutdownGraceOpen(owner))) || + retirementGraceOpen(owner); const base = (record: ComputerRecord) => `/home/user/paperclip/${segment(record.companyId)}`; async function admitRecord( @@ -173,14 +184,19 @@ export function createComputerService( const control = await backend.runner(record, { control: true }); let pinnedProcess = owner.process ? structuredClone(owner.process) : null; async function processScope() { - if (!pinnedProcess) return scoped({ ...record, owner: ownerRef }); + if (!pinnedProcess) { + const current = await scoped({ ...record, owner: ownerRef }); + if (!["active", "starting", "warm"].includes(current.owner.phase) || expired(current.owner, now())) + throw new ComputerError("conflict", "Computer process capability has expired"); + return current; + } const latest = await repository.get(record); const candidate = latest.ledger.owners.find( (value) => value.id === owner.id, ); if ( !candidate?.process || - !["active", "starting", "warm"].includes(candidate.phase) || + !processControlAllowed(candidate) || candidate.process.nonce !== pinnedProcess.nonce || candidate.process.unitName !== pinnedProcess.unitName || candidate.process.bootId !== pinnedProcess.bootId || @@ -200,7 +216,8 @@ export function createComputerService( const current = await (processScoped ? processScope() : scoped({ ...record, owner: ownerRef })); - if (!["active", "starting", "warm"].includes(current.owner.phase)) + if ((!["active", "starting", "warm"].includes(current.owner.phase) || expired(current.owner, now())) && + !(processScoped && pinnedProcess && processControlAllowed(current.owner))) throw new ComputerError("conflict", "Computer owner is retired"); // Start a command unit under the same owner slice while holding the remote tombstone lock. // The lock is released only after systemd knows the unit, so retirement cannot miss a delayed launch. @@ -235,6 +252,9 @@ try: with open(root+'/lock','a') as lock: fcntl.flock(lock,fcntl.LOCK_EX) if os.path.exists(root+'/retired'):raise RuntimeError('computer owner retired') + if os.path.exists(root+'/retiring.json'): + with open(root+'/retiring.json') as retirement:deadline=json.load(retirement)['deadline'] + if not p.get('processClaim') or time.time()*1000>=deadline:raise RuntimeError('computer owner is retiring') if p.get('processClaim'): claim=json.load(open(root+'/claim.json')) if any(claim[k]!=p['processClaim'][k] for k in ['nonce','unitName','bootId','launchGeneration']) or claim['bootId']!=open('/proc/sys/kernel/random/boot_id').read().strip():raise RuntimeError('computer process superseded') @@ -347,7 +367,7 @@ finally: }, async ingress(input?: { port?: number; path?: string }) { const current = await scoped({ ...record, owner: ownerRef }); - if (current.owner.phase === "retired") + if (!["active", "starting", "warm"].includes(current.owner.phase) || expired(current.owner, now())) throw new ComputerError("conflict", "Computer owner retired"); return backend.ingress( current.record, @@ -389,10 +409,7 @@ finally: if (owner && owner.phase !== "warm") throw new ComputerSessionBusyError(owner.runId, owner.agentId); if (owner && expired(owner, now())) - throw new ComputerError( - "conflict", - "Computer warm session is retiring", - ); + throw new ComputerSessionBusyError(owner.runId, owner.agentId); if (owner) { owner.generation++; owner.phase = "starting"; @@ -504,19 +521,53 @@ finally: owner.runId === input.runId, ); } - async function retire(input: Scope & { owner: OwnerRef }) { + async function retire(input: Scope & { owner: OwnerRef; beforeStop?: () => Promise }) { + const key = `${input.companyId}:${input.environmentId}:${input.owner.computerId}:${input.owner.ownerId}:${input.owner.generation}`; + const existing = retirements.get(key); + if (existing) return existing; + const pending = performRetirement(input); + retirements.set(key, pending); + try { return await pending; } + finally { if (retirements.get(key) === pending) retirements.delete(key); } + } + async function performRetirement(input: Scope & { owner: OwnerRef; beforeStop?: () => Promise }) { const state = await repository.update(input, (record) => { - const owner = record.ledger.owners.find( - (value) => value.id === input.owner.ownerId, - ); + const owner = record.ledger.owners.find((value) => value.id === input.owner.ownerId); if (record.id !== input.owner.computerId || !owner) throw new ComputerError("not_found", "Computer owner not found"); if (owner.generation !== input.owner.generation) return null; - if (owner.phase !== "retired") owner.phase = "retiring"; - return { record: structuredClone(record), owner: structuredClone(owner) }; + if (retirementGraceOpen(owner)) + throw new ComputerError("conflict", "Computer graceful retirement is pending"); + const retirementDeadline = owner.phase === "warm" && owner.deadline + ? Math.min(now().getTime() + RETIREMENT_GRACE_MS, Date.parse(owner.deadline) + RETIREMENT_GRACE_MS) + : now().getTime() + RETIREMENT_GRACE_MS; + const graceful = owner.phase !== "retiring" && owner.phase !== "retired" && + owner.kind === "runner" && !!owner.process && !!input.beforeStop && retirementDeadline > now().getTime(); + if (owner.phase !== "retired") { + owner.phase = "retiring"; + owner.retirementDeadline = graceful ? new Date(retirementDeadline).toISOString() : null; + } + return { record: structuredClone(record), owner: structuredClone(owner), graceful }; }); if (!state) return { retired: false }; if (state.owner.phase === "retired") return { retired: true }; + if (state.graceful) { + let timer: ReturnType | undefined; + try { + await Promise.race([ + (async () => { + await backend.advance(state.record, state.owner); + if (retirementGraceOpen(state.owner)) await input.beforeStop!(); + })(), + new Promise((resolve) => { timer = setTimeout(resolve, Math.max(0, Date.parse(state.owner.retirementDeadline!) - now().getTime())); }), + ]); + } catch { + // Graceful provider shutdown is best effort; physical retirement is mandatory. + } finally { if (timer) clearTimeout(timer); } + await repository.update(input, (record) => { + exactOwner(record, input.owner).retirementDeadline = null; + }); + } if (state.owner.kind === "runner") { const status = await backend.inspect(state.record); if (!["stopped", "archived"].includes(status.state)) @@ -526,6 +577,7 @@ finally: const owner = exactOwner(record, input.owner); owner.phase = "retired"; owner.deadline = null; + owner.retirementDeadline = null; }); return { retired: true }; } @@ -885,7 +937,8 @@ finally: } record = await repository.get(record); for (const owner of liveOwners(record.ledger)) - if (owner.phase === "retiring" || expired(owner, now())) + if (owner.phase === "retiring" ? !retirementGraceOpen(owner) + : expired(owner, now()) && !warmShutdownGraceOpen(owner)) await retire({ ...record, owner: ref(record, owner) }); await repository.update(record, (current) => { current.ledger.owners = current.ledger.owners.filter( diff --git a/server/src/modules/computers/domain/ledger.ts b/server/src/modules/computers/domain/ledger.ts index 136e0c444a6..c433e576b81 100644 --- a/server/src/modules/computers/domain/ledger.ts +++ b/server/src/modules/computers/domain/ledger.ts @@ -23,6 +23,8 @@ export interface Owner { port: number; deadline: string | null; absoluteDeadline: string | null; + /** Bounded graceful shutdown reservation; only pinned process control remains valid. */ + retirementDeadline?: string | null; process: ProcessClaim | null; } export interface Ledger { diff --git a/server/src/services/computer-environment-driver.ts b/server/src/services/computer-environment-driver.ts index fd1e0ef8dcd..505c2885ca7 100644 --- a/server/src/services/computer-environment-driver.ts +++ b/server/src/services/computer-environment-driver.ts @@ -121,8 +121,8 @@ export async function computerExecutionTarget(db: Db, lease: EnvironmentLease, i workspaceRealization: { mode: "in_place", authoritativeRoot: binding.remoteCwd, pathAliases: [], outboundRestorePaths: [] }, launch: binding.launch, inspectProcess: binding.inspectProcess, retainWarm: (idleTimeoutMs) => computers.retainWarm({ ...scope, owner, idleTimeoutMs }), - retire: async () => { - const result = await computers.retire({ ...scope, owner }); + retire: async (options) => { + const result = await computers.retire({ ...scope, owner, beforeStop: options?.beforeStop }); if (result.retired) await environmentService(db).releaseLease(lease.id, "released", { cleanupStatus: "success", remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" }) }); return result.retired;