diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index bb69c0b91f7..6819091b8ac 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -27,7 +27,7 @@ import { createNestedWorkspaceImportVerifier, onlineOrdinarySourceBlockers, offl import { inspectOnlineWorkspaceSource, requiresOnlineWorkspaceSource, type OnlineWorkspaceSource } from "./online-workspace-source.js"; import { withWorkspaceSourceLocks, withIndependentWorkspaceSourceLocks, IsolatedRepositorySourceBusy } from "./workspace-source-lock.js"; -import { inspectOnlineLegacyWorkspaceRecoverySource, inspectOnlineLegacyWorkspaceRecoveryHold, lockOnlineLegacyWorkspaceRecoverySources, type OnlineLegacyWorkspaceRecoveryHold } from "./online-legacy-workspace-recovery.js"; +import { inspectOnlineLegacyWorkspaceRecoverySource, inspectOnlineLegacyWorkspaceRecoveryHold, lockOnlineLegacyWorkspaceRecoverySources, ONLINE_LEGACY_WORKSPACE_PRESERVATION, type OnlineLegacyWorkspaceRecoveryHold } from "./online-legacy-workspace-recovery.js"; import { inspectNativeWorkspaceMigration, nativeWorkspaceMigrationMarker } from "./native-runtime/native-workspace-migration.js"; const SCHEMA = "paperclip.remote-workspace-migration.v1"; @@ -341,6 +341,17 @@ async function resolveMigrationSource(input: MigrationInput) { return onlineSource; } +function recoverySourcesResumeIdentity(value: unknown): unknown { + if (!Array.isArray(value)) return value; + return value.map(source => { + if (!source || typeof source !== "object" || source.onlineProof == null) return source; + const proof = object(source.onlineProof); + if (proof.schema !== ONLINE_LEGACY_WORKSPACE_PRESERVATION || typeof proof.fingerprint !== "string" + || !/^[a-f0-9]{64}$/.test(proof.fingerprint)) throw new Error("remote_workspace_migration_recovery_proof_invalid"); + return { ...source, onlineProof: { ...proof, fingerprint: null } }; + }); +} + async function reserveMigration(input: MigrationInput, onlineSource?: OnlineWorkspaceSource, cohort?: WorkspaceMigrationCohort, expectedSourceFingerprint?: string) { return input.db.transaction(async tx => { @@ -374,7 +385,8 @@ async function reserveMigration(input: MigrationInput, onlineSource?: OnlineWork if (Boolean(operation.metadata?.retainedDirectory) !== Boolean(current.retainedDirectory)) throw new Error("remote_workspace_migration_source_changed"); if (canonical(operation.metadata?.archivedGitSource ?? null) !== canonical(gitSourceProof(current.archivedGitSource))) throw new Error("remote_workspace_migration_source_changed"); if (canonical(operation.metadata?.finalizedRecoveryHolds ?? []) !== canonical(current.finalizedRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed"); - if (canonical(operation.metadata?.legacyRecoverySources ?? []) !== canonical(current.legacyRecoverySources)) throw new Error("remote_workspace_migration_source_changed"); + if (canonical(recoverySourcesResumeIdentity(operation.metadata?.legacyRecoverySources ?? [])) + !== canonical(recoverySourcesResumeIdentity(current.legacyRecoverySources))) throw new Error("remote_workspace_migration_source_changed"); if (canonical(operation.metadata?.onlineRecoveryHolds ?? []) !== canonical(current.onlineRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed"); return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds }; } @@ -434,6 +446,15 @@ function createMigrationChildVerifier(input: MigrationInput, onlineSource: Onlin async function captureMigration(input: MigrationInput, reserved: ReservedMigration, onlineSource?: OnlineWorkspaceSource, childVerifier?: ReturnType) { let operation = reserved.operation!; + if (canonical(operation.metadata?.legacyRecoverySources ?? []) !== canonical(reserved.legacyRecoverySources)) { + const [refreshed] = await input.db.update(workspaceOperations).set({ metadata: { ...operation.metadata, + legacyRecoverySources: reserved.legacyRecoverySources }, updatedAt: new Date() }).where(and( + eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.executionWorkspaceId, input.workspaceId), + eq(workspaceOperations.id, operation.id), eq(workspaceOperations.phase, "remote_workspace_migration"), eq(workspaceOperations.status, "running"), + sql`${workspaceOperations.finishedAt} is null`, eq(workspaceOperations.metadata, operation.metadata!))).returning(); + if (!refreshed) throw new Error("remote_workspace_migration_operation_changed"); + operation = refreshed; + } const snapshot = await beginWorkspaceSnapshot(input.db, { companyId: input.companyId, workspaceId: input.workspaceId, operationId: operation.id, generation: 1, epoch: 1 }); const scratch = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-migration-"))); diff --git a/server/src/services/remote-workspace-online-source.test.ts b/server/src/services/remote-workspace-online-source.test.ts index 85527595038..c8af7667b7f 100644 --- a/server/src/services/remote-workspace-online-source.test.ts +++ b/server/src/services/remote-workspace-online-source.test.ts @@ -13,9 +13,10 @@ import { resolveManagedProjectWorkspaceDir, resolvePaperclipInstanceRoot } from import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; import { createStorageService } from "../storage/service.js"; import type { StorageService } from "../storage/types.js"; -import { inventoryRemoteWorkspaceMigration, migrateRemoteWorkspace } from "./remote-workspace-migration.js"; +import { inventoryRemoteWorkspaceMigration, migrateRemoteWorkspace, migrateRemoteWorkspaceCohort } from "./remote-workspace-migration.js"; import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "./remote-workspace-snapshot-storage.js"; import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; +import * as workspaceArchive from "./remote-workspace-snapshot-archive.js"; import { materializeManagedProjectWorkspace } from "./managed-repository-checkout.js"; import { withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js"; import { IsolatedRepositorySourceBusy, withIsolatedRepositorySourceLock, withWorkspaceSourceLocks } from "./workspace-source-lock.js"; @@ -528,6 +529,116 @@ describe.skipIf(!support.supported)("online ordinary physical source fence", () status: "active", evidence: { runId } }); return { ...f, run, owner, lease, warning }; } + async function interruptedRecoveryFixture(cohort = false) { + const f = await recoveryFixture(); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.warning.id)); + const reference = f.run.runnerProfileJson!.nativeWorkspaceSync as Record; + const extra = Array.from({ length: 7 }, () => ({ runId: randomUUID(), leaseId: randomUUID(), providerLeaseId: randomUUID() })); + await db.insert(heartbeatRuns).values(extra.map(row => ({ id: row.runId, companyId: f.companyId, agentId: f.run.agentId, + nativeIssueId: f.run.nativeIssueId, status: "failed", runtimeMode: "native", finishedAt: new Date(), + runnerProfileJson: { nativeWorkspaceSync: { ...reference, leaseId: row.leaseId, providerLeaseId: row.providerLeaseId } } }))); + await db.insert(nativeRunFinalizations).values(extra.map(row => ({ companyId: f.companyId, + issueId: f.run.nativeIssueId!, runId: row.runId, phase: "terminal_failure" }))); + await db.insert(environmentLeases).values(extra.map(row => ({ id: row.leaseId, companyId: f.companyId, + environmentId: f.lease.environmentId, executionWorkspaceId: f.workspaceId, issueId: f.run.nativeIssueId, + heartbeatRunId: row.runId, provider: "daytona", providerLeaseId: row.providerLeaseId, + status: "retained", leasePolicy: "retain_on_failure", metadata: { remoteCwd: reference.remoteCwd } }))); + const apply = () => cohort ? migrateRemoteWorkspaceCohort({ db, storage, companyId: f.companyId, + workspaceIds: [f.workspaceId], apply: true, signal: AbortSignal.timeout(30_000) }) + : migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: f.workspaceId, apply: true }); + const restore = workspaceArchive.restoreRemoteWorkspaceArchive; + const interrupt = vi.spyOn(workspaceArchive, "restoreRemoteWorkspaceArchive").mockImplementationOnce(async input => { + await restore(input); + expect(await db.select().from(executionWorkspaceSnapshots)).toEqual([expect.objectContaining({ status: "uploaded" })]); + throw new Error("interrupted after verified upload"); + }); + try { await expect(apply()).rejects.toThrow("interrupted after verified upload"); } + finally { interrupt.mockRestore(); } + const [operation] = await db.select().from(workspaceOperations), [snapshot] = await db.select().from(executionWorkspaceSnapshots); + const sources = operation.metadata!.legacyRecoverySources as Array<{ runId: string; issueId: string; + reference: Record; onlineProof: { schema: string; fingerprint: string } }>; + expect(sources).toHaveLength(8); + expect(operation.status).toBe("running"); expect(snapshot.status).toBe("uploaded"); + return { ...f, apply, operation, snapshot, sources, changed: sources[6] }; + } + it.each([false, true])("resumes eight saved recovery sources after accounting changes (cohort=%s)", async cohort => { + const f = await interruptedRecoveryFixture(cohort); + await db.update(heartbeatRuns).set({ accountingAttemptCount: 1, accountingLastAttemptAt: new Date(), + accountingLastError: "retry scheduled", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.changed.runId)); + await expect(f.apply()).resolves.toMatchObject({ applied: true }); + const operations = await db.select().from(workspaceOperations), snapshots = await db.select().from(executionWorkspaceSnapshots); + expect(operations).toEqual([expect.objectContaining({ id: f.operation.id, status: "succeeded" })]); + expect(snapshots).toEqual([expect.objectContaining({ id: f.snapshot.id, sourceOperationId: f.operation.id, status: "committed", + objectKey: f.snapshot.objectKey, sha256: f.snapshot.sha256, manifestDigest: f.snapshot.manifestDigest, byteSize: f.snapshot.byteSize })]); + expect(operations[0].metadata!.nestedWorkspaceScope).toEqual(f.operation.metadata!.nestedWorkspaceScope); + expect(operations[0].metadata!.cohortMigration).toEqual(f.operation.metadata!.cohortMigration); + const sources = operations[0].metadata!.legacyRecoverySources as typeof f.sources; + expect(sources.map(source => ({ ...source, onlineProof: { ...source.onlineProof, fingerprint: null } }))) + .toEqual(f.sources.map(source => ({ ...source, onlineProof: { ...source.onlineProof, fingerprint: null } }))); + expect(sources[6].onlineProof.fingerprint).not.toBe(f.changed.onlineProof.fingerprint); + expect(sources.filter((_, index) => index !== 6)).toEqual(f.sources.filter((_, index) => index !== 6)); + expect(readRemoteWorkspaceAuthority((await stored(f.workspaceId)).metadata)?.snapshotId).toBe(f.snapshot.id); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("saved legacy edits\n"); + await expect(f.apply()).resolves.toMatchObject({ applied: true, replayed: true }); + expect(await db.select().from(workspaceOperations)).toHaveLength(1); + expect(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(1); + }); + it.each(["reference", "allocation", "owner", "proof-version"] as const)("refuses a changed saved recovery %s on resume", async change => { + const f = await interruptedRecoveryFixture(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.changed.runId)); + if (change === "reference" || change === "allocation") { + const reference: Record = { ...f.changed.reference, ...(change === "reference" ? { baselineSha256: "56".repeat(32) } : { providerLeaseId: randomUUID() }) }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, nativeWorkspaceSync: reference } }) + .where(eq(heartbeatRuns.id, run.id)); + if (change === "allocation") await db.update(environmentLeases).set({ providerLeaseId: reference.providerLeaseId as string }) + .where(eq(environmentLeases.id, reference.leaseId as string)); + expect(await inspectOnlineLegacyWorkspaceRecoverySource(db, { workspace: await stored(f.workspaceId), runId: run.id })).not.toBeNull(); + } + if (change === "owner") await db.update(nativeRunFinalizations).set({ leaseOwner: "competing-finalizer", + leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, run.id)); + if (change === "proof-version") { + const sources = f.sources.map((source, index) => index === 6 ? { ...source, onlineProof: { ...source.onlineProof, schema: "unsupported.v2" } } : source); + await db.update(workspaceOperations).set({ metadata: { ...f.operation.metadata, legacyRecoverySources: sources } }) + .where(eq(workspaceOperations.id, f.operation.id)); + } + const [before] = await db.select().from(workspaceOperations); + await expect(f.apply()).rejects.toThrow(change === "proof-version" ? "remote_workspace_migration_recovery_proof_invalid" + : change === "owner" ? "remote_copyback_unresolved" : "remote_workspace_migration_source_changed"); + expect(await db.select().from(workspaceOperations)).toEqual([before]); + expect(await db.select().from(executionWorkspaceSnapshots)).toEqual([f.snapshot]); + expect(readRemoteWorkspaceAuthority((await stored(f.workspaceId)).metadata)).toBeNull(); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("saved legacy edits\n"); + }); + it("still refuses recovery accounting changes within one admitted attempt", async () => { + const f = await recoveryFixture(), put = storage.putFile.bind(storage); + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.warning.id)); + vi.spyOn(storage, "putFile").mockImplementationOnce(async input => { + const result = await put(input); + await db.update(heartbeatRuns).set({ accountingAttemptCount: 1 }).where(eq(heartbeatRuns.id, f.run.id)); + return result; + }); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_workspace_migration_source_changed"); + expect(readRemoteWorkspaceAuthority((await stored(f.workspaceId)).metadata)).toBeNull(); + expect((await db.select().from(workspaceOperations))[0].status).toBe("running"); + }); + it("does not refresh a saved recovery proof before its finalization fence is acquired", async () => { + const f = await interruptedRecoveryFixture(), other = createDb(database.connectionString); + await db.update(heartbeatRuns).set({ accountingAttemptCount: 1 }).where(eq(heartbeatRuns.id, f.changed.runId)); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const pending = new Promise(resolve => { release = resolve; }); + const held = other.transaction(async tx => { + expect(await tryLockNativeWorkspaceFinalization(tx as unknown as Db, { companyId: f.companyId, runId: f.changed.runId })).toBe(true); + entered(); await pending; + }); + await ready; + try { + await expect(f.apply()).rejects.toThrow("remote_copyback_owned"); + expect(await db.select().from(workspaceOperations)).toEqual([f.operation]); + expect(await db.select().from(executionWorkspaceSnapshots)).toEqual([f.snapshot]); + expect(readRemoteWorkspaceAuthority((await stored(f.workspaceId)).metadata)).toBeNull(); + } finally { release(); await held; await other.$client.end({ timeout: 1 }); } + }); async function artifactFailureFixture(descendant = false, cause: "runner_remote_provider_artifact_incompatible" | "native_session_interrupted" = "runner_remote_provider_artifact_incompatible") { const f = await recoveryFixture(descendant), statusVersion = 2; const profile = { ...f.run.runnerProfileJson, sessionCheckpoint: cause === "native_session_interrupted"