mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 02:26:12 +02:00
fix(runtime): bind export cleanup to its recorded provider plugin
Refuse a duplicate driver key or unavailable original plugin during stop-only recovery; preserve the pending intent and fence later provider rebinding. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
f1f956714a
commit
a3fc644557
6 files changed
+42
-9
No files matched your search
@@ -116,6 +116,9 @@ delay that sweep until the claim expires. Neither an unconfirmed stop nor a
|
||||
stale receipt grants admission, and a changed lease or competing sandbox owner
|
||||
prevents cleanup from taking ownership. Retry export after the lease has a new
|
||||
confirmed stopped receipt. No provider turn is created by this recovery.
|
||||
Cleanup and its readiness probe use the plugin ID recorded on that lease. Another
|
||||
plugin with the same provider name cannot take over; an unavailable original
|
||||
plugin defers cleanup without consuming an attempt.
|
||||
|
||||
The opt-in `native-workspace-export-resume.live.test.ts` is a provider-boundary
|
||||
fault integration, separate from the browser Product E2E. After building the
|
||||
|
||||
@@ -5037,13 +5037,21 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
|
||||
expect(await db.select().from(environmentLeases)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each(["stopped", "unconfirmed", "foreign_marker", "competing_owner", "live_request", "late_receipt"])("recovers a pending export resume with stop-only cleanup after restart: %s", async outcome => {
|
||||
it.each(["stopped", "unconfirmed", "foreign_marker", "competing_owner", "live_request", "late_receipt", "duplicate_key", "owner_stopped", "owner_driver_changed", "owner_missing"])("recovers a pending export resume with stop-only cleanup after restart: %s", async outcome => {
|
||||
const seeded = await seedReusablePluginSandboxLease("paperclip_runner");
|
||||
const lease = seeded.reusableLease;
|
||||
const requestId = randomUUID();
|
||||
if (["duplicate_key", "owner_stopped", "owner_driver_changed", "owner_missing"].includes(outcome)) {
|
||||
const [owner] = await db.select().from(plugins).where(eq(plugins.id, seeded.pluginId));
|
||||
await db.insert(plugins).values({ ...owner, id: randomUUID(), pluginKey: "other.same-driver", packageName: "@other/same-driver",
|
||||
installOrder: -1, manifestJson: { ...owner.manifestJson, id: "other.same-driver" } });
|
||||
if (outcome === "owner_driver_changed") await db.update(plugins).set({ manifestJson: { ...owner.manifestJson, environmentDrivers: [] } }).where(eq(plugins.id, seeded.pluginId));
|
||||
if (outcome === "owner_missing") await db.delete(plugins).where(eq(plugins.id, seeded.pluginId));
|
||||
}
|
||||
await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", metadata: {
|
||||
...lease.metadata,
|
||||
nativeWorkspaceExportResume: { schema: "paperclip.workspace-export-resume.v1", requestId, companyId: seeded.companyId,
|
||||
pluginId: seeded.pluginId,
|
||||
runId: outcome === "foreign_marker" ? randomUUID() : lease.heartbeatRunId, leaseId: lease.id,
|
||||
provider: lease.provider, providerLeaseId: lease.providerLeaseId, resultId: randomUUID() },
|
||||
pendingCleanupAttemptId: requestId, pendingCleanupInFlight: true,
|
||||
@@ -5057,14 +5065,14 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
|
||||
cleanupStatus: null, metadata: { newOwner: true } }).where(eq(environmentLeases.id, lease.id));
|
||||
return outcome === "unconfirmed" ? undefined : { providerLeaseId: lease.providerLeaseId, state: "stopped" };
|
||||
});
|
||||
const workerManager = { isRunning: () => true, call,
|
||||
const workerManager = { isRunning: (id: string) => !(outcome === "owner_stopped" && id === seeded.pluginId), call,
|
||||
getWorker: () => ({ supportedMethods: ["environmentReleaseLease", "environmentDestroyLease"] }),
|
||||
} as unknown as PluginWorkerManager;
|
||||
const restarted = environmentRuntimeService(db, { pluginWorkerManager: workerManager });
|
||||
await heartbeatService(db, { environmentRuntime: restarted }).sweepPendingCleanupLeases({ backoffMs: 0 });
|
||||
expect(call.mock.calls.some((entry) => entry[1] === "environmentDestroyLease")).toBe(false);
|
||||
const persisted = (await environmentService(db).getLeaseById(lease.id))!;
|
||||
if (outcome === "stopped") {
|
||||
if (["stopped", "duplicate_key"].includes(outcome)) {
|
||||
expect(call).toHaveBeenCalledWith(seeded.pluginId, "environmentReleaseLease", expect.objectContaining({ providerLeaseId: lease.providerLeaseId, cancelActiveWork: true }), expect.any(Number));
|
||||
expect(persisted).toMatchObject({ status: "expired", cleanupStatus: "success", metadata: { remoteExecutionTermination: { state: "stopped", runId: lease.heartbeatRunId } } });
|
||||
expect(persisted.metadata?.nativeWorkspaceExportResume).toBeUndefined();
|
||||
@@ -5073,7 +5081,10 @@ describeEmbeddedPostgres("environmentRuntimeService", () => {
|
||||
} else {
|
||||
expect(persisted).toMatchObject({ status: "pending_cleanup", cleanupStatus: "failed" });
|
||||
expect(persisted.metadata?.remoteExecutionTermination).toBeUndefined();
|
||||
if (["foreign_marker", "competing_owner", "live_request"].includes(outcome)) expect(call).not.toHaveBeenCalled();
|
||||
if (["foreign_marker", "competing_owner", "live_request", "owner_stopped", "owner_driver_changed", "owner_missing"].includes(outcome)) expect(call).not.toHaveBeenCalled();
|
||||
if (["owner_stopped", "owner_driver_changed", "owner_missing"].includes(outcome)) {
|
||||
expect(persisted.metadata?.pendingCleanupAttemptId).toBe(requestId);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@@ -1671,7 +1671,14 @@ function createSandboxEnvironmentDriver(
|
||||
return runParent !== undefined ? runWithRuntimeParent(runParent, call) : call();
|
||||
}
|
||||
|
||||
async function resolveSandboxProviderPlugin(input: { provider: string }) {
|
||||
async function resolveSandboxProviderPlugin(input: { provider: string; pluginId?: string }) {
|
||||
if (input.pluginId) {
|
||||
const pinned = await resolvePluginSandboxProviderDriverById({ db, pluginId: input.pluginId, driverKey: input.provider });
|
||||
if (!pinned) return { state: "missing" as const, resolved: null };
|
||||
if (pinned.plugin.status !== "ready") return { state: "not_ready" as const, resolved: pinned };
|
||||
if (!pluginWorkerManager?.isRunning(pinned.plugin.id)) return { state: "worker_unavailable" as const, resolved: pinned };
|
||||
return { state: "running" as const, resolved: pinned };
|
||||
}
|
||||
const running = await resolvePluginSandboxProviderDriverByKey({
|
||||
db,
|
||||
driverKey: input.provider,
|
||||
@@ -2614,7 +2621,9 @@ function createSandboxEnvironmentDriver(
|
||||
`Sandbox provider "${recordedProvider}" needs a plugin worker manager for cleanup, but none is available.`,
|
||||
);
|
||||
}
|
||||
const pluginProvider = await resolveSandboxProviderPlugin({ provider: recordedProvider });
|
||||
const pinnedPluginId = exportResume ? readString(input.lease.metadata?.pluginId) : null;
|
||||
if (exportResume && !pinnedPluginId) throw new Error("Workspace export cleanup has no recorded provider plugin.");
|
||||
const pluginProvider = await resolveSandboxProviderPlugin({ provider: recordedProvider, ...(pinnedPluginId ? { pluginId: pinnedPluginId } : {}) });
|
||||
if (pluginProvider.state !== "running") {
|
||||
throw new Error(
|
||||
`Sandbox provider plugin for "${recordedProvider}" is ${pluginProvider.state}, so the cleanup teardown cannot run yet.`,
|
||||
@@ -2711,6 +2720,12 @@ function createSandboxEnvironmentDriver(
|
||||
// teardown runs, throws its own "no worker manager" error, and counts
|
||||
// toward the cap.
|
||||
if (!pluginWorkerManager) return true;
|
||||
if (hasNativeWorkspaceExportResume(input.lease)) {
|
||||
const pinnedPluginId = readString(input.lease.metadata?.pluginId);
|
||||
if (!pinnedPluginId) return true; // Permanent invalid intent, never a by-key fallback.
|
||||
const pinned = await resolvePluginSandboxProviderDriverById({ db, pluginId: pinnedPluginId, driverKey: recordedProvider });
|
||||
return Boolean(pinned?.plugin.status === "ready" && pluginWorkerManager.isRunning(pinned.plugin.id));
|
||||
}
|
||||
// Resolve the installed plugin without a wait. A plugin reload or a plugin
|
||||
// reinstall can remove the plugin row for a short window, so a missing
|
||||
// plugin is a transient condition, not a permanent one. Report not ready,
|
||||
|
||||
@@ -24,6 +24,7 @@ export function readNativeWorkspaceExportResume(lease: Lease) {
|
||||
|| marker.companyId !== lease.companyId || marker.runId !== lease.heartbeatRunId || !lease.heartbeatRunId
|
||||
|| marker.leaseId !== lease.id || marker.provider !== lease.provider || !lease.provider
|
||||
|| marker.providerLeaseId !== lease.providerLeaseId || !lease.providerLeaseId
|
||||
|| typeof marker.pluginId !== "string" || !marker.pluginId || marker.pluginId !== lease.metadata?.pluginId
|
||||
|| typeof marker.requestId !== "string" || !marker.requestId
|
||||
|| typeof marker.resultId !== "string" || !marker.resultId) return null;
|
||||
return marker;
|
||||
@@ -51,6 +52,7 @@ export async function settleNativeWorkspaceExportResume(db: Db, lease: Lease, op
|
||||
eq(environmentLeases.providerLeaseId, lease.providerLeaseId!), eq(environmentLeases.status, "pending_cleanup"),
|
||||
sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${options.attemptId}`,
|
||||
sql`${environmentLeases.metadata}->'nativeWorkspaceExportResume'->>'requestId' = ${marker.requestId}`,
|
||||
sql`${environmentLeases.metadata}->>'pluginId' = ${marker.pluginId}`,
|
||||
)).returning();
|
||||
return row ?? null;
|
||||
}
|
||||
@@ -33,7 +33,7 @@ describe("board retry of accepted workspace export", () => {
|
||||
await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId, serverFingerprint: resultId, schemaStatus: "accepted", resultJson: { work: "already finished" }, canonicalSha256: resultId });
|
||||
await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "terminal_failure", resultId, failureCode: "native_workspace_sync_out_unsafe_archive", failureDetail: { workspaceFinalizeAttempt: 1 } });
|
||||
const identity = { id: leaseId, companyId, heartbeatRunId: runId, provider: "daytona", providerLeaseId };
|
||||
await db.insert(environmentLeases).values({ ...identity, environmentId, issueId, status: "released", releasedAt: new Date(), cleanupStatus: "success", leasePolicy: "reuse_by_environment", metadata: { remoteExecutionTermination: remoteTerminationReceipt(identity, { providerLeaseId, state: "stopped" }) } });
|
||||
await db.insert(environmentLeases).values({ ...identity, environmentId, issueId, status: "released", releasedAt: new Date(), cleanupStatus: "success", leasePolicy: "reuse_by_environment", metadata: { pluginId: randomUUID(), remoteExecutionTermination: remoteTerminationReceipt(identity, { providerLeaseId, state: "stopped" }) } });
|
||||
await db.insert(issueRecoveryActions).values({ id: actionId, companyId, sourceIssueId: issueId, kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: agentId, cause: "native_workspace_sync_out_unsafe_archive", fingerprint: runId, evidence: { runId }, nextAction: "Repair saved files" });
|
||||
const request = { db, companyId, issueId, actionId, runId, actorId: "board", repairNote: "Removed only the known unsafe fixture link and preserved the saved work.", environmentRuntime: {
|
||||
resumeRunLease: vi.fn().mockResolvedValue({ providerLeaseId, metadata: { remoteCwd: "/work" } }),
|
||||
|
||||
@@ -38,7 +38,8 @@ export async function retryNativeWorkspaceExport(input: {
|
||||
if (!reference || reference.state !== "prepared" || reference.resourceDisposition === "destroy") throw changed();
|
||||
const lease = await one(db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.id, reference.leaseId))).limit(1));
|
||||
if (!lease || lease.heartbeatRunId !== run.id || lease.issueId !== issue.id
|
||||
|| lease.providerLeaseId !== reference.providerLeaseId || !lease.environmentId) throw changed();
|
||||
|| lease.providerLeaseId !== reference.providerLeaseId || !lease.environmentId
|
||||
|| typeof lease.metadata?.pluginId !== "string" || !lease.metadata.pluginId) throw changed();
|
||||
const [otherLease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
||||
ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider!), eq(environmentLeases.providerLeaseId, lease.providerLeaseId!), inArray(environmentLeases.status, ["active", "pending_cleanup"]),
|
||||
)).limit(1);
|
||||
@@ -95,7 +96,8 @@ export async function retryNativeWorkspaceExport(input: {
|
||||
releasedAt: now, failureReason: "workspace_export_resume_pending", updatedAt: now,
|
||||
metadata: { ...current.lease.metadata, remoteExecutionTermination: undefined,
|
||||
[NATIVE_WORKSPACE_EXPORT_RESUME_KEY]: { schema: "paperclip.workspace-export-resume.v1", requestId,
|
||||
companyId: input.companyId, runId: input.runId, leaseId: current.lease.id,
|
||||
companyId: input.companyId, runId: input.runId, leaseId: current.lease.id,
|
||||
pluginId: current.lease.metadata?.pluginId,
|
||||
provider: current.lease.provider, providerLeaseId: current.lease.providerLeaseId, resultId: current.result.id },
|
||||
pendingCleanupAttemptId: requestId, pendingCleanupInFlight: true,
|
||||
pendingCleanupLeaseExpiresAtMs: now.getTime() + 15 * 60_000 },
|
||||
|
||||
Reference in new issue
Block a user