mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-13 23:04:38 +02:00
Reap retired command spools before resuming computer work
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
a9882c9dee
commit
46ec1f81de
2 files changed
+160
-52
No files matched your search
@@ -1,4 +1,5 @@
|
||||
import {
|
||||
existsSync,
|
||||
mkdtempSync,
|
||||
readFileSync,
|
||||
realpathSync,
|
||||
@@ -13,7 +14,7 @@ import { join } from "node:path";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { remoteProgram } from "./remote-program.js";
|
||||
import { afterEach, describe, it, expect, vi } from "vitest";
|
||||
import { boatBackend } from "./boat.js";
|
||||
import { boatBackend, processProgram } from "./boat.js";
|
||||
import { buildGitAuthInvocation } from "../../../services/git-credentials.js";
|
||||
import type { ComputerRecord } from "../domain/ledger.js";
|
||||
const record: ComputerRecord = {
|
||||
@@ -306,3 +307,80 @@ process.stdout.write(result.stdout||"");process.stderr.write(result.stderr||"");
|
||||
).toEqual(bytes);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("retired owner spool cleanup after provider resume", () => {
|
||||
function cleanupFixture(state = "inactive") {
|
||||
const temp = realpathSync(mkdtempSync(join(tmpdir(), "computer-retired-")));
|
||||
roots.push(temp);
|
||||
const owners = join(temp, "owners");
|
||||
const bin = join(temp, "bin");
|
||||
const calls = join(temp, "systemctl-calls");
|
||||
mkdirSync(bin);
|
||||
writeFileSync(join(bin, "systemctl"), `#!/bin/sh
|
||||
printf '%s\\n' "$*" >> '${calls}'
|
||||
case "$*" in *show*) printf '%s\\n' '${state}';; esac
|
||||
`, { mode: 0o700 });
|
||||
const owner = (id: string, generation = 7) => {
|
||||
const root = join(owners, id);
|
||||
mkdirSync(join(root, "command-one"), { recursive: true });
|
||||
writeFileSync(join(root, "generation"), String(generation));
|
||||
writeFileSync(join(root, "command-one", "stdin"), "private command input", { mode: 0o600 });
|
||||
writeFileSync(join(root, "preserve"), "ownership metadata");
|
||||
return root;
|
||||
};
|
||||
const cleanup = (input = [{ id: "retired-owner", generation: 7 }]) => spawnSync(
|
||||
"python3", ["-c", processProgram.replaceAll("/home/user/.paperclip-owners", owners)],
|
||||
{ input: JSON.stringify({ action: "cleanup-retired", owners: input }), encoding: "utf8", env: { ...process.env, PATH: `${bin}:${process.env.PATH}` } },
|
||||
);
|
||||
return { temp, owners, calls, owner, cleanup };
|
||||
}
|
||||
|
||||
it("reaps only recorded retired command spools and is idempotent", () => {
|
||||
const f = cleanupFixture();
|
||||
const retired = f.owner("retired-owner");
|
||||
const active = f.owner("active-owner");
|
||||
const outside = join(f.temp, "user-files");
|
||||
mkdirSync(outside);
|
||||
writeFileSync(join(outside, "preserve"), "user data");
|
||||
symlinkSync(outside, join(retired, "command-symlink"));
|
||||
expect(f.cleanup().status).toBe(0);
|
||||
expect(existsSync(join(retired, "command-one"))).toBe(false);
|
||||
expect(existsSync(join(retired, "retired"))).toBe(true);
|
||||
expect(readFileSync(join(retired, "preserve"), "utf8")).toBe("ownership metadata");
|
||||
expect(readFileSync(join(active, "command-one", "stdin"), "utf8")).toBe("private command input");
|
||||
expect(readFileSync(join(outside, "preserve"), "utf8")).toBe("user data");
|
||||
const calls = readFileSync(f.calls, "utf8");
|
||||
expect(calls).toContain("--user stop paperclip-retired-owner.slice paperclip-retired-owner.service");
|
||||
expect(calls).not.toContain("active-owner");
|
||||
expect(f.cleanup().status).toBe(0);
|
||||
expect(readFileSync(f.calls, "utf8")).toBe(calls);
|
||||
});
|
||||
|
||||
it("preserves spools when retirement cannot be confirmed", () => {
|
||||
const f = cleanupFixture("active");
|
||||
const retired = f.owner("retired-owner");
|
||||
const result = f.cleanup();
|
||||
expect(result.status).not.toBe(0);
|
||||
expect(result.stderr).toContain("process retirement unconfirmed");
|
||||
expect(existsSync(join(retired, "command-one", "stdin"))).toBe(true);
|
||||
});
|
||||
|
||||
it("rejects a stale generation without stopping its current process", () => {
|
||||
const f = cleanupFixture();
|
||||
const retired = f.owner("retired-owner", 8);
|
||||
expect(JSON.parse(f.cleanup().stdout)).toEqual({ error: "conflict" });
|
||||
expect(existsSync(join(retired, "command-one", "stdin"))).toBe(true);
|
||||
expect(existsSync(f.calls)).toBe(false);
|
||||
});
|
||||
|
||||
it("rejects symlink owner roots and ignores missing retired roots", () => {
|
||||
const f = cleanupFixture();
|
||||
const active = f.owner("active-owner");
|
||||
symlinkSync(active, join(f.owners, "retired-owner"));
|
||||
expect(f.cleanup().stderr).toContain("invalid retired owner directory");
|
||||
expect(existsSync(join(active, "command-one", "stdin"))).toBe(true);
|
||||
expect(f.cleanup([{ id: "missing-owner", generation: 7 }]).status).toBe(0);
|
||||
expect(existsSync(f.calls)).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -19,6 +19,75 @@ const desktopCache = new Map<
|
||||
string,
|
||||
{ viewerUrl: string; expiresAt: string }
|
||||
>();
|
||||
export const processProgram = String.raw`
|
||||
import os,sys,json,fcntl,subprocess,shutil,re
|
||||
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()
|
||||
unit='paperclip-'+oid+'.service';slice='paperclip-'+oid+'.slice'
|
||||
subprocess.run(['systemctl','--user','stop',slice,unit],capture_output=True)
|
||||
state=subprocess.run(['systemctl','--user','show',slice,'--property=ActiveState','--value'],capture_output=True,text=True).stdout.strip()
|
||||
if state not in ('inactive','failed',''):raise RuntimeError('process retirement unconfirmed')
|
||||
for entry in os.scandir(root):
|
||||
if entry.name.startswith('command-') and entry.is_dir(follow_symlinks=False):
|
||||
try:shutil.rmtree(entry.path)
|
||||
except FileNotFoundError:pass
|
||||
if p['action']=='cleanup-retired':
|
||||
# The controller supplies only its durable retired runner identities. Do not
|
||||
# discover owners from the filesystem or touch any agent/project directory.
|
||||
for owner in p['owners']:
|
||||
oid=owner['id']
|
||||
if not isinstance(oid,str) or not re.fullmatch('[A-Za-z0-9_-]{1,128}',oid):raise RuntimeError('invalid retired owner')
|
||||
root=os.path.join(base,oid)
|
||||
if os.path.islink(root):raise RuntimeError('invalid retired owner directory')
|
||||
if not os.path.isdir(root):continue
|
||||
with open(os.path.join(root,'lock'),'a') as lock:
|
||||
fcntl.flock(lock,fcntl.LOCK_EX)
|
||||
generation_path=os.path.join(root,'generation')
|
||||
generation=int(open(generation_path).read()) if os.path.exists(generation_path) else 0
|
||||
if owner['generation']<generation:print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
if os.path.exists(os.path.join(root,'retired')) and not any(e.name.startswith('command-') and e.is_dir(follow_symlinks=False) for e in os.scandir(root)):continue
|
||||
retire_owner(root,oid)
|
||||
print('{}');sys.exit(0)
|
||||
owner=p['owner'];oid=owner['id'];root=os.path.join(base,oid);os.makedirs(root,exist_ok=True)
|
||||
with open(os.path.join(root,'lock'),'a') as lock:
|
||||
fcntl.flock(lock,fcntl.LOCK_EX)
|
||||
tombstone=os.path.join(root,'retired');unit='paperclip-'+oid+'.service';slice='paperclip-'+oid+'.slice'
|
||||
boot=open('/proc/sys/kernel/random/boot_id').read().strip()
|
||||
generation_path=os.path.join(root,'generation')
|
||||
generation=int(open(generation_path).read()) if os.path.exists(generation_path) else 0
|
||||
if owner['generation']<generation:print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
if p['action']=='advance':
|
||||
if os.path.exists(tombstone):print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
with open(generation_path,'w') as f:f.write(str(owner['generation']))
|
||||
print('{}')
|
||||
elif p['action']=='retire':
|
||||
retire_owner(root,oid)
|
||||
print('{}')
|
||||
elif p['action']=='inspect':
|
||||
claim=owner.get('process');marker=os.path.join(root,'claim.json');actual=json.load(open(marker)) if os.path.exists(marker) else None
|
||||
valid=claim and actual and claim['nonce']==actual['nonce'] and claim['launchGeneration']==actual['launchGeneration'] and actual['bootId']==boot
|
||||
state=subprocess.run(['systemctl','--user','is-active',unit],capture_output=True,text=True).stdout.strip()
|
||||
print(json.dumps({'running':bool(valid and state=='active' and not os.path.exists(tombstone)),'claim':actual if valid else claim}))
|
||||
else:
|
||||
if os.path.exists(tombstone):print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
claim=owner['process'];claim['bootId']=boot
|
||||
marker=os.path.join(root,'claim.json')
|
||||
if os.path.exists(marker):
|
||||
old=json.load(open(marker))
|
||||
if old['launchGeneration']>claim['launchGeneration'] or (old['launchGeneration']==claim['launchGeneration'] and old['nonce']!=claim['nonce']):print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
active=subprocess.run(['systemctl','--user','is-active',unit],capture_output=True,text=True).stdout.strip()=='active'
|
||||
if active and old['nonce']!=claim['nonce']:print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
with open(marker,'w') as f:json.dump(claim,f)
|
||||
# Shell launch scripts own expansion of their positional arguments and identity markers.
|
||||
payload=p['input'];args=['systemd-run','--user','--expand-environment=no','--unit='+unit,'--slice='+slice,'--collect','--property=KillMode=control-group','--working-directory='+payload.get('cwd','/home/user')]
|
||||
for key,value in payload.get('env',{}).items():args.append('--setenv='+key+'='+value)
|
||||
state=subprocess.run(['systemctl','--user','is-active',unit],capture_output=True,text=True).stdout.strip()
|
||||
if state!='active':
|
||||
result=subprocess.run(args+['--',payload['command']]+payload.get('args',[]),capture_output=True)
|
||||
if result.returncode:raise RuntimeError('launch failed')
|
||||
print(json.dumps(claim))
|
||||
`;
|
||||
export function boatBackend(
|
||||
resolveKey: (record: ComputerRecord) => Promise<string>,
|
||||
fetcher: typeof fetch = fetch,
|
||||
@@ -195,57 +264,7 @@ export function boatBackend(
|
||||
);
|
||||
return value;
|
||||
}
|
||||
const processProgram = String.raw`
|
||||
import os,sys,json,fcntl,subprocess,shutil
|
||||
p=json.load(sys.stdin);base='/home/user/.paperclip-owners';os.makedirs(base,exist_ok=True)
|
||||
owner=p['owner'];oid=owner['id'];root=os.path.join(base,oid);os.makedirs(root,exist_ok=True)
|
||||
with open(os.path.join(root,'lock'),'a') as lock:
|
||||
fcntl.flock(lock,fcntl.LOCK_EX)
|
||||
tombstone=os.path.join(root,'retired');unit='paperclip-'+oid+'.service';slice='paperclip-'+oid+'.slice'
|
||||
boot=open('/proc/sys/kernel/random/boot_id').read().strip()
|
||||
generation_path=os.path.join(root,'generation')
|
||||
generation=int(open(generation_path).read()) if os.path.exists(generation_path) else 0
|
||||
if owner['generation']<generation:print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
if p['action']=='advance':
|
||||
if os.path.exists(tombstone):print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
with open(generation_path,'w') as f:f.write(str(owner['generation']))
|
||||
print('{}')
|
||||
elif p['action']=='retire':
|
||||
open(tombstone,'a').close()
|
||||
subprocess.run(['systemctl','--user','stop',slice,unit],capture_output=True)
|
||||
state=subprocess.run(['systemctl','--user','show',slice,'--property=ActiveState','--value'],capture_output=True,text=True).stdout.strip()
|
||||
if state not in ('inactive','failed',''):raise RuntimeError('process retirement unconfirmed')
|
||||
# Reap only this retired owner's command spools after its cgroup is stopped.
|
||||
# This covers controller crashes and SSH timeouts that skipped caller cleanup.
|
||||
for entry in os.scandir(root):
|
||||
if entry.name.startswith('command-') and entry.is_dir(follow_symlinks=False):
|
||||
try:shutil.rmtree(entry.path)
|
||||
except FileNotFoundError:pass
|
||||
print('{}')
|
||||
elif p['action']=='inspect':
|
||||
claim=owner.get('process');marker=os.path.join(root,'claim.json');actual=json.load(open(marker)) if os.path.exists(marker) else None
|
||||
valid=claim and actual and claim['nonce']==actual['nonce'] and claim['launchGeneration']==actual['launchGeneration'] and actual['bootId']==boot
|
||||
state=subprocess.run(['systemctl','--user','is-active',unit],capture_output=True,text=True).stdout.strip()
|
||||
print(json.dumps({'running':bool(valid and state=='active' and not os.path.exists(tombstone)),'claim':actual if valid else claim}))
|
||||
else:
|
||||
if os.path.exists(tombstone):print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
claim=owner['process'];claim['bootId']=boot
|
||||
marker=os.path.join(root,'claim.json')
|
||||
if os.path.exists(marker):
|
||||
old=json.load(open(marker))
|
||||
if old['launchGeneration']>claim['launchGeneration'] or (old['launchGeneration']==claim['launchGeneration'] and old['nonce']!=claim['nonce']):print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
active=subprocess.run(['systemctl','--user','is-active',unit],capture_output=True,text=True).stdout.strip()=='active'
|
||||
if active and old['nonce']!=claim['nonce']:print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
with open(marker,'w') as f:json.dump(claim,f)
|
||||
# Shell launch scripts own expansion of their positional arguments and identity markers.
|
||||
payload=p['input'];args=['systemd-run','--user','--expand-environment=no','--unit='+unit,'--slice='+slice,'--collect','--property=KillMode=control-group','--working-directory='+payload.get('cwd','/home/user')]
|
||||
for key,value in payload.get('env',{}).items():args.append('--setenv='+key+'='+value)
|
||||
state=subprocess.run(['systemctl','--user','is-active',unit],capture_output=True,text=True).stdout.strip()
|
||||
if state!='active':
|
||||
result=subprocess.run(args+['--',payload['command']]+payload.get('args',[]),capture_output=True)
|
||||
if result.returncode:raise RuntimeError('launch failed')
|
||||
print(json.dumps(claim))
|
||||
`;
|
||||
|
||||
async function host(record: ComputerRecord, port: number) {
|
||||
if (!Number.isInteger(port) || port < 1024 || port > 65535)
|
||||
throw new ComputerError("invalid", "Invalid preview port");
|
||||
@@ -330,6 +349,17 @@ print('{}')
|
||||
computerId: record.id,
|
||||
},
|
||||
);
|
||||
const retiredOwners = record.ledger.owners
|
||||
.filter((owner) => owner.kind === "runner" && owner.phase === "retired")
|
||||
.map(({ id, generation }) => ({ id, generation }));
|
||||
if (retiredOwners.length) {
|
||||
// Archived providers cannot clean remote files. Settle their exact
|
||||
// retired owners after resume, before admitting any new command.
|
||||
await execute(record, processProgram, {
|
||||
action: "cleanup-retired",
|
||||
owners: retiredOwners,
|
||||
});
|
||||
}
|
||||
},
|
||||
async advance(record, owner) {
|
||||
await execute(record, processProgram, { action: "advance", owner });
|
||||
|
||||
Reference in new issue
Block a user