mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 01:57:31 +01:00
fix(recovery): reclaim verified inactive provider caches on full disks (#19192)
Recover from a full-disk workspace save by reclaiming only a verified inactive provider installation under the existing source and maintenance authority. Commit the preparation inventory in Core before changing cache contents. Bind interrupted cleanup to stable filesystem identities and keep small staged generation markers when directory birth times are unavailable. Skip read-only or permission-denied assets and stop freeing blocks once marker space is available. Preserve task files, Git, session history and uncertain action receipts. Only the normal saved-workspace and provider-stop receipts permit continuation. Validated with 446 complete recovery regressions on the repair revision, subsequent focused recovery and transport suites, typecheck/build, and all exact-head hosted CI. Greptile reviewed the final head at 5/5 with no unresolved findings. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip authored and GitHub committed
1 parent
037c58bb22
commit
07a2941f0c
4 files changed
+800
-3
No files matched your search
@@ -163,6 +163,44 @@ verified snapshot. Upload operations have stable identities and manifests.
|
||||
Recovered snapshots attach separately with provenance; they never become an
|
||||
invented baseline for overwriting newer files.
|
||||
|
||||
When that owned files-only export reports a full sandbox disk, recovery can
|
||||
reclaim a verified inactive native provider installation and retry the same save
|
||||
once. The failed source must have a terminal coordinator with no controller
|
||||
owner or resumable failure, and its Core-verified provider-pack admission must
|
||||
match the exact run, company, agent, workspace and lease. Unverified historical
|
||||
installations remain held for operator repair.
|
||||
|
||||
Only the reserved provider-pack `node_modules` and `provider-assets` directories
|
||||
are eligible. The installation manifest, task files, Git history, sessions,
|
||||
checkpoints and action receipts stay intact. A read-only inspection supplies
|
||||
manifest and directory identities for a source-bound journal in Core's database.
|
||||
That journal commits before cache contents change. Where directory creation
|
||||
times are available, cleanup needs only reads and deletion, so it can start even
|
||||
when archive-helper writes fail for lack of space.
|
||||
|
||||
On filesystems without creation times, deleting children changes the directory's
|
||||
fallback timestamp. Recovery first releases blocks by truncating only regular
|
||||
installation files with no additional hard links through verified file handles. It
|
||||
skips files without write permission and stops scanning as soon as marker space
|
||||
is available, including when an earlier eligible cache supplied that space. It
|
||||
then moves the cache to an operation-specific path and pins a small identity
|
||||
marker in the existing Core journal before recursive deletion. An interrupted
|
||||
cleanup can resume against that marker without accepting a new installation,
|
||||
a replacement staged directory or a copied marker. The emptied staged directory
|
||||
and its marker remain in reserved runtime storage as generation evidence; the
|
||||
canonical installation path stays absent so later installation rebuilds it.
|
||||
Neither task files nor sessions enter this staging path. Changed identities or
|
||||
symlinks keep the original export failure.
|
||||
|
||||
Inspection, optional staging and deletion each have a bounded remote call. The
|
||||
block-release scan visits at most 8,192 entries across the eligible caches and
|
||||
stops once 256 KiB is available for the markers. If capacity cannot be read, it
|
||||
uses a bounded 256 KiB estimate of released blocks. Journal creation and
|
||||
cleanup record separate terminal operations; neither is a file-save receipt or
|
||||
a new background obligation. An uncertain acknowledgement never proves that
|
||||
files were saved. This adds no provider turn, recovery generation or retry budget,
|
||||
and the normal save and stop receipts still govern continuation.
|
||||
|
||||
Known publication receipts and exact API mutation receipts carry forward.
|
||||
Uncertain API mutations and upstream tool mutations remain fenced across run
|
||||
replacement. Reads and independent work remain available. Generic upstream
|
||||
|
||||
@@ -0,0 +1,261 @@
|
||||
import { and, asc, eq, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, nativeRunFinalizations, withDedicatedDbConnection, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { z } from "zod";
|
||||
import type { RemoteWorkspaceTarget } from "../remote-workspace-transport.js";
|
||||
|
||||
const schema = "paperclip.native-provider-cache-reclaim.v1";
|
||||
const prepareSchema = "paperclip.native-provider-cache-prepare.v1";
|
||||
const inventorySchema = "paperclip.native-provider-cache-inventory.v1";
|
||||
const preparePhase = "native_provider_cache_reclaim_prepare";
|
||||
const digestPattern = /^sha256:[a-f0-9]{64}$/;
|
||||
const uuidPattern = /^[a-f0-9-]{36}$/;
|
||||
const record = (value: unknown): Record<string, unknown> => value && typeof value === "object" && !Array.isArray(value)
|
||||
? value as Record<string, unknown> : {};
|
||||
const numericIdentity = z.string().regex(/^\d{1,24}$/);
|
||||
const identity = z.object({ dev: numericIdentity, ino: numericIdentity, generation: z.string().regex(/^[bc]:\d{1,24}$/) });
|
||||
const inventoryCodec = z.object({ schema: z.literal(inventorySchema), sourceRunId: z.string().regex(uuidPattern),
|
||||
manifestDigest: z.string().regex(digestPattern), manifestSha256: z.string().regex(/^[a-f0-9]{64}$/),
|
||||
root: identity, pack: identity, manifest: identity, sessions: identity,
|
||||
targets: z.array(identity.extend({ name: z.enum(["node_modules", "provider-assets"]) })).max(2),
|
||||
availableBytesBefore: z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER).nullable(),
|
||||
});
|
||||
type Inventory = z.infer<typeof inventoryCodec>;
|
||||
const stagedCodec = z.array(z.object({ name: z.enum(["node_modules", "provider-assets"]),
|
||||
cache: identity, marker: identity, markerSha256: z.string().regex(/^[a-f0-9]{64}$/) })).max(2);
|
||||
type Staged = z.infer<typeof stagedCodec>;
|
||||
|
||||
function verifiedCacheDigest(run: typeof heartbeatRuns.$inferSelect, workspaceId: string, target: RemoteWorkspaceTarget): string | null {
|
||||
const profile = record(run.runnerProfileJson), state = record(profile.nativeProviderPackAdmissionState);
|
||||
const admission = record(profile.nativeProviderPackAdmission), workspace = record(profile.remoteWorkspace);
|
||||
if (!uuidPattern.test(target.leaseId ?? "") || state.schema !== "paperclip.native-provider-pack-admission-state.v1" || state.state !== "verified"
|
||||
|| typeof state.admissionEpoch !== "string" || !uuidPattern.test(state.admissionEpoch)
|
||||
|| admission.schema !== "paperclip.native-provider-pack-admission.v1" || admission.stage !== "session_admitted"
|
||||
|| admission.authority !== "core_verified_selection" || admission.runId !== run.id
|
||||
|| admission.companyId !== run.companyId || admission.agentId !== run.agentId
|
||||
|| workspace.workspaceId !== workspaceId || workspace.remoteCwd !== target.remoteCwd || workspace.leaseId !== target.leaseId
|
||||
|| typeof admission.providerPackDigest !== "string" || !digestPattern.test(admission.providerPackDigest)) return null;
|
||||
return admission.providerPackDigest;
|
||||
}
|
||||
|
||||
const reclaimScript = String.raw`
|
||||
const fs=require('node:fs/promises'),path=require('node:path'),crypto=require('node:crypto'),{constants}=require('node:fs');
|
||||
(async()=>{
|
||||
const [mode,root,source,expected,raw]=process.argv.slice(1),schema='paperclip.native-provider-cache-reclaim.v1';
|
||||
if(!['inventory','stage','reclaim'].includes(mode)||!path.isAbsolute(root)||!/^[a-f0-9-]{36}$/.test(source)
|
||||
||!/^sha256:[a-f0-9]{64}$/.test(expected)||await fs.realpath(root)!==root)throw Error('invalid_binding');
|
||||
const runtime=path.join(root,'.paperclip-runtime'),runner=path.join(runtime,'paperclip-runner'),pack=path.join(runner,'provider-pack');
|
||||
const sessions=path.join(runner,'sessions'),file=path.join(pack,'provider-pack.json');
|
||||
const stat=async p=>fs.lstat(p,{bigint:true}).catch(e=>{if(e.code==='ENOENT')return null;throw e;});
|
||||
const directory=async p=>{const x=await fs.lstat(p,{bigint:true});if(!x.isDirectory()||x.isSymbolicLink()||await fs.realpath(p)!==p)throw Error('unsafe_directory');return x;};
|
||||
const identity=x=>({dev:String(x.dev),ino:String(x.ino),generation:(x.birthtimeNs>0n?'b:'+x.birthtimeNs:'c:'+x.ctimeNs)});
|
||||
const same=(a,b,generation=true)=>a&&b&&['dev','ino',...(generation?['generation']:[])].every(k=>a[k]===b[k]);
|
||||
const rootIdentity=identity(await directory(root));await directory(runtime);await directory(runner);
|
||||
const packIdentity=identity(await directory(pack)),sessionIdentity=identity(await directory(sessions)),m=await stat(file);
|
||||
if(!m||!m.isFile()||m.isSymbolicLink()||m.nlink!==1n||m.size>524288n)throw Error('unsafe_manifest');
|
||||
const bytes=await fs.readFile(file),manifest=JSON.parse(bytes),manifestIdentity=identity(m);
|
||||
const canonical=v=>Array.isArray(v)?'['+v.map(canonical).join(',')+']':v&&typeof v==='object'?'{'+Object.entries(v).sort(([a],[b])=>a.localeCompare(b)).map(([k,x])=>JSON.stringify(k)+':'+canonical(x)).join(',')+'}':JSON.stringify(v);
|
||||
const manifestSha256=crypto.createHash('sha256').update(bytes).digest('hex');
|
||||
if(manifest.schema!=='paperclip-runner/remote-provider-pack/v1'||manifest.digest!==expected
|
||||
||expected!=='sha256:'+crypto.createHash('sha256').update(canonical(manifest.payload)).digest('hex'))throw Error('manifest_changed');
|
||||
const available=async()=>{try{const capacity=await fs.statfs(root),bytes=capacity.bavail*capacity.bsize;
|
||||
return Number.isSafeInteger(bytes)?Math.max(0,bytes):null;}catch{return null;}};
|
||||
const availableBytesBefore=await available();
|
||||
if(mode==='inventory'){
|
||||
const targets=[];
|
||||
for(const name of ['node_modules','provider-assets']){const p=path.join(pack,name),x=await stat(p);if(!x)continue;
|
||||
if(!x.isDirectory()||x.isSymbolicLink()||await fs.realpath(p)!==p)throw Error('unsafe_cache_directory');targets.push({name,...identity(x)});}
|
||||
console.log(JSON.stringify({schema:'paperclip.native-provider-cache-inventory.v1',sourceRunId:source,manifestDigest:expected,
|
||||
manifestSha256,root:rootIdentity,pack:packIdentity,manifest:manifestIdentity,sessions:sessionIdentity,
|
||||
targets,availableBytesBefore}));return;
|
||||
}
|
||||
const request=JSON.parse(raw),plan=request.inventory;
|
||||
if(!/^[a-f0-9-]{36}$/.test(request.preparationOperationId)||plan?.schema!=='paperclip.native-provider-cache-inventory.v1'
|
||||
||plan.sourceRunId!==source||plan.manifestDigest!==expected||plan.manifestSha256!==manifestSha256
|
||||
||!same(plan.root,rootIdentity)||!same(plan.pack,packIdentity,false)||!same(plan.manifest,manifestIdentity)||!same(plan.sessions,sessionIdentity)
|
||||
||!Array.isArray(plan.targets)||plan.targets.length>2||new Set(plan.targets.map(x=>x.name)).size!==plan.targets.length)throw Error('plan_changed');
|
||||
const markerName='.paperclip-reclaim-'+request.preparationOperationId+'.json';
|
||||
const stagedPath=item=>path.join(pack,item.name+'.paperclip-reclaim-'+request.preparationOperationId);
|
||||
const markerText=item=>JSON.stringify({preparationOperationId:request.preparationOperationId,source,expected,item});
|
||||
const markerState=async(p,item,partial=false)=>{const x=await stat(p);
|
||||
if(!x||!x.isFile()||x.isSymbolicLink()||x.nlink!==1n||x.size>4096n)throw Error('marker_changed');
|
||||
const bytes=await fs.readFile(p),text=bytes.toString();
|
||||
if(text!==markerText(item)&&!(partial&&markerText(item).startsWith(text)))throw Error('marker_changed');
|
||||
return {identity:identity(x),sha256:crypto.createHash('sha256').update(bytes).digest('hex')};};
|
||||
let freed=0n,visited=0;
|
||||
const markerSpaceAvailable=async()=>{const bytes=await available();return bytes===null?freed>=262144n:bytes>=262144;};
|
||||
const releaseBlocks=async root=>{
|
||||
const visit=async dir=>{await directory(dir);for await(const entry of await fs.opendir(dir)){
|
||||
if(++visited>8192||await markerSpaceAvailable())return;
|
||||
if(entry.name===markerName)continue;
|
||||
const p=path.join(dir,entry.name),x=await stat(p);if(!x||x.isSymbolicLink())continue;
|
||||
if(x.isDirectory()){await visit(p);continue;}
|
||||
if(!x.isFile()||x.nlink!==1n||x.size===0n||!(x.mode&0o222n))continue;
|
||||
let handle;
|
||||
try{handle=await fs.open(p,constants.O_RDWR|constants.O_NOFOLLOW);
|
||||
const opened=await handle.stat({bigint:true});if(!same(identity(x),identity(opened))||opened.nlink!==1n)throw Error('cache_changed');
|
||||
await handle.truncate(0);freed+=opened.blocks*512n;
|
||||
}catch(error){if(!['EACCES','EPERM','EROFS'].includes(error.code))throw error;}finally{await handle?.close();}
|
||||
}};await visit(root);};
|
||||
const fallback=plan.targets.filter(item=>item.generation.startsWith('c:'));
|
||||
for(const item of plan.targets){
|
||||
if(!['node_modules','provider-assets'].includes(item.name))throw Error('plan_changed');
|
||||
const p=path.join(pack,item.name),current=await stat(p);
|
||||
if(item.generation.startsWith('c:')){
|
||||
const staged=await stat(stagedPath(item));if(current&&staged)throw Error('cache_changed');
|
||||
if(staged){await directory(stagedPath(item));if(!same(item,identity(staged),false))throw Error('cache_changed');
|
||||
await markerState(path.join(stagedPath(item),markerName),item);}
|
||||
else if(current){await directory(p);const marker=await stat(path.join(p,markerName));
|
||||
if(!same(item,identity(current),!marker))throw Error('cache_changed');
|
||||
if(marker)await markerState(path.join(p,markerName),item,true);}
|
||||
}else if(current&&(!current.isDirectory()||current.isSymbolicLink()||await fs.realpath(p)!==p||!same(item,identity(current))))throw Error('cache_changed');
|
||||
}
|
||||
if(mode==='stage'){
|
||||
const staged=[];
|
||||
for(const item of plan.targets){if(await markerSpaceAvailable())break;
|
||||
const p=path.join(pack,item.name);
|
||||
if(await stat(p))await releaseBlocks(p);
|
||||
else if(item.generation.startsWith('c:')&&await stat(stagedPath(item)))await releaseBlocks(stagedPath(item));
|
||||
}
|
||||
for(const item of fallback){const p=path.join(pack,item.name),dest=stagedPath(item),current=await stat(p);
|
||||
if(current){const marker=path.join(p,markerName),existing=await stat(marker);
|
||||
if(!existing||((await fs.readFile(marker)).toString()!==markerText(item))){
|
||||
const handle=await fs.open(marker,constants.O_WRONLY|constants.O_NOFOLLOW|(existing?0:constants.O_CREAT|constants.O_EXCL),0o600);
|
||||
try{if(existing&&!same(identity(existing),identity(await handle.stat({bigint:true}))))throw Error('marker_changed');
|
||||
await handle.writeFile(markerText(item));await handle.sync();}finally{await handle.close();}
|
||||
}
|
||||
await fs.rename(p,dest);
|
||||
}
|
||||
const cache=await stat(dest);if(!cache)continue;
|
||||
await directory(dest);if(!same(item,identity(cache),false))throw Error('cache_changed');
|
||||
const marker=await markerState(path.join(dest,markerName),item);
|
||||
staged.push({name:item.name,cache:identity(cache),marker:marker.identity,markerSha256:marker.sha256});
|
||||
}
|
||||
console.log(JSON.stringify({preparationOperationId:request.preparationOperationId,staged}));return;
|
||||
}
|
||||
if(!Array.isArray(request.staged)||request.staged.length>2||new Set(request.staged.map(x=>x.name)).size!==request.staged.length)throw Error('plan_changed');
|
||||
for(const item of fallback){const witness=request.staged.find(x=>x.name===item.name),dest=stagedPath(item),current=await stat(dest);
|
||||
if(await stat(path.join(pack,item.name)))throw Error('cache_changed');
|
||||
if(!witness&&!current)continue;
|
||||
if(!witness||!current||!same(witness.cache,identity(current),false)||!same(item,identity(current),false))throw Error('cache_changed');
|
||||
const marker=await markerState(path.join(dest,markerName),item);
|
||||
if(!same(witness.marker,marker.identity)||witness.markerSha256!==marker.sha256)throw Error('marker_changed');
|
||||
}
|
||||
let changed=false;
|
||||
for(const item of plan.targets){
|
||||
if(item.generation.startsWith('c:')){const p=stagedPath(item);if(!await stat(p))continue;
|
||||
for await(const entry of await fs.opendir(p)){if(entry.name!==markerName)await fs.rm(path.join(p,entry.name),{recursive:true,force:false});}
|
||||
changed=true;continue;}
|
||||
const p=path.join(pack,item.name),current=await stat(p);if(!current)continue;
|
||||
if(!current.isDirectory()||current.isSymbolicLink()||!same(item,identity(current)))throw Error('cache_changed');
|
||||
await fs.rm(p,{recursive:true,force:false});changed=true;}
|
||||
if(!same(plan.sessions,identity(await directory(sessions))))throw Error('sessions_changed');
|
||||
console.log(JSON.stringify({schema,sourceRunId:source,manifestDigest:expected,
|
||||
preparationOperationId:request.preparationOperationId,status:'evicted',changed,sessionDirectoryPreserved:true,
|
||||
availableBytesBefore,availableBytesAfter:await available()}));
|
||||
})().catch(()=>{process.stderr.write('native_provider_cache_reclaim_unconfirmed');process.exitCode=1;});`;
|
||||
|
||||
function readInventory(value: unknown, runId: string, expected: string): Inventory {
|
||||
const parsed = inventoryCodec.parse(value);
|
||||
if (parsed.sourceRunId !== runId || parsed.manifestDigest !== expected
|
||||
|| new Set(parsed.targets.map(target => target.name)).size !== parsed.targets.length)
|
||||
throw new Error("native_provider_cache_reclaim_unconfirmed");
|
||||
return parsed;
|
||||
}
|
||||
|
||||
export async function reclaimInactiveNativeProviderCache(input: {
|
||||
db: Db; companyId: string; issueId: string; runId: string; workspaceId: string;
|
||||
target: RemoteWorkspaceTarget; assertOwned: () => Promise<void>;
|
||||
}): Promise<boolean> {
|
||||
await input.assertOwned();
|
||||
const [source] = await input.db.select({ run: heartbeatRuns, coordinator: nativeRunFinalizations })
|
||||
.from(heartbeatRuns).innerJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId),
|
||||
eq(nativeRunFinalizations.runId, heartbeatRuns.id)))
|
||||
.where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId),
|
||||
eq(nativeRunFinalizations.issueId, input.issueId))).limit(1);
|
||||
if (!source || source.run.status !== "failed" || source.run.runtimeMode !== "native"
|
||||
|| (source.run.nativeIssueId ?? source.run.issueId ?? source.run.contextSnapshot?.issueId) !== input.issueId
|
||||
|| source.coordinator.phase !== "terminal_failure" || source.coordinator.leaseOwner !== null
|
||||
|| record(source.coordinator.failureDetail).recoverable === true) return false;
|
||||
const expected = verifiedCacheDigest(source.run, input.workspaceId, input.target);
|
||||
if (!expected) return false;
|
||||
const scope = and(eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.issueId, input.issueId),
|
||||
eq(workspaceOperations.heartbeatRunId, input.runId), eq(workspaceOperations.executionWorkspaceId, input.workspaceId));
|
||||
const execute = async (mode: "inventory" | "stage" | "reclaim",
|
||||
preparation?: { preparationOperationId: string; inventory: Inventory; staged?: Staged }) => {
|
||||
await input.assertOwned();
|
||||
const response = await input.target.runner!.execute({ command: "node", args: ["-e", reclaimScript,
|
||||
mode, input.target.remoteCwd, input.runId, expected, JSON.stringify(preparation ?? null)],
|
||||
cwd: input.target.remoteCwd, bypassSession: true, env: { NODE_OPTIONS: "", NODE_PATH: "" }, timeoutMs: 45_000 });
|
||||
await input.assertOwned();
|
||||
if (response.exitCode !== 0 || response.timedOut || Buffer.byteLength(response.stdout) > 16_384)
|
||||
throw new Error("native_provider_cache_reclaim_unconfirmed");
|
||||
return record(JSON.parse(response.stdout));
|
||||
};
|
||||
const startedAt = new Date();
|
||||
let confirmed = false, changed = false, preparationOperationId: string | null = null;
|
||||
let metadata: Record<string, unknown> = { schema, manifestDigest: expected, outcome: "unconfirmed" };
|
||||
try {
|
||||
let [prepared] = await input.db.select().from(workspaceOperations)
|
||||
.where(and(scope, eq(workspaceOperations.phase, preparePhase), eq(workspaceOperations.status, "succeeded")))
|
||||
.orderBy(asc(workspaceOperations.startedAt), asc(workspaceOperations.id)).limit(1);
|
||||
if (!prepared) {
|
||||
const inventory = readInventory(await execute("inventory"), input.runId, expected);
|
||||
if (!inventory.targets.length) return false;
|
||||
await input.assertOwned();
|
||||
[prepared] = await withDedicatedDbConnection(input.db, journal => journal.transaction(async tx => {
|
||||
await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`);
|
||||
return tx.insert(workspaceOperations).values({ companyId: input.companyId, issueId: input.issueId,
|
||||
heartbeatRunId: input.runId, executionWorkspaceId: input.workspaceId, phase: preparePhase,
|
||||
status: "succeeded", startedAt, finishedAt: new Date(),
|
||||
metadata: { schema: prepareSchema, intentOnly: true, leaseId: input.target.leaseId,
|
||||
remoteCwd: input.target.remoteCwd, inventory },
|
||||
}).returning();
|
||||
}));
|
||||
}
|
||||
if (!prepared || prepared.metadata?.schema !== prepareSchema || prepared.metadata.intentOnly !== true
|
||||
|| prepared.metadata.leaseId !== input.target.leaseId || prepared.metadata.remoteCwd !== input.target.remoteCwd)
|
||||
throw new Error("native_provider_cache_reclaim_unconfirmed");
|
||||
const inventory = readInventory(prepared.metadata.inventory, input.runId, expected);
|
||||
preparationOperationId = prepared.id;
|
||||
metadata = { ...metadata, preparationOperationId };
|
||||
const [completed] = await input.db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(scope,
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim"), eq(workspaceOperations.status, "succeeded"),
|
||||
sql`${workspaceOperations.metadata}->>'preparationOperationId' = ${preparationOperationId}`,
|
||||
sql`${workspaceOperations.metadata}->>'outcome' = 'evicted'`)).limit(1);
|
||||
if (completed) return false;
|
||||
let staged: Staged = [];
|
||||
if (inventory.targets.some(target => target.generation.startsWith("c:"))) {
|
||||
if (prepared.metadata.staged === undefined) {
|
||||
const response = await execute("stage", { preparationOperationId, inventory });
|
||||
if (response.preparationOperationId !== preparationOperationId) throw new Error("native_provider_cache_reclaim_unconfirmed");
|
||||
staged = stagedCodec.parse(response.staged);
|
||||
if (new Set(staged.map(item => item.name)).size !== staged.length || staged.some(item => !inventory.targets.some(target =>
|
||||
target.name === item.name && target.generation.startsWith("c:") && target.dev === item.cache.dev && target.ino === item.cache.ino)))
|
||||
throw new Error("native_provider_cache_reclaim_unconfirmed");
|
||||
await input.assertOwned();
|
||||
[prepared] = await withDedicatedDbConnection(input.db, journal => journal.transaction(async tx => {
|
||||
await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`);
|
||||
return tx.update(workspaceOperations).set({ metadata: { ...prepared.metadata, staged } })
|
||||
.where(and(scope, eq(workspaceOperations.id, preparationOperationId!))).returning();
|
||||
}));
|
||||
if (!prepared) throw new Error("native_provider_cache_reclaim_unconfirmed");
|
||||
}
|
||||
staged = stagedCodec.parse(prepared.metadata?.staged);
|
||||
}
|
||||
const receipt = await execute("reclaim", { preparationOperationId, inventory, staged });
|
||||
if (receipt.schema !== schema || receipt.sourceRunId !== input.runId || receipt.manifestDigest !== expected
|
||||
|| receipt.preparationOperationId !== preparationOperationId || receipt.status !== "evicted"
|
||||
|| typeof receipt.changed !== "boolean" || receipt.sessionDirectoryPreserved !== true)
|
||||
throw new Error("native_provider_cache_reclaim_unconfirmed");
|
||||
metadata = { schema, manifestDigest: expected, preparationOperationId, outcome: receipt.status, changed: receipt.changed,
|
||||
sessionDirectoryPreserved: true,
|
||||
availableBytesBefore: typeof receipt.availableBytesBefore === "number" ? receipt.availableBytesBefore : null,
|
||||
availableBytesAfter: typeof receipt.availableBytesAfter === "number" ? receipt.availableBytesAfter : null };
|
||||
confirmed = true;
|
||||
changed = receipt.changed;
|
||||
} catch {}
|
||||
await input.db.insert(workspaceOperations).values({ companyId: input.companyId, issueId: input.issueId,
|
||||
heartbeatRunId: input.runId, executionWorkspaceId: input.workspaceId, phase: "native_provider_cache_reclaim",
|
||||
status: confirmed ? "succeeded" : "failed", startedAt, finishedAt: new Date(), metadata });
|
||||
return changed;
|
||||
}
|
||||
@@ -45,8 +45,8 @@ import { requestTaskRetry } from "./task-retry.js";
|
||||
import { readFreshWorkspaceReadiness } from "./fresh-workspace-continuation.js";
|
||||
import { materializeIsolatedTaskDirectory } from "./isolated-task-directory.js";
|
||||
import { execFile } from "node:child_process";
|
||||
import { generateKeyPairSync, randomBytes, randomUUID, sign } from "node:crypto";
|
||||
import { chmod, cp, lstat, mkdir, mkdtemp, readFile, readdir, realpath, rename, rm, writeFile } from "node:fs/promises";
|
||||
import { createHash, generateKeyPairSync, randomBytes, randomUUID, sign } from "node:crypto";
|
||||
import { chmod, cp, lstat, mkdir, mkdtemp, readFile, readdir, realpath, rename, rm, symlink, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import express from "express";
|
||||
@@ -81,6 +81,9 @@ import { executionWorkspaceSnapshots } from "@paperclipai/db";
|
||||
import { createSavedWorkspaceLoader, prepareRemoteWorkspace, readRemoteWorkspaceAuthority, reserveRemoteWorkspace, saveRemoteWorkspace } from "./remote-workspaces.js";
|
||||
import { createReusableLeaseHandoff } from "./environment-lease-handoff.js";
|
||||
import { preserveRemoteWorkspacePreparationFailure } from "./native-runtime/native-workspace-finalization-state.js";
|
||||
import { canonicalProviderPackJson } from "./native-runtime/remote-provider-pack-verification.js";
|
||||
import { REMOTE_WORKSPACE_SANDBOX_DISK_FULL } from "./remote-workspace-transport.js";
|
||||
import { withDedicatedDbConnection } from "@paperclipai/db";
|
||||
import { getExecutionBlocker } from "./execution-blocker.js";
|
||||
import { buildExecutionContinuation } from "./execution-continuation.js";
|
||||
import { createHeartbeatRunOutput } from "./heartbeat/run-output.js";
|
||||
@@ -698,6 +701,485 @@ describe("automatic retained workspace recovery", () => {
|
||||
return { ...f, ...(sourceLayout === "published" ? { remote: remoteCwd } : {}), source, workspaceId, resultId, git, commit };
|
||||
}
|
||||
|
||||
async function pressuredToolCache(pressure: "transfer" | "sandbox-write" = "transfer") {
|
||||
const f = await acceptedRemoteExport();
|
||||
const runner = path.join(f.remote, ".paperclip-runtime", "paperclip-runner");
|
||||
const pack = path.join(runner, "provider-pack"), sessions = path.join(runner, "sessions", "retained");
|
||||
await mkdir(path.join(pack, "node_modules"), { recursive: true });
|
||||
await mkdir(path.join(pack, "provider-assets")); await mkdir(sessions, { recursive: true });
|
||||
await writeFile(path.join(pack, "node_modules", "installed-tool"), "rebuildable dependency\n");
|
||||
await writeFile(path.join(pack, "provider-assets", "installed-agent"), "rebuildable provider\n");
|
||||
await writeFile(path.join(sessions, "pending-receipt.json"), "uncertain action history\n");
|
||||
const payload = { target: { platform: "linux", architecture: "x64" } };
|
||||
const digest = `sha256:${createHash("sha256").update(canonicalProviderPackJson(payload)).digest("hex")}`;
|
||||
await writeFile(path.join(pack, "provider-pack.json"), JSON.stringify({
|
||||
schema: "paperclip-runner/remote-provider-pack/v1", digest, payload,
|
||||
}));
|
||||
const before = await run(f.runId);
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...before.runnerProfileJson,
|
||||
nativeProviderPackAdmissionState: { schema: "paperclip.native-provider-pack-admission-state.v1",
|
||||
state: "verified", admissionEpoch: randomUUID() },
|
||||
nativeProviderPackAdmission: { schema: "paperclip.native-provider-pack-admission.v1", stage: "session_admitted",
|
||||
authority: "core_verified_selection", runId: f.runId, companyId: f.companyId, agentId: before.agentId,
|
||||
providerPackDigest: digest },
|
||||
} }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const target = targets.get(f.lease.id)!, syncOut = target.runner!.syncOut!, execute = target.runner!.execute;
|
||||
let transfers = 0, sandboxWriteFailures = 0;
|
||||
if (pressure === "sandbox-write") target.runner!.execute = async input => {
|
||||
const fault = `(()=>{const fs=require('node:fs/promises'),sync=require('node:fs');
|
||||
const full=()=>${JSON.stringify([path.join(pack, "node_modules"), path.join(pack, "provider-assets")])}.some(p=>sync.existsSync(p));
|
||||
const statfs=fs.statfs;fs.statfs=async(...args)=>{const value=await statfs(...args);
|
||||
return full()?{...value,bavail:typeof value.bavail==='bigint'?0n:0}:value;};
|
||||
for(const name of ['mkdir','writeFile','appendFile','mkdtemp','copyFile','rename','open']){
|
||||
const original=fs[name];fs[name]=async(...args)=>{
|
||||
const writes=name!=='open'||typeof args[1]!=='string'||/[wa+]/.test(args[1]);
|
||||
if(full()&&writes&&!(name==='mkdir'&&sync.existsSync(args[0])))throw Object.assign(new Error('No space left on device'),{code:'ENOSPC'});
|
||||
return original(...args);};}})();`;
|
||||
const args = input.command === "node" && input.args?.[0] === "-e"
|
||||
? ["-e", fault + input.args[1], ...input.args.slice(2)] : input.args;
|
||||
const response = await execute({ ...input, args });
|
||||
if (response.stderr === REMOTE_WORKSPACE_SANDBOX_DISK_FULL) sandboxWriteFailures++;
|
||||
return response;
|
||||
};
|
||||
target.runner!.syncOut = async operations => {
|
||||
transfers++;
|
||||
if (pressure === "transfer" && (await Promise.all(["node_modules", "provider-assets"].map(name =>
|
||||
lstat(path.join(pack, name)).then(() => true, () => false)))).some(Boolean))
|
||||
throw new Error(REMOTE_WORKSPACE_SANDBOX_DISK_FULL);
|
||||
return syncOut(operations);
|
||||
};
|
||||
return { ...f, runner, pack, sessions, digest, transfers: () => transfers,
|
||||
sandboxWriteFailures: () => sandboxWriteFailures };
|
||||
}
|
||||
|
||||
it("recovers when sandbox writes fail before any archive can be created", async () => {
|
||||
const f = await pressuredToolCache("sandbox-write");
|
||||
const target = targets.get(f.lease.id)!, execute = target.runner!.execute;
|
||||
const staging = path.join(f.remote, ".paperclip-runtime", "workspace-snapshots");
|
||||
const before = await readdir(staging).catch(() => []);
|
||||
let committedPreparation = false;
|
||||
target.runner!.execute = async input => {
|
||||
if (input.args?.[2] === "reclaim") {
|
||||
const records = await withDedicatedDbConnection(db, observer => observer.select().from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, f.companyId), eq(workspaceOperations.heartbeatRunId, f.runId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim_prepare"))));
|
||||
expect(records).toHaveLength(1);
|
||||
expect(records[0]).toMatchObject({ status: "succeeded", metadata: { intentOnly: true,
|
||||
inventory: { availableBytesBefore: 0 } } });
|
||||
expect(await readdir(staging).catch(() => [])).toEqual(before);
|
||||
committedPreparation = true;
|
||||
}
|
||||
return execute(input);
|
||||
};
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(committedPreparation).toBe(true);
|
||||
expect(f.sandboxWriteFailures()).toBe(1);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).toBe("recovered");
|
||||
expect(f.transfers()).toBe(1);
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
const authority = readRemoteWorkspaceAuthority(workspace.metadata)!;
|
||||
const saved = await createSavedWorkspaceLoader(db)({ companyId: f.companyId, workspaceId: f.workspaceId,
|
||||
snapshotId: authority.snapshotId! });
|
||||
try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); }
|
||||
finally { await saved.cleanup(); }
|
||||
const [prepared] = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, f.companyId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim_prepare")));
|
||||
expect(readSystemRecovery((await action(f.actionId)).evidence)?.preservationOperationIds).not.toContain(prepared.id);
|
||||
expect(await readFile(path.join(f.sessions, "pending-receipt.json"), "utf8")).toBe("uncertain action history\n");
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0);
|
||||
}, 30_000);
|
||||
|
||||
it("reclaims only a verified inactive tool installation before retrying the same workspace save", async () => {
|
||||
const f = await pressuredToolCache();
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).toBe("recovered");
|
||||
expect(f.transfers()).toBe(2);
|
||||
await expect(lstat(path.join(f.pack, "node_modules"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
await expect(lstat(path.join(f.pack, "provider-assets"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(JSON.parse(await readFile(path.join(f.pack, "provider-pack.json"), "utf8")).digest).toBe(f.digest);
|
||||
expect(await readFile(path.join(f.sessions, "pending-receipt.json"), "utf8")).toBe("uncertain action history\n");
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
const authority = readRemoteWorkspaceAuthority(workspace.metadata)!;
|
||||
const saved = await createSavedWorkspaceLoader(db)({ companyId: f.companyId, workspaceId: f.workspaceId,
|
||||
snapshotId: authority.snapshotId! });
|
||||
try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); }
|
||||
finally { await saved.cleanup(); }
|
||||
const audits = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, f.companyId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim")));
|
||||
expect(audits).toHaveLength(1);
|
||||
expect(audits[0]).toMatchObject({ status: "succeeded", heartbeatRunId: f.runId,
|
||||
metadata: { manifestDigest: f.digest, changed: true, sessionDirectoryPreserved: true } });
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0);
|
||||
}, 30_000);
|
||||
|
||||
it("keeps lost cache acknowledgements separate from later verified workspace preservation", async () => {
|
||||
const f = await pressuredToolCache(), target = targets.get(f.lease.id)!, execute = target.runner!.execute;
|
||||
let lost = false;
|
||||
target.runner!.execute = async input => {
|
||||
const result = await execute(input);
|
||||
if (!lost && input.args?.[2] === "reclaim") {
|
||||
lost = true;
|
||||
return { ...result, exitCode: null, timedOut: true, stdout: "", stderr: "" };
|
||||
}
|
||||
return result;
|
||||
};
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(lost).toBe(true);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).not.toBe("recovered");
|
||||
expect(f.transfers()).toBe(1);
|
||||
const [audit] = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, f.companyId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim")));
|
||||
expect(audit).toMatchObject({ status: "failed", metadata: { outcome: "unconfirmed" } });
|
||||
await makeRetryDue(f.actionId);
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).toBe("recovered");
|
||||
expect(f.transfers()).toBe(2);
|
||||
expect(await readFile(path.join(f.sessions, "pending-receipt.json"), "utf8")).toBe("uncertain action history\n");
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
|
||||
}, 30_000);
|
||||
|
||||
it("does not delete caches when Core cannot commit the prepared journal", async () => {
|
||||
const f = await pressuredToolCache("sandbox-write"), trigger = `cache_prepare_${randomUUID().replaceAll("-", "")}`;
|
||||
await db.execute(sql.raw(`CREATE FUNCTION ${trigger}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN
|
||||
RAISE EXCEPTION 'test cache preparation write failed'; END $$`));
|
||||
try {
|
||||
await db.execute(sql.raw(`CREATE TRIGGER ${trigger} BEFORE INSERT ON workspace_operations FOR EACH ROW
|
||||
WHEN (NEW.company_id = '${f.companyId}'::uuid AND NEW.phase = 'native_provider_cache_reclaim_prepare')
|
||||
EXECUTE FUNCTION ${trigger}()`));
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).not.toBe("recovered");
|
||||
expect(f.sandboxWriteFailures()).toBe(1);
|
||||
expect(f.transfers()).toBe(0);
|
||||
expect(await readFile(path.join(f.pack, "node_modules", "installed-tool"), "utf8")).toBe("rebuildable dependency\n");
|
||||
expect(await readFile(path.join(f.pack, "provider-assets", "installed-agent"), "utf8")).toBe("rebuildable provider\n");
|
||||
expect(await readFile(path.join(f.sessions, "pending-receipt.json"), "utf8")).toBe("uncertain action history\n");
|
||||
const records = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, f.companyId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim_prepare")));
|
||||
expect(records).toHaveLength(0);
|
||||
} finally {
|
||||
await db.execute(sql.raw(`DROP TRIGGER IF EXISTS ${trigger} ON workspace_operations`));
|
||||
await db.execute(sql.raw(`DROP FUNCTION ${trigger}()`));
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
it("does not reclaim a replacement after an acknowledged cleanup and a later save failure", async () => {
|
||||
const f = await pressuredToolCache(), target = targets.get(f.lease.id)!, syncOut = target.runner!.syncOut!;
|
||||
let transfers = 0;
|
||||
target.runner!.syncOut = async operations => {
|
||||
if (++transfers === 2) throw new Error("controlled later transfer failure");
|
||||
return syncOut(operations);
|
||||
};
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).not.toBe("recovered");
|
||||
const [completed] = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, f.companyId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim"), eq(workspaceOperations.status, "succeeded")));
|
||||
expect(completed.metadata).toMatchObject({ outcome: "evicted", changed: true });
|
||||
await mkdir(path.join(f.pack, "node_modules"));
|
||||
await writeFile(path.join(f.pack, "node_modules", "replacement-tool"), "replacement installation\n");
|
||||
let repeatedDeletion = false;
|
||||
const execute = target.runner!.execute;
|
||||
target.runner!.execute = async input => {
|
||||
if (input.args?.[2] === "reclaim") repeatedDeletion = true;
|
||||
return execute(input);
|
||||
};
|
||||
await makeRetryDue(f.actionId);
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(repeatedDeletion).toBe(false);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).not.toBe("recovered");
|
||||
expect(await readFile(path.join(f.pack, "node_modules", "replacement-tool"), "utf8")).toBe("replacement installation\n");
|
||||
expect(await readFile(path.join(f.sessions, "pending-receipt.json"), "utf8")).toBe("uncertain action history\n");
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
}, 30_000);
|
||||
|
||||
it.each(["unchanged", "replaced"])("resumes a Core-prepared deletion only for an %s installation", async installation => {
|
||||
const f = await pressuredToolCache(), target = targets.get(f.lease.id)!, execute = target.runner!.execute;
|
||||
const interruptedTarget = await realpath(path.join(f.pack, "node_modules"));
|
||||
let interrupted = false;
|
||||
target.runner!.execute = async input => {
|
||||
if (!interrupted && input.args?.[2] === "reclaim") {
|
||||
interrupted = true;
|
||||
const fault = `(()=>{const fs=require('node:fs/promises'),remove=fs.rm;
|
||||
fs.rm=async(p,options)=>{const result=await remove(p,options);
|
||||
if(p===${JSON.stringify(interruptedTarget)})process.exit(77);return result;};})();`;
|
||||
return execute({ ...input, args: ["-e", fault + input.args[1], ...input.args.slice(2)] });
|
||||
}
|
||||
return execute(input);
|
||||
};
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(interrupted).toBe(true);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).not.toBe("recovered");
|
||||
await expect(lstat(path.join(f.pack, "node_modules"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(await readFile(path.join(f.pack, "provider-assets", "installed-agent"), "utf8")).toBe("rebuildable provider\n");
|
||||
if (installation === "replaced") {
|
||||
await mkdir(path.join(f.pack, "node_modules"));
|
||||
await writeFile(path.join(f.pack, "node_modules", "replacement-tool"), "replacement installation\n");
|
||||
}
|
||||
await makeRetryDue(f.actionId);
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
if (installation === "unchanged") {
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).toBe("recovered");
|
||||
expect(f.transfers()).toBe(3);
|
||||
await expect(lstat(path.join(f.pack, "provider-assets"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
} else {
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).not.toBe("recovered");
|
||||
expect(f.transfers()).toBe(2);
|
||||
expect(await readFile(path.join(f.pack, "node_modules", "replacement-tool"), "utf8")).toBe("replacement installation\n");
|
||||
expect(await readFile(path.join(f.pack, "provider-assets", "installed-agent"), "utf8")).toBe("rebuildable provider\n");
|
||||
}
|
||||
const preparations = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, f.companyId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim_prepare")));
|
||||
expect(preparations).toHaveLength(1);
|
||||
expect(await readFile(path.join(f.sessions, "pending-receipt.json"), "utf8")).toBe("uncertain action history\n");
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
}, 30_000);
|
||||
|
||||
it.each(["unchanged", "full_disk", "replacement", "staged_replacement", "marker_replaced"])(
|
||||
"resumes deletion interrupted inside a cache without directory creation times: %s", async scenario => {
|
||||
const f = await pressuredToolCache(), target = targets.get(f.lease.id)!, execute = target.runner!.execute;
|
||||
await writeFile(path.join(f.pack, "node_modules", "installed-tool"), Buffer.alloc(1024 * 1024, 7));
|
||||
await writeFile(path.join(f.pack, "node_modules", "remaining-tool"), "remaining rebuildable dependency\n");
|
||||
const interruptedTarget = await realpath(path.join(f.pack, "node_modules"));
|
||||
let interrupted = false, cleanupConfirmed = false;
|
||||
const syncOut = target.runner!.syncOut!;
|
||||
target.runner!.syncOut = async operations => {
|
||||
if (!cleanupConfirmed) throw new Error(REMOTE_WORKSPACE_SANDBOX_DISK_FULL);
|
||||
return syncOut(operations);
|
||||
};
|
||||
target.runner!.execute = async input => {
|
||||
if (input.command !== "node" || input.args?.[0] !== "-e") return execute(input);
|
||||
let fault = `(()=>{const fs=require('node:fs/promises'),stat=fs.lstat,open=fs.open;
|
||||
fs.lstat=async(...args)=>{const value=await stat(...args);
|
||||
if(args[1]?.bigint)value.birthtimeNs=0n;return value;};
|
||||
fs.open=async(...args)=>{const handle=await open(...args),stat=handle.stat.bind(handle);
|
||||
handle.stat=async(...args)=>{const value=await stat(...args);if(args[0]?.bigint)value.birthtimeNs=0n;return value;};return handle;};})();`;
|
||||
if (scenario === "full_disk") fault += `(()=>{const fs=require('node:fs/promises'),sync=require('node:fs');
|
||||
const full=()=>{try{return sync.statSync(${JSON.stringify(path.join(f.pack, "node_modules", "installed-tool"))}).size>0;}catch{return false;}};
|
||||
const statfs=fs.statfs;fs.statfs=async(...args)=>{const value=await statfs(...args);return full()?{...value,bavail:0}:value;};
|
||||
for(const name of ['mkdir','writeFile','appendFile','mkdtemp','copyFile','rename','open']){
|
||||
const original=fs[name];fs[name]=async(...args)=>{const writes=name!=='open'||(typeof args[1]==='number'
|
||||
?!!(args[1]&(sync.constants.O_CREAT|sync.constants.O_TRUNC)):typeof args[1]==='string'&&/[wa]/.test(args[1]));
|
||||
if(full()&&writes&&!(name==='mkdir'&&sync.existsSync(args[0])))throw Object.assign(new Error('No space left on device'),{code:'ENOSPC'});
|
||||
return original(...args);};}})();`;
|
||||
if (input.args[2] === "reclaim") {
|
||||
const [prepared] = await withDedicatedDbConnection(db, observer => observer.select().from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, f.companyId), eq(workspaceOperations.phase, "native_provider_cache_reclaim_prepare"))));
|
||||
expect(prepared.metadata?.staged).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ name: "node_modules", marker: expect.objectContaining({ generation: expect.stringMatching(/^c:/) }) }),
|
||||
]));
|
||||
}
|
||||
if (!interrupted && input.args[2] === "reclaim") {
|
||||
interrupted = true;
|
||||
fault += `(()=>{const fs=require('node:fs/promises'),path=require('node:path'),remove=fs.rm;
|
||||
fs.rm=async(p,options)=>{if(p===${JSON.stringify(interruptedTarget)}||p.startsWith(${JSON.stringify(interruptedTarget + ".paperclip-reclaim-")})){
|
||||
await fs.unlink(path.join(p===${JSON.stringify(interruptedTarget)}?p:path.dirname(p),'installed-tool'));process.exit(77);
|
||||
}return remove(p,options);};})();`;
|
||||
}
|
||||
const result = await execute({ ...input, args: ["-e", fault + input.args[1], ...input.args.slice(2)] });
|
||||
if (input.args[2] === "reclaim" && result.exitCode === 0) cleanupConfirmed = true;
|
||||
return result;
|
||||
};
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(interrupted).toBe(true);
|
||||
expect(cleanupConfirmed).toBe(false);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).not.toBe("recovered");
|
||||
const [prepared] = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, f.companyId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim_prepare")));
|
||||
const staged = path.join(f.pack, `node_modules.paperclip-reclaim-${prepared.id}`);
|
||||
const interruptedDirectory = await lstat(staged).then(() => staged, () => path.join(f.pack, "node_modules"));
|
||||
await expect(lstat(path.join(interruptedDirectory, "installed-tool"))).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect((await lstat(path.join(interruptedDirectory, "remaining-tool"))).isFile()).toBe(true);
|
||||
const remainingBeforeRetry = await readFile(path.join(interruptedDirectory, "remaining-tool"), "utf8");
|
||||
if (scenario === "replacement") {
|
||||
await mkdir(path.join(f.pack, "node_modules"));
|
||||
await writeFile(path.join(f.pack, "node_modules", "replacement-tool"), "replacement installation\n");
|
||||
}
|
||||
if (scenario === "staged_replacement") {
|
||||
await rename(staged, `${staged}.original`);
|
||||
await mkdir(staged);
|
||||
await writeFile(path.join(staged, "replacement-tool"), "replacement installation\n");
|
||||
}
|
||||
if (scenario === "marker_replaced") {
|
||||
const marker = path.join(staged, `.paperclip-reclaim-${prepared.id}.json`), bytes = await readFile(marker);
|
||||
await rename(marker, `${marker}.original`);
|
||||
await writeFile(marker, bytes);
|
||||
}
|
||||
await makeRetryDue(f.actionId);
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
if (["unchanged", "full_disk"].includes(scenario)) {
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).toBe("recovered");
|
||||
expect(cleanupConfirmed).toBe(true);
|
||||
expect(await readdir(staged)).toEqual([`.paperclip-reclaim-${prepared.id}.json`]);
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
const authority = readRemoteWorkspaceAuthority(workspace.metadata)!;
|
||||
const saved = await createSavedWorkspaceLoader(db)({ companyId: f.companyId, workspaceId: f.workspaceId, snapshotId: authority.snapshotId! });
|
||||
try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); }
|
||||
finally { await saved.cleanup(); }
|
||||
} else {
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).not.toBe("recovered");
|
||||
expect(cleanupConfirmed).toBe(false);
|
||||
if (scenario === "replacement") expect(await readFile(path.join(f.pack, "node_modules", "replacement-tool"), "utf8")).toBe("replacement installation\n");
|
||||
if (scenario === "staged_replacement") expect(await readFile(path.join(staged, "replacement-tool"), "utf8")).toBe("replacement installation\n");
|
||||
else expect(await readFile(path.join(staged, "remaining-tool"), "utf8")).toBe(remainingBeforeRetry);
|
||||
}
|
||||
expect(await readFile(path.join(f.sessions, "pending-receipt.json"), "utf8")).toBe("uncertain action history\n");
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0);
|
||||
}, 30_000);
|
||||
|
||||
it.each(["space_available", "later_cache", "denied_open"])("recovers read-only provider assets without directory creation times: %s", async scenario => {
|
||||
const f = await pressuredToolCache(), target = targets.get(f.lease.id)!, execute = target.runner!.execute;
|
||||
const assets = path.join(f.pack, "provider-assets"), executable = path.join(assets, "cursor-executable");
|
||||
const metadata = path.join(assets, "cursor-metadata.json"), writable = path.join(assets, "installed-agent");
|
||||
await writeFile(executable, "immutable executable\n"); await chmod(executable, 0o500);
|
||||
await writeFile(metadata, "immutable metadata\n"); await chmod(metadata, 0o400);
|
||||
await writeFile(writable, Buffer.alloc(1024 * 1024, 7));
|
||||
const dependency = path.join(f.pack, "node_modules", "installed-tool");
|
||||
if (scenario === "later_cache") await chmod(dependency, 0o500);
|
||||
let inspected = false;
|
||||
target.runner!.execute = async input => {
|
||||
if (input.command !== "node" || input.args?.[0] !== "-e") return execute(input);
|
||||
if (input.args[2] === "reclaim") {
|
||||
const [prepared] = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, f.companyId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim_prepare")));
|
||||
const staged = path.join(f.pack, `provider-assets.paperclip-reclaim-${prepared.id}`);
|
||||
expect(await readFile(path.join(staged, "cursor-executable"), "utf8")).toBe("immutable executable\n");
|
||||
expect((await lstat(path.join(staged, "cursor-executable"))).mode & 0o777).toBe(0o500);
|
||||
expect(await readFile(path.join(staged, "cursor-metadata.json"), "utf8")).toBe("immutable metadata\n");
|
||||
expect((await lstat(path.join(staged, "cursor-metadata.json"))).mode & 0o777).toBe(0o400);
|
||||
if (scenario === "space_available") expect(await readFile(path.join(staged, "installed-agent"))).toEqual(Buffer.alloc(1024 * 1024, 7));
|
||||
inspected = true;
|
||||
}
|
||||
const fault = `(()=>{const fs=require('node:fs/promises'),sync=require('node:fs'),stat=fs.lstat,open=fs.open,statfs=fs.statfs;
|
||||
const full=()=>${JSON.stringify(scenario)}!=='space_available'&&(()=>{try{return sync.statSync(${JSON.stringify(writable)}).size>0;}catch{return false;}})();
|
||||
fs.statfs=async(...args)=>{const value=await statfs(...args);return full()?{...value,bavail:0}:value;};
|
||||
fs.lstat=async(...args)=>{const value=await stat(...args);if(args[1]?.bigint)value.birthtimeNs=0n;return value;};
|
||||
fs.open=async(...args)=>{if(typeof args[1]==='number'&&(args[1]&sync.constants.O_RDWR)){
|
||||
const value=sync.lstatSync(args[0]);if(!(value.mode&0o222)||(${JSON.stringify(scenario)}==='denied_open'&&args[0].endsWith('/node_modules/installed-tool')))
|
||||
throw Object.assign(new Error('Permission denied'),{code:'EACCES'});
|
||||
if(${JSON.stringify(scenario)}==='space_available')throw Error('unnecessary_cache_write');
|
||||
}
|
||||
const creates=typeof args[1]==='number'?!!(args[1]&(sync.constants.O_CREAT|sync.constants.O_TRUNC)):/[wa]/.test(args[1]);
|
||||
if(full()&&creates)throw Object.assign(new Error('No space left on device'),{code:'ENOSPC'});
|
||||
const handle=await open(...args),stat=handle.stat.bind(handle);
|
||||
handle.stat=async(...args)=>{const value=await stat(...args);if(args[0]?.bigint)value.birthtimeNs=0n;return value;};return handle;};})();`;
|
||||
return execute({ ...input, args: ["-e", fault + input.args[1], ...input.args.slice(2)] });
|
||||
};
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(inspected).toBe(true);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).toBe("recovered");
|
||||
expect(await readFile(path.join(f.sessions, "pending-receipt.json"), "utf8")).toBe("uncertain action history\n");
|
||||
const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId));
|
||||
const authority = readRemoteWorkspaceAuthority(workspace.metadata)!;
|
||||
const saved = await createSavedWorkspaceLoader(db)({ companyId: f.companyId, workspaceId: f.workspaceId, snapshotId: authority.snapshotId! });
|
||||
try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); }
|
||||
finally { await saved.cleanup(); }
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0);
|
||||
}, 30_000);
|
||||
|
||||
it.each(["acknowledgement", "journal", "marker_write"])("resumes cache staging after its %s is interrupted", async failure => {
|
||||
const f = await pressuredToolCache(), target = targets.get(f.lease.id)!, execute = target.runner!.execute;
|
||||
await writeFile(path.join(f.pack, "node_modules", "installed-tool"), Buffer.alloc(1024 * 1024, 7));
|
||||
await writeFile(path.join(f.pack, "node_modules", "remaining-tool"), "remaining rebuildable dependency\n");
|
||||
let lost = false, reclaimCalls = 0, cleanupConfirmed = false;
|
||||
const syncOut = target.runner!.syncOut!, trigger = `cache_stage_${randomUUID().replaceAll("-", "")}`;
|
||||
target.runner!.syncOut = async operations => {
|
||||
if (!cleanupConfirmed) throw new Error(REMOTE_WORKSPACE_SANDBOX_DISK_FULL);
|
||||
return syncOut(operations);
|
||||
};
|
||||
target.runner!.execute = async input => {
|
||||
if (input.command !== "node" || input.args?.[0] !== "-e") return execute(input);
|
||||
let fault = `(()=>{const fs=require('node:fs/promises'),stat=fs.lstat,open=fs.open;
|
||||
fs.lstat=async(...args)=>{const value=await stat(...args);if(args[1]?.bigint)value.birthtimeNs=0n;return value;};
|
||||
fs.open=async(...args)=>{const handle=await open(...args),stat=handle.stat.bind(handle);
|
||||
handle.stat=async(...args)=>{const value=await stat(...args);if(args[0]?.bigint)value.birthtimeNs=0n;return value;};return handle;};})();`;
|
||||
if (failure === "marker_write" && !lost && input.args[2] === "stage") {
|
||||
lost = true;
|
||||
fault += `(()=>{const fs=require('node:fs/promises'),open=fs.open;
|
||||
fs.open=async(...args)=>{const handle=await open(...args),write=handle.writeFile.bind(handle);
|
||||
if(args[0].includes('/.paperclip-reclaim-'))handle.writeFile=async bytes=>{await write(bytes.slice(0,32));process.exit(77);};
|
||||
return handle;};})();`;
|
||||
}
|
||||
const result = await execute({ ...input, args: ["-e", fault + input.args[1], ...input.args.slice(2)] });
|
||||
if (input.args[2] === "reclaim") { reclaimCalls++; if (result.exitCode === 0) cleanupConfirmed = true; }
|
||||
if (failure === "acknowledgement" && !lost && input.args[2] === "stage") {
|
||||
lost = true;
|
||||
return { ...result, exitCode: null, timedOut: true, stdout: "", stderr: "" };
|
||||
}
|
||||
return result;
|
||||
};
|
||||
if (failure === "journal") {
|
||||
await db.execute(sql.raw(`CREATE FUNCTION ${trigger}() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN
|
||||
RAISE EXCEPTION 'test staged cache journal write failed'; END $$`));
|
||||
await db.execute(sql.raw(`CREATE TRIGGER ${trigger} BEFORE UPDATE ON workspace_operations FOR EACH ROW
|
||||
WHEN (NEW.company_id = '${f.companyId}'::uuid AND NEW.phase = 'native_provider_cache_reclaim_prepare')
|
||||
EXECUTE FUNCTION ${trigger}()`));
|
||||
}
|
||||
try {
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(reclaimCalls).toBe(0);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).not.toBe("recovered");
|
||||
const [prepared] = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, f.companyId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim_prepare")));
|
||||
expect(prepared.metadata?.staged).toBeUndefined();
|
||||
const staged = path.join(f.pack, `node_modules.paperclip-reclaim-${prepared.id}`);
|
||||
const preserved = await lstat(staged).then(() => staged, () => path.join(f.pack, "node_modules"));
|
||||
expect((await lstat(path.join(preserved, "remaining-tool"))).isFile()).toBe(true);
|
||||
} finally {
|
||||
if (failure === "journal") {
|
||||
await db.execute(sql.raw(`DROP TRIGGER IF EXISTS ${trigger} ON workspace_operations`));
|
||||
await db.execute(sql.raw(`DROP FUNCTION ${trigger}()`));
|
||||
}
|
||||
}
|
||||
await makeRetryDue(f.actionId);
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(reclaimCalls).toBe(1);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).toBe("recovered");
|
||||
const prepared = await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, f.companyId),
|
||||
eq(workspaceOperations.phase, "native_provider_cache_reclaim_prepare")));
|
||||
expect(prepared).toHaveLength(1);
|
||||
expect(await readFile(path.join(f.sessions, "pending-receipt.json"), "utf8")).toBe("uncertain action history\n");
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
}, 30_000);
|
||||
|
||||
it.each(["pending", "foreign", "lease", "active", "recoverable", "manifest", "symlink"])("keeps files intact when %s cache authority is unproven", async failure => {
|
||||
const f = await pressuredToolCache(), before = await run(f.runId);
|
||||
if (failure === "pending") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...before.runnerProfileJson,
|
||||
nativeProviderPackAdmissionState: { schema: "paperclip.native-provider-pack-admission-state.v1",
|
||||
state: "pending", admissionEpoch: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (failure === "foreign") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...before.runnerProfileJson,
|
||||
nativeProviderPackAdmission: { ...before.runnerProfileJson!.nativeProviderPackAdmission as Record<string, unknown>,
|
||||
companyId: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (failure === "lease") await db.update(heartbeatRuns).set({ runnerProfileJson: { ...before.runnerProfileJson,
|
||||
remoteWorkspace: { ...before.runnerProfileJson!.remoteWorkspace as Record<string, unknown>, leaseId: randomUUID() } } })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
if (failure === "active") await db.update(nativeRunFinalizations).set({ leaseOwner: "another-controller" })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (failure === "recoverable") await db.update(nativeRunFinalizations).set({ failureDetail: { recoverable: true } })
|
||||
.where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
if (failure === "manifest") await writeFile(path.join(f.pack, "provider-pack.json"), JSON.stringify({
|
||||
schema: "paperclip-runner/remote-provider-pack/v1", digest: `sha256:${"f".repeat(64)}`, payload: {},
|
||||
}));
|
||||
if (failure === "symlink") {
|
||||
const outside = path.join(path.dirname(f.remote), "retained-installation");
|
||||
await mkdir(outside); await writeFile(path.join(outside, "installed-tool"), "rebuildable dependency\n");
|
||||
await rm(path.join(f.pack, "node_modules"), { recursive: true });
|
||||
await symlink(outside, path.join(f.pack, "node_modules"));
|
||||
}
|
||||
await recoverRetainedWorkspace(f.input);
|
||||
expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).not.toBe("recovered");
|
||||
expect(f.transfers()).toBe(["lease", "active"].includes(failure) ? 0 : 1);
|
||||
expect(await readFile(path.join(f.pack, "node_modules", "installed-tool"), "utf8")).toBe("rebuildable dependency\n");
|
||||
expect(await readFile(path.join(f.sessions, "pending-receipt.json"), "utf8")).toBe("uncertain action history\n");
|
||||
expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n");
|
||||
}, 30_000);
|
||||
|
||||
async function interruptedRemotePreparation(interruption: "shutdown" | "lease" = "shutdown", sourceLayout: RecoverySourceLayout = "retained") {
|
||||
const f = await acceptedRemoteExport(sourceLayout);
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
|
||||
@@ -9,6 +9,8 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
|
||||
import { beginIdleTrackedWork } from "./task-admission.js";
|
||||
import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js";
|
||||
import { resumePreservedResultFinalization } from "./preserved-result-finalization.js";
|
||||
import { reclaimInactiveNativeProviderCache } from "./native-runtime/retained-provider-cache.js";
|
||||
import { REMOTE_WORKSPACE_SANDBOX_DISK_FULL } from "./remote-workspace-transport.js";
|
||||
import { and, asc, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm";
|
||||
import { agentWakeupRequests, environmentLeases, executionWorkspaces, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, workspaceOperations, type Db } from "@paperclipai/db";
|
||||
import { remoteWorkspaceRunReferenceSchema, type Environment, type EnvironmentLease } from "@paperclipai/shared";
|
||||
@@ -295,8 +297,22 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ
|
||||
eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, source.remoteWorkspace.id),
|
||||
));
|
||||
if (!workspace) throw workspaceSourceBindingFailure("workspace_reference_mismatch");
|
||||
const saved = await saveRemoteWorkspace({ db, workspace, runId: input.runId,
|
||||
const save = () => saveRemoteWorkspace({ db, workspace, runId: input.runId,
|
||||
lease: cleanupLease as EnvironmentLease, target, assertOwned: ownership.assertHeld });
|
||||
let saved: Awaited<ReturnType<typeof save>>;
|
||||
try { saved = await save(); }
|
||||
catch (error) {
|
||||
if (!(error instanceof Error) || error.message !== REMOTE_WORKSPACE_SANDBOX_DISK_FULL) throw error;
|
||||
let reclaimed = false;
|
||||
try {
|
||||
reclaimed = await reclaimInactiveNativeProviderCache({ db, companyId: input.companyId,
|
||||
issueId: input.issueId, runId: input.runId, workspaceId: workspace.id, target,
|
||||
assertOwned: ownership.assertHeld });
|
||||
} catch { logger.warn({ runId: input.runId }, "Retained provider cache cleanup could not be confirmed"); }
|
||||
if (!reclaimed) throw error;
|
||||
await ownership.assertHeld();
|
||||
saved = await save();
|
||||
}
|
||||
await ownership.assertHeld();
|
||||
const [existing] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
|
||||
eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.heartbeatRunId, input.runId),
|
||||
|
||||
Reference in new issue
Block a user