From 53c9be8d4169af77beac382a43891788baf7eb57 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Mon, 9 Nov 2026 23:27:25 -0800 Subject: [PATCH] Reuse identical captured workspace archives (#20476) Reuse the current committed snapshot when a fresh capture matches its manifest and plaintext archive hash and size. Preserve the original encryption binding and verify each new run receipt for recovery, concurrent replay and storage request statistics. Verified with 338 focused save and recovery tests, server typecheck and build, all final-head CI checks, and Greptile 5/5. Co-Authored-By: Paperclip --- doc/execution-semantics.md | 20 + .../native-runtime/native-safe-replacement.ts | 13 +- ...remote-workspace-snapshot-reuse-receipt.ts | 50 ++ .../remote-workspace-snapshot-reuse.test.ts | 462 ++++++++++++++++++ .../remote-workspace-snapshot-storage.ts | 25 + ...remote-workspace-unchanged-release.test.ts | 14 +- .../remote-workspaces.integration.test.ts | 5 + server/src/services/remote-workspaces.ts | 89 +++- .../services/workspace-recovery-receipt.ts | 18 +- 9 files changed, 662 insertions(+), 34 deletions(-) create mode 100644 server/src/services/remote-workspace-snapshot-reuse-receipt.ts create mode 100644 server/src/services/remote-workspace-snapshot-reuse.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 5ce43c1ac39..304f312b917 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -616,6 +616,26 @@ keeps the full save. The next run's reservation, cancelled-workspace preservation and a later final save each apply this release. A repeated save returns the same version. +A save that requires a full capture also keeps the current saved version when +the captured manifest, archive hash and size match the saved version. This +comparison applies only to the same company and workspace. The saved encryption +key must be readable. The archive must still exist with its recorded size. The save then completes its +normal ownership checks, run receipt, repository metadata and waiter wake +without uploading another archive. Its unused generation has no object and is +marked deleted. A retry returns the completed receipt. An earlier upload attempt +keeps its recovery path; it cannot be discarded by this comparison. Changed +contents still create a new version and retain the previous version for seven +days after replacement. Active work and recovery references can extend retention. + +The reused archive keeps its original source run, generation and encryption +binding. Recovery and native replacement verify the new run's final save through +its succeeded operation and empty retired generation, including its epoch, +workspace, sandbox location, content hash and settled work. They do not borrow +the old run's authority. Read-only historical attachments keep this proof after +a later writer advances the workspace. Existing saves from earlier releases +keep their original receipt checks. Reuse metadata requests, including retries +and failures, count toward the save's storage request statistics. + A writer stopped mid-run, for example by reassignment, keeps its unsaved files in its stopped sandbox. When the run used a reusable (warm) sandbox, the stop leaves that lease `released` under its reusable policy with a stop-and-retain diff --git a/server/src/services/native-runtime/native-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts index cb0b6100da5..91d99925903 100644 --- a/server/src/services/native-runtime/native-safe-replacement.ts +++ b/server/src/services/native-runtime/native-safe-replacement.ts @@ -30,6 +30,7 @@ import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js" import { NATIVE_PROVIDER_DURABLE_EVENT_TYPES } from "./native-retired-backup-source.js"; import { completedWorkspaceRecoveryCondition } from "../workspace-recovery-receipt.js"; import { nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js"; +import { verifiedWorkspaceSnapshotReuse } from "../remote-workspace-snapshot-reuse-receipt.js"; export const NATIVE_SAFE_REPLACEMENT_REASON = "native_safe_replacement"; const record = (v: unknown): Record => @@ -169,7 +170,7 @@ async function preClaimFailureEvidence(db: ReplacementDb, run: typeof heartbeatR } /** A saved remote source is proved by its current ledger, never a controller path. */ -async function remoteWorkspaceReplacementEvidence( +export async function remoteWorkspaceReplacementEvidence( db: ReplacementDb, run: typeof heartbeatRuns.$inferSelect, issueId: string, @@ -215,11 +216,11 @@ async function remoteWorkspaceReplacementEvidence( const snapshotQuery = db.select().from(executionWorkspaceSnapshots).where(and( eq(executionWorkspaceSnapshots.companyId, run.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.id, reference.snapshotId!), eq(executionWorkspaceSnapshots.status, "committed"), - eq(executionWorkspaceSnapshots.epoch, reference.epoch), eq(executionWorkspaceSnapshots.sourceRunId, run.id), )); const [snapshot] = await (lock ? snapshotQuery.for("update") : snapshotQuery); if (!snapshot?.sha256 || !snapshot.objectKey || !snapshot.key || !snapshot.manifestDigest || !snapshot.byteSize) return denied; - const [operation] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + const [operation] = snapshot.sourceRunId === run.id && snapshot.epoch === reference.epoch + ? await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( eq(workspaceOperations.id, snapshot.sourceOperationId), eq(workspaceOperations.companyId, run.companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id), eq(workspaceOperations.heartbeatRunId, run.id), eq(workspaceOperations.issueId, issueId), eq(workspaceOperations.phase, "remote_workspace_snapshot"), @@ -228,8 +229,10 @@ async function remoteWorkspaceReplacementEvidence( sql`${workspaceOperations.metadata}->>'epoch' = ${String(reference.epoch)}`, sql`${workspaceOperations.metadata}->>'snapshotId' = ${snapshot.id}`, sql`${workspaceOperations.metadata}->>'sha256' = ${snapshot.sha256}`, - )); - return { workspacePreserved: Boolean(operation), providerStopped }; + )) : []; + const reused = operation ? null : await verifiedWorkspaceSnapshotReuse(db, { + companyId: run.companyId, issueId, runId: run.id, reference, snapshot }); + return { workspacePreserved: Boolean(operation || reused), providerStopped }; } export function nativeReplacementPending() { diff --git a/server/src/services/remote-workspace-snapshot-reuse-receipt.ts b/server/src/services/remote-workspace-snapshot-reuse-receipt.ts new file mode 100644 index 00000000000..bc01c4df918 --- /dev/null +++ b/server/src/services/remote-workspace-snapshot-reuse-receipt.ts @@ -0,0 +1,50 @@ +import { and, eq, isNull, sql } from "drizzle-orm"; +import { executionWorkspaceSnapshots, workspaceOperations, type Db } from "@paperclipai/db"; +import type { RemoteWorkspaceRunReference } from "@paperclipai/shared"; + +type Snapshot = typeof executionWorkspaceSnapshots.$inferSelect; + +export async function verifiedWorkspaceSnapshotReuse(db: Pick, input: { + companyId: string; issueId: string; runId: string; reference: RemoteWorkspaceRunReference; snapshot: Snapshot; +}): Promise { + const { reference, snapshot } = input; + if (reference.state !== "saved" || reference.snapshotId !== snapshot.id || snapshot.companyId !== input.companyId + || snapshot.executionWorkspaceId !== reference.workspaceId || !["committed", "superseded"].includes(snapshot.status) + || !snapshot.objectKey?.startsWith(`${input.companyId}/workspace-snapshots/${reference.workspaceId}/${snapshot.id}/`) + || !snapshot.key || !snapshot.sha256 || !snapshot.manifestDigest || !snapshot.byteSize) return null; + const rows = await db.select({ id: workspaceOperations.id, metadata: workspaceOperations.metadata, + generation: executionWorkspaceSnapshots.generation }).from(workspaceOperations) + .innerJoin(executionWorkspaceSnapshots, and( + eq(executionWorkspaceSnapshots.companyId, input.companyId), + eq(executionWorkspaceSnapshots.executionWorkspaceId, reference.workspaceId), + eq(executionWorkspaceSnapshots.sourceOperationId, workspaceOperations.id), + eq(executionWorkspaceSnapshots.sourceRunId, input.runId), + eq(executionWorkspaceSnapshots.epoch, reference.epoch), + eq(executionWorkspaceSnapshots.parentSnapshotId, snapshot.id), + eq(executionWorkspaceSnapshots.status, "deleted"), + isNull(executionWorkspaceSnapshots.objectKey), isNull(executionWorkspaceSnapshots.sha256), + isNull(executionWorkspaceSnapshots.byteSize), isNull(executionWorkspaceSnapshots.key), + isNull(executionWorkspaceSnapshots.manifestDigest), + sql`coalesce(${executionWorkspaceSnapshots.metadata}->'uploads', '[]'::jsonb) = '[]'::jsonb`)) + .where(and(eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.issueId, input.issueId), + eq(workspaceOperations.executionWorkspaceId, reference.workspaceId), eq(workspaceOperations.heartbeatRunId, input.runId), + eq(workspaceOperations.phase, "remote_workspace_snapshot"), eq(workspaceOperations.status, "succeeded"), + sql`${workspaceOperations.metadata}->>'boundary' = 'final'`, + sql`${workspaceOperations.metadata}->>'epoch' = ${String(reference.epoch)}`, + sql`${workspaceOperations.metadata}->>'parentSnapshotId' = ${snapshot.id}`, + sql`${workspaceOperations.metadata}->>'snapshotId' = ${snapshot.id}`, + sql`${workspaceOperations.metadata}->>'reusedSnapshotId' = ${snapshot.id}`, + sql`${workspaceOperations.metadata}->>'sha256' = ${snapshot.sha256}`, + sql`${workspaceOperations.metadata}#>>'{exportCleanup,leaseId}' = ${reference.leaseId}`, + sql`${workspaceOperations.metadata}#>>'{exportCleanup,providerLeaseId}' = ${reference.providerLeaseId}`, + sql`${workspaceOperations.metadata}#>>'{exportCleanup,remoteCwd}' = ${reference.remoteCwd}`)).limit(2); + if (rows.length !== 1) return null; + const [receipt] = rows, work = receipt.metadata?.saveWork; + if (receipt.generation <= snapshot.generation || receipt.metadata?.generation !== receipt.generation + || !work || typeof work !== "object" || Array.isArray(work)) return null; + const settled = work as Record; + if (settled.schema !== "paperclip.remote-workspace-save-work.v1" || settled.unknown !== false + || !Array.isArray(settled.activeIds) || settled.activeIds.length !== 0 + || typeof settled.quiescedAt !== "string" || !Number.isFinite(Date.parse(settled.quiescedAt))) return null; + return receipt.id; +} diff --git a/server/src/services/remote-workspace-snapshot-reuse.test.ts b/server/src/services/remote-workspace-snapshot-reuse.test.ts new file mode 100644 index 00000000000..de93b42ec5e --- /dev/null +++ b/server/src/services/remote-workspace-snapshot-reuse.test.ts @@ -0,0 +1,462 @@ +import { execFile } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; +import { cp, mkdir, mkdtemp, readFile, realpath, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { promisify } from "node:util"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { + activityLog, agents, closeRegisteredClients, companies, createDb, environmentLeases, environments, + executionWorkspaces, executionWorkspaceSnapshots, heartbeatRuns, issueRecoveryActions, issues, workspaceOperations, +} from "@paperclipai/db"; +import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import { remoteWorkspaceRunReferenceSchema, type EnvironmentLease } from "@paperclipai/shared"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { createStorageService } from "../storage/service.js"; +import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; +import { remoteWorkspaceSnapshotRetentionService } from "./remote-workspace-snapshot-retention.js"; +import { beginWorkspaceSnapshot } from "./remote-workspace-snapshot-storage.js"; +import { captureRemoteWorkspaceDirectory } from "./remote-workspace-snapshot-archive.js"; +import { immutableRemoteWorkspaceRecoveryAttachment, verifiedWorkspaceRepairReceipt } from "./workspace-recovery-receipt.js"; +import { remoteWorkspaceReplacementEvidence } from "./native-runtime/native-safe-replacement.js"; +import { remoteTerminationReceipt } from "./remote-execution-termination.js"; +import { + createSavedWorkspaceLoader, prepareRemoteWorkspace, readRemoteWorkspaceAuthority, reserveRemoteWorkspace, saveRemoteWorkspace, +} from "./remote-workspaces.js"; + +const exec = promisify(execFile); +const support = await getEmbeddedPostgresTestSupport(); +function deferred() { + let resolve!: () => void; + const promise = new Promise(ready => { resolve = ready; }); + return { promise, resolve }; +} + +describe.skipIf(!support.supported)("captured unchanged workspace snapshots", () => { + let database: Awaited>; + let db: ReturnType; + let root: string; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("workspace-snapshot-reuse-"); + db = createDb(database.connectionString); + root = await realpath(await mkdtemp(path.join(os.tmpdir(), "workspace-snapshot-reuse-"))); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "64".repeat(32)); + vi.stubEnv("DATABASE_URL", database.connectionString); + }, 60_000); + afterAll(async () => { + if (database) await closeRegisteredClients(database.connectionString); + await database?.cleanup(); + if (root) await rm(root, { recursive: true, force: true }); + vi.unstubAllEnvs(); + }); + + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Snapshot reuse", issuePrefix: `S${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Worker", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Save workspace", status: "in_progress", assigneeAgentId: agentId }); + const [environment] = await db.insert(environments).values({ name: issueId, driver: "sandbox" }).returning(); + const storage = { ...createStorageService(createLocalDiskStorageProvider(path.join(root, issueId, "objects"))), provider: "s3" as const }; + const puts = vi.spyOn(storage, "putFile"); + const remoteCwd = path.join(root, issueId, "remote"), providerLeaseId = randomUUID(); + await mkdir(remoteCwd, { recursive: true }); + let captures = 0; + const transfer: NonNullable["syncIn"]> = async operations => { + for (const operation of operations) for (const file of operation.files) { + await mkdir(path.dirname(file.targetPath), { recursive: true }); + await cp(file.sourcePath, file.targetPath, { recursive: file.kind === "directory" }); + } + return { operations: operations.map(operation => ({ operationId: operation.operationId, + filesTransferred: operation.files.length, bytesTransferred: 0 })) }; + }; + const target = (leaseId: string): AdapterSandboxExecutionTarget => ({ + kind: "remote", transport: "sandbox", leaseId, remoteCwd, timeoutMs: 30_000, + runner: { syncIn: transfer, syncOut: async operations => { captures++; return transfer(operations); }, execute: async input => { + const result = await exec(input.command, input.args ?? [], { cwd: input.cwd ?? remoteCwd, + env: { ...process.env, ...input.env }, timeout: input.timeoutMs, maxBuffer: 8 * 1024 * 1024 }); + return { exitCode: 0, stdout: result.stdout, stderr: result.stderr, signal: null, timedOut: false, + pid: process.pid, startedAt: new Date().toISOString() }; + } }, + }); + const writer = async (existingWorkspaceId: string | null) => { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, runtimeMode: "legacy", + status: "running", contextSnapshot: { issueId } }); + const [lease] = await db.insert(environmentLeases).values({ companyId, issueId, heartbeatRunId: runId, + environmentId: environment.id, provider: "daytona", providerLeaseId, status: "active" }).returning(); + const reserved = await reserveRemoteWorkspace({ db, storage, companyId, issueId, runId, agentId, + environmentId: environment.id, projectId: null, projectWorkspaceId: null, existingWorkspaceId, source: { kind: "empty" } }); + const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId, + lease: lease as EnvironmentLease, target: target(lease.id) }); + return { runId, lease: prepared.lease, workspaceId: reserved.id, remoteRoot: prepared.remoteRoot }; + }; + const a = await writer(null); + const workspace = async () => (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, a.workspaceId)))[0]!; + await writeFile(path.join(a.remoteRoot, "saved.txt"), "saved result\n"); + const savedA = await saveRemoteWorkspace({ db, storage, workspace: await workspace(), runId: a.runId, + lease: a.lease, target: target(a.lease.id), assertOwned: async () => undefined }); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, a.runId)); + const b = await writer(a.workspaceId); + puts.mockClear(); + const capturesBefore = captures; + const save = async (boundary?: string) => saveRemoteWorkspace({ db, storage, workspace: await workspace(), runId: b.runId, + lease: b.lease, target: target(b.lease.id), assertOwned: async () => undefined, boundary }); + const snapshots = () => db.select().from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, a.workspaceId))); + const operations = () => db.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, companyId), + eq(workspaceOperations.heartbeatRunId, b.runId), eq(workspaceOperations.phase, "remote_workspace_snapshot"))); + return { companyId, agentId, issueId, a, b, storage, puts, savedA, workspace, writer, save, snapshots, operations, target, + captures: () => captures - capturesBefore }; + } + + async function recoveredReuseFixture() { + const f = await fixture(), saved = await f.save(); + const run = async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.b.runId)))[0]!; + const current = await run(), reference = remoteWorkspaceRunReferenceSchema.parse(current.runnerProfileJson?.remoteWorkspace); + const [originalLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.b.lease.id)); + const [stoppedLease] = await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success", + metadata: { ...originalLease.metadata, remoteExecutionTermination: remoteTerminationReceipt(originalLease, + { providerLeaseId: originalLease.providerLeaseId!, state: "stopped" }) } }).where(eq(environmentLeases.id, originalLease.id)).returning(); + await db.update(issues).set({ executionWorkspaceId: f.a.workspaceId }).where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set({ status: "failed", runtimeMode: "native", nativeIssueId: f.issueId, finishedAt: new Date(), + runnerProfileJson: { ...current.runnerProfileJson, nativeExecutionInput: { workspace: { cwd: reference.remoteCwd } } } }) + .where(eq(heartbeatRuns.id, f.b.runId)); + const [action] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: "stranded_assigned_issue", cause: "fixture", fingerprint: randomUUID(), ownerType: "board", + nextAction: "Continue recovered work", evidence: { workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", + runId: f.b.runId, status: "recovered", stage: "saving_receipts", attempt: 1, requestIds: [] } } }).returning(); + const [receipt] = await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, + executionWorkspaceId: f.a.workspaceId, heartbeatRunId: f.b.runId, phase: "workspace_recovery", status: "succeeded", + metadata: { storage: "remote_workspace", recoveryActionId: action.id, leaseId: reference.leaseId, + providerLeaseId: reference.providerLeaseId, epoch: reference.epoch, snapshotId: saved.snapshotId, + sha256: saved.sha256, generation: saved.generation } }).returning(); + return { ...f, saved, run, stoppedLease, receipt }; + } + + it("accepts a verified reused archive for recovery and stopped native replacement", async () => { + const f = await recoveredReuseFixture(), run = await f.run(); + expect(await verifiedWorkspaceRepairReceipt(db, run)).toBe(f.receipt.id); + expect(await immutableRemoteWorkspaceRecoveryAttachment(db, run)).toMatchObject({ snapshotId: f.savedA.snapshotId, + operationId: f.receipt.id, generation: f.savedA.generation }); + expect(await remoteWorkspaceReplacementEvidence(db, run, f.issueId, [f.stoppedLease])) + .toEqual({ workspacePreserved: true, providerStopped: true }); + expect(await db.transaction(tx => remoteWorkspaceReplacementEvidence(tx, run, f.issueId, [f.stoppedLease], true))) + .toEqual({ workspacePreserved: true, providerStopped: true }); + expect((await f.snapshots()).find(s => s.id === f.savedA.snapshotId)).toMatchObject({ sourceRunId: f.a.runId, epoch: 1 }); + }); + + it("keeps an immutable recovery attachment after the next writer advances but refuses old execution authority", async () => { + const f = await recoveredReuseFixture(), run = await f.run(), next = await f.writer(f.a.workspaceId); + await writeFile(path.join(next.remoteRoot, "saved.txt"), "next writer contents\n"); + await saveRemoteWorkspace({ db, storage: f.storage, workspace: await f.workspace(), runId: next.runId, + lease: next.lease, target: f.target(next.lease.id), assertOwned: async () => undefined }); + expect((await f.snapshots()).find(s => s.id === f.savedA.snapshotId)!.status).toBe("superseded"); + expect(await immutableRemoteWorkspaceRecoveryAttachment(db, run)).toMatchObject({ snapshotId: f.savedA.snapshotId }); + expect(await verifiedWorkspaceRepairReceipt(db, run)).toBeNull(); + expect(await remoteWorkspaceReplacementEvidence(db, run, f.issueId, [f.stoppedLease])) + .toEqual({ workspacePreserved: false, providerStopped: true }); + }); + + it.each(["missing_marker", "pending_generation", "journaled_upload", "wrong_epoch", "wrong_lease", + "unfinished_work", "unknown_work", "wrong_generation", "wrong_boundary", "foreign_run"])( + "refuses an unverified reuse receipt for all recovery readers: %s", async fault => { + const f = await recoveredReuseFixture(), run = await f.run(); + const operation = (await f.operations())[0]!, metadata = structuredClone(operation.metadata!); + const pending = (await f.snapshots()).find(s => s.sourceRunId === f.b.runId)!; + if (fault === "missing_marker") delete metadata.reusedSnapshotId; + if (fault === "wrong_epoch") metadata.epoch = Number(metadata.epoch) + 1; + if (fault === "wrong_generation") metadata.generation = Number(metadata.generation) + 1; + if (fault === "wrong_boundary") metadata.boundary = "repositories:checkpoint"; + if (fault === "wrong_lease") metadata.exportCleanup = { ...(metadata.exportCleanup as Record), leaseId: randomUUID() }; + if (fault === "unfinished_work") metadata.saveWork = { ...(metadata.saveWork as Record), activeIds: [randomUUID()] }; + if (fault === "unknown_work") metadata.saveWork = { ...(metadata.saveWork as Record), unknown: true }; + await db.update(workspaceOperations).set({ metadata, ...(fault === "foreign_run" ? { heartbeatRunId: f.a.runId } : {}) }) + .where(eq(workspaceOperations.id, operation.id)); + if (fault === "pending_generation") await db.update(executionWorkspaceSnapshots).set({ status: "pending" }) + .where(eq(executionWorkspaceSnapshots.id, pending.id)); + if (fault === "journaled_upload") await db.update(executionWorkspaceSnapshots).set({ metadata: { uploads: [{ objectKey: "owned-uncertain-attempt" }] } }) + .where(eq(executionWorkspaceSnapshots.id, pending.id)); + expect(await verifiedWorkspaceRepairReceipt(db, run)).toBeNull(); + expect(await immutableRemoteWorkspaceRecoveryAttachment(db, run)).toBeNull(); + expect(await remoteWorkspaceReplacementEvidence(db, run, f.issueId, [f.stoppedLease])) + .toEqual({ workspacePreserved: false, providerStopped: true }); + }); + + it.each(["retry", "missing", "forbidden"])("counts real S3 metadata attempts in save statistics: %s", async outcome => { + const f = await fixture(), parent = (await f.snapshots()).find(s => s.id === f.savedA.snapshotId)!; + let requests = 0; + const server = createServer((request, response) => { + requests++; + if (request.method !== "HEAD") { response.writeHead(405); response.end(); return; } + const status = outcome === "missing" ? 404 : outcome === "forbidden" ? 403 : requests === 1 ? 503 : 200; + response.writeHead(status, status === 200 ? { "content-length": String(parent.byteSize) } : {}); + response.end(); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const names = ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN", "AWS_PROFILE", "AWS_DEFAULT_PROFILE", "AWS_MAX_ATTEMPTS", "AWS_EC2_METADATA_DISABLED"]; + const previous = new Map(names.map(name => [name, process.env[name]])); + process.env.AWS_ACCESS_KEY_ID = "snapshot-reuse-test"; + process.env.AWS_SECRET_ACCESS_KEY = "snapshot-reuse-test-secret"; + process.env.AWS_MAX_ATTEMPTS = "2"; + process.env.AWS_EC2_METADATA_DISABLED = "true"; + for (const name of ["AWS_SESSION_TOKEN", "AWS_PROFILE", "AWS_DEFAULT_PROFILE"]) delete process.env[name]; + try { + const provider = createStorageService(createS3StorageProvider({ bucket: "snapshots", region: "us-east-1", + endpoint: `http://127.0.0.1:${(server.address() as AddressInfo).port}`, forcePathStyle: true })); + vi.spyOn(f.storage, "headObject").mockImplementation(provider.headObject); + if (outcome === "forbidden") { + await expect(f.save()).rejects.toBeDefined(); + expect(f.puts).not.toHaveBeenCalled(); + expect(readRemoteWorkspaceAuthority((await f.workspace()).metadata)).toMatchObject({ dirty: true, writerRunId: f.b.runId }); + } else { + const saved = await f.save(); + if (outcome === "retry") expect(saved).toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); + else expect(saved.snapshotId).not.toBe(f.savedA.snapshotId); + } + const operation = (await f.operations())[0]!; + const timing = operation.metadata!.saveTiming as { storage: { requests: { requests: number; attempts: number; failedRequests: number } } }; + expect(timing.storage.requests).toMatchObject({ requests: 1, attempts: outcome === "retry" ? 2 : 1, + failedRequests: outcome === "retry" ? 0 : 1 }); + expect(requests).toBe(outcome === "retry" ? 2 : 1); + } finally { + for (const [name, value] of previous) { if (value === undefined) delete process.env[name]; else process.env[name] = value; } + server.closeAllConnections(); + await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + } + }); + + it("reuses the saved archive after a real capture and records finalization, replay, restore and waiter wake", async () => { + const f = await fixture(), waiterId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: waiterId, companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, + status: "scheduled_retry", scheduledRetryReason: "workspace_busy", scheduledRetryAt: new Date(Date.now() + 300_000), + contextSnapshot: { issueId: f.issueId, workspaceBusyWait: { key: `execution:${f.a.workspaceId}`, holderRunId: f.b.runId, + waitingSince: new Date().toISOString(), issueExecutionWorkspaceId: f.a.workspaceId, issueProjectWorkspaceId: null } } }); + const saved = await f.save(); + expect(saved).toMatchObject({ snapshotId: f.savedA.snapshotId, generation: f.savedA.generation, unchanged: true }); + expect(f.captures()).toBeGreaterThan(0); + expect(f.puts).not.toHaveBeenCalled(); + expect(readRemoteWorkspaceAuthority((await f.workspace()).metadata)).toMatchObject({ snapshotId: saved.snapshotId, dirty: false, writerRunId: null }); + expect(await f.operations()).toEqual([expect.objectContaining({ status: "succeeded", + metadata: expect.objectContaining({ snapshotId: saved.snapshotId, reusedSnapshotId: saved.snapshotId, + exportCleanup: expect.objectContaining({ state: "cleaned" }), saveWork: expect.objectContaining({ activeIds: [] }) }) })]); + expect((await f.snapshots()).filter(s => s.sourceRunId === f.b.runId)) + .toEqual([expect.objectContaining({ status: "deleted", objectKey: null, deleteAfter: null })]); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.b.runId)); + expect(run.runnerProfileJson?.remoteWorkspace).toMatchObject({ snapshotId: saved.snapshotId, state: "saved" }); + const [waiter] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, waiterId)); + expect(waiter.scheduledRetryAt!.getTime()).toBeLessThanOrEqual(Date.now()); + const beforeReplay = f.captures(); + await expect(f.save()).resolves.toMatchObject({ snapshotId: saved.snapshotId, unchanged: true }); + expect(f.captures()).toBe(beforeReplay); + expect(f.puts).not.toHaveBeenCalled(); + const loader = await createSavedWorkspaceLoader(db, f.storage)({ companyId: f.companyId, + workspaceId: f.a.workspaceId, snapshotId: saved.snapshotId }); + try { expect(await readFile(path.join(loader.root, "saved.txt"), "utf8")).toBe("saved result\n"); } + finally { await loader.cleanup(); } + expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, f.companyId), + eq(activityLog.runId, f.b.runId), eq(activityLog.action, "workspace.snapshot_skipped_unchanged")))).toHaveLength(1); + }); + + it("uploads changed contents and leaves their prior version on normal retention", async () => { + const f = await fixture(); + await writeFile(path.join(f.b.remoteRoot, "saved.txt"), "changed result\n"); + const saved = await f.save(); + expect(saved.snapshotId).not.toBe(f.savedA.snapshotId); + expect(saved).not.toHaveProperty("unchanged"); + expect(f.puts).toHaveBeenCalledTimes(1); + const previous = (await f.snapshots()).find(s => s.id === f.savedA.snapshotId)!; + expect(previous.status).toBe("superseded"); + expect(previous.deleteAfter!.getTime()).toBeGreaterThan(Date.now() + 6 * 86400_000); + }); + + it("reuses intermediate checkpoints without ending the writer, then reuses the final save", async () => { + const f = await fixture(); + await expect(f.save("repositories:unchanged")).resolves.toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); + expect(readRemoteWorkspaceAuthority((await f.workspace()).metadata)).toMatchObject({ dirty: true, writerRunId: f.b.runId }); + await expect(f.save()).resolves.toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); + expect(f.puts).not.toHaveBeenCalled(); + expect(await f.operations()).toHaveLength(2); + }); + + it("stores a fresh archive if the matching parent object is missing", async () => { + const f = await fixture(); + const parent = (await f.snapshots()).find(s => s.id === f.savedA.snapshotId)!; + await f.storage.deleteObject(f.companyId, parent.objectKey!); + const saved = await f.save(); + expect(saved.snapshotId).not.toBe(parent.id); + expect(f.puts).toHaveBeenCalledTimes(1); + expect(readRemoteWorkspaceAuthority((await f.workspace()).metadata)).toMatchObject({ dirty: false, writerRunId: null }); + }); + + it("requires the wrapped plaintext hash as well as matching manifest metadata", async () => { + const f = await fixture(); + await writeFile(path.join(f.b.remoteRoot, "saved.txt"), "new content despite stale metadata\n"); + const captured = await captureRemoteWorkspaceDirectory({ cwd: f.b.remoteRoot, archivePath: path.join(root, `${f.issueId}.pcws`) }); + await db.update(executionWorkspaceSnapshots).set({ manifestDigest: captured.manifestDigest }) + .where(eq(executionWorkspaceSnapshots.id, f.savedA.snapshotId)); + const saved = await f.save(); + expect(saved.snapshotId).not.toBe(f.savedA.snapshotId); + expect(f.puts).toHaveBeenCalledTimes(1); + const loaded = await createSavedWorkspaceLoader(db, f.storage)({ companyId: f.companyId, + workspaceId: f.a.workspaceId, snapshotId: saved.snapshotId }); + try { expect(await readFile(path.join(loaded.root, "saved.txt"), "utf8")).toBe("new content despite stale metadata\n"); } + finally { await loaded.cleanup(); } + }); + + it("does not reuse an archive whose object metadata has the wrong size", async () => { + const f = await fixture(); + vi.spyOn(f.storage, "headObject").mockResolvedValueOnce({ exists: true, contentLength: 1 }); + expect((await f.save()).snapshotId).not.toBe(f.savedA.snapshotId); + expect(f.puts).toHaveBeenCalledTimes(1); + }); + + it("does not reuse another company's identical manifest", async () => { + const f = await fixture(), foreign = await fixture(); + const row = await f.workspace(), authority = readRemoteWorkspaceAuthority(row.metadata)!; + await db.update(executionWorkspaces).set({ metadata: { ...row.metadata, + remoteWorkspace: { ...authority, snapshotId: foreign.savedA.snapshotId } } }).where(eq(executionWorkspaces.id, row.id)); + const heads = vi.spyOn(f.storage, "headObject"); + const saved = await f.save(); + expect(saved.snapshotId).not.toBe(foreign.savedA.snapshotId); + expect(heads).not.toHaveBeenCalled(); + expect(f.puts).toHaveBeenCalledTimes(1); + expect((await foreign.snapshots()).find(s => s.id === foreign.savedA.snapshotId)!.status).toBe("committed"); + }); + + it.each(["epoch", "writer", "location"])("fences a changed %s while the object metadata request is in flight", async changed => { + const f = await fixture(), head = f.storage.headObject.bind(f.storage); + let moved = readRemoteWorkspaceAuthority((await f.workspace()).metadata)!; + vi.spyOn(f.storage, "headObject").mockImplementationOnce(async (...args) => { + const result = await head(...args), row = await f.workspace(), authority = readRemoteWorkspaceAuthority(row.metadata)!; + moved = { ...authority, ...(changed === "epoch" ? { epoch: authority.epoch + 1 } + : changed === "writer" ? { writerRunId: randomUUID() } + : { location: { ...authority.location!, providerLeaseId: randomUUID() } }) }; + await db.update(executionWorkspaces).set({ metadata: { ...row.metadata, remoteWorkspace: moved } }) + .where(eq(executionWorkspaces.id, row.id)); + return result; + }); + await expect(f.save()).rejects.toThrow("publication_fenced"); + expect(f.puts).not.toHaveBeenCalled(); + expect(readRemoteWorkspaceAuthority((await f.workspace()).metadata)).toEqual(moved); + expect((await f.snapshots()).filter(s => s.sourceRunId === f.b.runId)).toEqual([expect.objectContaining({ status: "pending" })]); + }); + + it("preserves a newly journaled upload attempt instead of discarding uncertain producer work", async () => { + const f = await fixture(), head = f.storage.headObject.bind(f.storage); + vi.spyOn(f.storage, "headObject").mockImplementationOnce(async (...args) => { + const result = await head(...args); + const pending = (await f.snapshots()).find(s => s.sourceRunId === f.b.runId)!; + await db.update(executionWorkspaceSnapshots).set({ metadata: { uploads: [{ objectKey: `${f.companyId}/pending-owned-attempt` }] } }) + .where(eq(executionWorkspaceSnapshots.id, pending.id)); + return result; + }); + await expect(f.save()).rejects.toThrow("publication_changed"); + expect(f.puts).not.toHaveBeenCalled(); + expect((await f.snapshots()).find(s => s.sourceRunId === f.b.runId)).toMatchObject({ status: "pending", + metadata: { uploads: [{ objectKey: `${f.companyId}/pending-owned-attempt` }] } }); + expect(readRemoteWorkspaceAuthority((await f.workspace()).metadata)).toMatchObject({ dirty: true, writerRunId: f.b.runId }); + }); + + it("does not hold the workspace lock during a slow storage metadata request", async () => { + const f = await fixture(), started = deferred(), release = deferred(), head = f.storage.headObject.bind(f.storage); + vi.spyOn(f.storage, "headObject").mockImplementationOnce(async (...args) => { + started.resolve(); await release.promise; return head(...args); + }); + const save = f.save(); + await started.promise; + try { + await db.transaction(async tx => { + await tx.execute(sql`set local lock_timeout = '500ms'`); + expect(await tx.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.a.workspaceId)).for("update")).toHaveLength(1); + }); + } finally { release.resolve(); } + await expect(save).resolves.toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); + }); + + it("keeps concurrent final saves on one durable reused receipt", async () => { + const f = await fixture(), first = deferred(), both = deferred(), head = f.storage.headObject.bind(f.storage); + let requests = 0; + vi.spyOn(f.storage, "headObject").mockImplementation(async (...args) => { + requests++; first.resolve(); if (requests === 2) both.resolve(); await both.promise; return head(...args); + }); + const a = f.save(); await first.promise; const b = f.save(); + const saved = await Promise.all([a, b]); + expect(saved).toEqual([expect.objectContaining({ snapshotId: f.savedA.snapshotId, unchanged: true }), + expect.objectContaining({ snapshotId: f.savedA.snapshotId, unchanged: true })]); + expect(f.puts).not.toHaveBeenCalled(); + expect(await f.operations()).toEqual([expect.objectContaining({ status: "succeeded" })]); + expect((await f.snapshots()).filter(s => s.sourceRunId === f.b.runId)).toHaveLength(1); + }); + + it("reads a peer's completed receipt when its own registered save has not begun its snapshot", async () => { + const f = await fixture(), headStarted = deferred(), releaseHead = deferred(), beginStarted = deferred(), releaseBegin = deferred(); + const head = f.storage.headObject.bind(f.storage), transaction = db.transaction.bind(db); + vi.spyOn(f.storage, "headObject").mockImplementationOnce(async (...args) => { + headStarted.resolve(); await releaseHead.promise; return head(...args); + }); + const first = f.save(); + await headStarted.promise; + let transactions = 0; + const delayed = vi.spyOn(db, "transaction").mockImplementation(async (action, config) => { + if (++transactions === 2) { beginStarted.resolve(); await releaseBegin.promise; } + return transaction(action, config); + }); + const second = f.save(); + try { + await beginStarted.promise; + releaseHead.resolve(); + await expect(first).resolves.toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); + expect((await f.snapshots()).find(s => s.sourceRunId === f.b.runId)!.status).toBe("deleted"); + releaseBegin.resolve(); + await expect(second).resolves.toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); + } finally { + releaseHead.resolve(); releaseBegin.resolve(); + await Promise.allSettled([first, second]); + delayed.mockRestore(); + } + expect(f.captures()).toBe(1); + expect(f.puts).not.toHaveBeenCalled(); + expect(await f.operations()).toEqual([expect.objectContaining({ status: "succeeded", + metadata: expect.objectContaining({ saveWork: expect.objectContaining({ activeIds: [], unknown: false }) }) })]); + }); + + it.each([false, true])("retires a prior save only when its work is known to be settled (unknown=%s)", async unknown => { + const f = await fixture(), row = await f.workspace(), authority = readRemoteWorkspaceAuthority(row.metadata)!; + const operationId = randomUUID(), at = new Date(); + await db.insert(workspaceOperations).values({ id: operationId, companyId: f.companyId, executionWorkspaceId: row.id, + heartbeatRunId: f.b.runId, issueId: f.issueId, phase: "remote_workspace_snapshot", status: "failed", + metadata: { boundary: "repositories:prior", epoch: authority.epoch, generation: f.savedA.generation + 1, + lastFailure: { code: "remote_workspace_save_failed", at: at.toISOString() }, + saveWork: { schema: "paperclip.remote-workspace-save-work.v1", activeIds: [], unknown, quiescedAt: at.toISOString() }, + exportCleanup: { ...authority.location, state: "pending" } } }); + await beginWorkspaceSnapshot(db, { companyId: f.companyId, workspaceId: row.id, operationId, runId: f.b.runId, + parentSnapshotId: f.savedA.snapshotId, generation: f.savedA.generation + 1, epoch: authority.epoch }); + await expect(f.save()).resolves.toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); + const previous = (await f.operations()).find(o => o.id === operationId)!; + if (unknown) expect(previous.metadata).not.toHaveProperty("retiredSave"); + else expect(previous.metadata).toMatchObject({ retiredSave: { snapshotId: f.savedA.snapshotId }, + exportCleanup: { state: "cleaned" } }); + expect(f.puts).not.toHaveBeenCalled(); + }); + + it("keeps the reused current archive pinned when its old retention deadline passes", async () => { + const f = await fixture(); + await f.save(); + const future = new Date(Date.now() + 30 * 86400_000); + await db.update(executionWorkspaceSnapshots).set({ deleteAfter: new Date(0) }).where(eq(executionWorkspaceSnapshots.id, f.savedA.snapshotId)); + const retention = remoteWorkspaceSnapshotRetentionService(db, { storage: f.storage, now: () => future }); + for (let pass = 0; pass < 10; pass++) { + const result = await retention.tick(); + expect(result.failed).toBe(0); + if (!result.truncated) break; + expect(pass).toBeLessThan(9); + } + const parent = (await f.snapshots()).find(s => s.id === f.savedA.snapshotId)!; + expect(parent).toMatchObject({ status: "committed", deleteAfter: null }); + await expect(f.storage.headObject(f.companyId, parent.objectKey!)).resolves.toMatchObject({ exists: true }); + }); +}); diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index 2e7f38fe228..b79a1dcf21e 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -105,6 +105,31 @@ export async function getWorkspaceSnapshot(db: Db, input: { companyId: string; w return row && ["uploaded", "committed", "superseded"].includes(row.status) ? reference(row) : null; } +export async function findUnchangedWorkspaceSnapshot(input: { + db: Db; storage: StorageService; snapshot: SnapshotRow; manifestDigest: string; archiveSha256: string; archiveBytes: number; + timing?: WorkspaceSnapshotStorageTiming; +}): Promise { + return withRequestStats(input.timing, async () => { + const { snapshot } = input; + if (!snapshot.parentSnapshotId || snapshot.status !== "pending" || snapshot.objectKey !== null + || snapshot.sha256 !== null || snapshot.byteSize !== null || snapshot.key !== null || snapshot.manifestDigest !== null + || snapshot.metadata?.uploads !== undefined + && (!Array.isArray(snapshot.metadata.uploads) || snapshot.metadata.uploads.length !== 0)) return null; + const [parent] = await input.db.select().from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, snapshot.companyId), + eq(executionWorkspaceSnapshots.executionWorkspaceId, snapshot.executionWorkspaceId), + eq(executionWorkspaceSnapshots.id, snapshot.parentSnapshotId), + eq(executionWorkspaceSnapshots.status, "committed"), + eq(executionWorkspaceSnapshots.manifestDigest, input.manifestDigest))); + if (!parent) return null; + const ref = reference(parent); + const key = await snapshotKey(ref); + if (key.plaintextSha256 !== input.archiveSha256 || key.plaintextBytes !== input.archiveBytes) return null; + const head = await input.storage.headObject(ref.companyId, ref.objectKey); + return head.exists && head.contentLength === ref.byteSize ? ref : null; + }); +} + export async function recoverWorkspaceSnapshotArchive(input: { db: Db; storage: StorageService; snapshotId: string; companyId: string; workspaceId: string; targetPath: string; signal?: AbortSignal; diff --git a/server/src/services/remote-workspace-unchanged-release.test.ts b/server/src/services/remote-workspace-unchanged-release.test.ts index d571f2ad533..dadd0b2a426 100644 --- a/server/src/services/remote-workspace-unchanged-release.test.ts +++ b/server/src/services/remote-workspace-unchanged-release.test.ts @@ -123,12 +123,14 @@ describe.skipIf(!support.supported)("remote workspace writer that ended without captures: () => captures }; } - async function expectFullSave(f: Awaited>) { + async function expectFullSave(f: Awaited>, changed = false) { await expect(f.reserveNext()).rejects.toBeInstanceOf(RemoteWorkspaceWriterActiveError); const before = f.captures(); const saved = await f.saveB(); - expect(saved).not.toHaveProperty("unchanged"); - expect(saved.snapshotId).not.toBe(f.savedA.snapshotId); + if (changed) { + expect(saved).not.toHaveProperty("unchanged"); + expect(saved.snapshotId).not.toBe(f.savedA.snapshotId); + } else expect(saved).toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); expect(f.captures()).toBeGreaterThan(before); expect(await f.snapshotOperations()).toHaveLength(1); } @@ -174,7 +176,7 @@ describe.skipIf(!support.supported)("remote workspace writer that ended without it("keeps the full save when setup wrote into the workspace", async () => { const f = await preparedFollowUp({ runtimeProvisionCommand: "echo provisioned > provisioned.txt" }); await f.end(); - await expectFullSave(f); + await expectFullSave(f, true); }); it("keeps the full save while the writer's owner may still act, and while it is still running", async () => { @@ -183,8 +185,10 @@ describe.skipIf(!support.supported)("remote workspace writer that ended without await expect(f.reserveNext()).rejects.toBeInstanceOf(RemoteWorkspaceWriterActiveError); await db.update(heartbeatRuns).set({ status: "running", controllerLeaseExpiresAt: null, finishedAt: null }) .where(eq(heartbeatRuns.id, f.runB)); + const before = f.captures(); const saved = await f.saveB(); - expect(saved).not.toHaveProperty("unchanged"); + expect(saved).toMatchObject({ snapshotId: f.savedA.snapshotId, unchanged: true }); + expect(f.captures()).toBeGreaterThan(before); expect(await f.snapshotOperations()).toHaveLength(1); }); diff --git a/server/src/services/remote-workspaces.integration.test.ts b/server/src/services/remote-workspaces.integration.test.ts index 06788b767a2..a990f11d667 100644 --- a/server/src/services/remote-workspaces.integration.test.ts +++ b/server/src/services/remote-workspaces.integration.test.ts @@ -1342,6 +1342,7 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem it.each(["journal", "ledger"])("respects historical %s generation reservations without changing the committed parent", async highest => { const reserved = await reserve({ kind: "empty" }), first = await target("reserved-generations"); const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: ids.run, lease: await lease(), target: first }); + await fs.writeFile(path.join(prepared.remoteRoot, "new-generation.txt"), "new workspace contents\n"); const parent = readRemoteWorkspaceAuthority(prepared.workspace.metadata)!.snapshotId; // A crash after journaling can leave no ledger row; retention can leave a // deleted ledger after the source operation is gone. Both remain reserved. @@ -1399,6 +1400,7 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem it("reserves distinct generations for competing boundaries and fences the losing publication", async () => { const reserved = await reserve({ kind: "empty" }), first = await target("competing-boundaries"); const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: ids.run, lease: await lease(), target: first }); + await fs.writeFile(path.join(prepared.remoteRoot, "competing-save.txt"), "new workspace contents\n"); const parent = readRemoteWorkspaceAuthority(prepared.workspace.metadata)!.snapshotId; let arrivals = 0, release!: () => void; const gate = new Promise(resolve => { release = resolve; }); @@ -1435,6 +1437,7 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem it.each(["failed", "succeeded"])("protects an active duplicate after its peer %s and cleans it only after the last invocation settles", async peer => { const reserved = await reserve({ kind: "empty" }), first = await target("active-duplicate"); const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: ids.run, lease: await lease(), target: first }); + await fs.writeFile(path.join(prepared.remoteRoot, "duplicate-save.txt"), "new workspace contents\n"); let release!: () => void, entered!: () => void, releaseOwnership!: () => void, postUpload!: () => void, uploads = 0, checks = 0; const blocked = new Promise(resolve => { release = resolve; }); const started = new Promise(resolve => { entered = resolve; }); @@ -1503,6 +1506,7 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem it.each(["untracked", "malformed-owner", "malformed-active"])("keeps adopted %s work pinned after a newer final save", async marker => { const reserved = await reserve({ kind: "empty" }), first = await target("unknown-save-work"); const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: ids.run, lease: await lease(), target: first }); + await fs.writeFile(path.join(prepared.remoteRoot, "adopted-save.txt"), "new workspace contents\n"); const input = { db, storage, workspace: prepared.workspace, runId: ids.run, lease: prepared.lease, target: first, assertOwned: async () => undefined, boundary: "repository:legacy" }; const failed: StorageService = { ...storage, putFile: async () => { throw new Error("old upload failed"); } }; @@ -1518,6 +1522,7 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem if (marker === "malformed-active") await db.update(workspaceOperations).set({ metadata: sql`jsonb_set(jsonb_set(${workspaceOperations.metadata}, '{saveWork,unknown}', 'false'::jsonb), '{saveWork,activeIds}', '{"unfinished":"unknown"}'::jsonb)` }) .where(eq(workspaceOperations.id, old!.sourceOperationId)); + await fs.writeFile(path.join(prepared.remoteRoot, "after-adopted-save.txt"), "newer final workspace contents\n"); const final = await saveRemoteWorkspace({ ...input, workspace: await workspace(reserved.id), boundary: "final" }); const [held] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, old!.sourceOperationId)); expect(held).toMatchObject({ status: "succeeded", metadata: { exportCleanup: { state: "pending" } } }); diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index ea511c813df..3c4c3c69376 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -7,7 +7,7 @@ import { existsSync } from "node:fs"; import { mkdtemp, readFile, rm, statfs } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { and, eq, inArray, sql } from "drizzle-orm"; +import { and, eq, inArray, isNull, or, sql } from "drizzle-orm"; import { companies, environmentLeases, executionWorkspaces, executionWorkspaceSnapshots, heartbeatRuns, issues, workspaceOperations, type Db, } from "@paperclipai/db"; @@ -34,7 +34,7 @@ import { type RemoteWorkspaceSavePhaseTiming, } from "./remote-workspace-transport.js"; import { - beginWorkspaceSnapshot, getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, newWorkspaceSnapshotStorageTiming, uploadWorkspaceSnapshot, + beginWorkspaceSnapshot, findUnchangedWorkspaceSnapshot, getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, newWorkspaceSnapshotStorageTiming, uploadWorkspaceSnapshot, type WorkspaceSnapshotStorageTiming, } from "./remote-workspace-snapshot-storage.js"; import { measureRunPhase } from "./execution-preparation-timing.js"; @@ -651,15 +651,20 @@ async function retireQuiescedSaves(tx: Transaction, workspace: Workspace, runId: || !authority.snapshotId || !sameExportLocation(authority.location, location)) return; const [final] = await tx.select({ id: executionWorkspaceSnapshots.id, generation: executionWorkspaceSnapshots.generation, operationId: workspaceOperations.id, metadata: workspaceOperations.metadata }).from(executionWorkspaceSnapshots) - .innerJoin(workspaceOperations, and(eq(workspaceOperations.id, executionWorkspaceSnapshots.sourceOperationId), + .innerJoin(workspaceOperations, and(or(eq(workspaceOperations.id, executionWorkspaceSnapshots.sourceOperationId), + sql`${workspaceOperations.metadata}->>'reusedSnapshotId' = ${executionWorkspaceSnapshots.id}::text`), eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id), eq(workspaceOperations.heartbeatRunId, runId), eq(workspaceOperations.phase, "remote_workspace_snapshot"), eq(workspaceOperations.status, "succeeded"), sql`${workspaceOperations.metadata}->>'boundary' = 'final'`, sql`${workspaceOperations.metadata}->>'epoch' = ${String(epoch)}`)) .where(and(eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), - eq(executionWorkspaceSnapshots.id, authority.snapshotId), eq(executionWorkspaceSnapshots.sourceRunId, runId), - eq(executionWorkspaceSnapshots.epoch, epoch), eq(executionWorkspaceSnapshots.status, "committed"))).limit(1); + eq(executionWorkspaceSnapshots.id, authority.snapshotId), + or(and(eq(executionWorkspaceSnapshots.sourceRunId, runId), eq(executionWorkspaceSnapshots.epoch, epoch)), + sql`${workspaceOperations.metadata}->>'reusedSnapshotId' = ${executionWorkspaceSnapshots.id}::text`), + eq(executionWorkspaceSnapshots.status, "committed"))).limit(1); if (!final || final.metadata?.snapshotId !== final.id || !sameExportLocation(final.metadata?.exportCleanup, location)) return; + const finalGeneration = final.metadata?.reusedSnapshotId === final.id ? Number(final.metadata.generation) : final.generation; + if (!Number.isSafeInteger(finalGeneration) || finalGeneration < final.generation) return; const candidates = await tx.select().from(workspaceOperations).where(and( eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id), eq(workspaceOperations.heartbeatRunId, runId), eq(workspaceOperations.phase, "remote_workspace_snapshot"), @@ -671,7 +676,7 @@ async function retireQuiescedSaves(tx: Transaction, workspace: Workspace, runId: || !candidate.metadata?.lastFailure)) continue; const work = readSaveWork(candidate.metadata), generation = Number(candidate.metadata?.generation); if (!work || work.unknown || work.activeIds.length || !work.quiescedAt || candidate.metadata?.boundary === "final" - || !Number.isSafeInteger(generation) || generation >= final.generation + || !Number.isSafeInteger(generation) || generation >= finalGeneration || !sameExportLocation(candidate.metadata?.exportCleanup, location)) continue; const [snapshot] = await tx.select({ status: executionWorkspaceSnapshots.status, generation: executionWorkspaceSnapshots.generation, epoch: executionWorkspaceSnapshots.epoch, runId: executionWorkspaceSnapshots.sourceRunId }).from(executionWorkspaceSnapshots).where(and( @@ -878,6 +883,25 @@ export async function saveRemoteWorkspace(input: { replayedCommittedSave = true; return { snapshotId: snapshot.id, generation: snapshot.generation, sha256: snapshot.sha256!, savedAt: snapshot.updatedAt.toISOString() }; } + if (snapshot.status === "deleted" && snapshot.parentSnapshotId) { + const [completed] = await input.db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.id, operation.id), eq(workspaceOperations.companyId, snapshot.companyId), + eq(workspaceOperations.executionWorkspaceId, snapshot.executionWorkspaceId), + eq(workspaceOperations.heartbeatRunId, input.runId), eq(workspaceOperations.phase, "remote_workspace_snapshot"), + eq(workspaceOperations.status, "succeeded"), + sql`${workspaceOperations.metadata}->>'boundary' = ${boundary}`, + sql`${workspaceOperations.metadata}->>'epoch' = ${String(expected.epoch)}`, + sql`${workspaceOperations.metadata}->>'reusedSnapshotId' = ${snapshot.parentSnapshotId}`)); + if (completed?.finishedAt) { + const saved = await getWorkspaceSnapshot(input.db, { companyId: snapshot.companyId, + workspaceId: snapshot.executionWorkspaceId, snapshotId: snapshot.parentSnapshotId }); + if (!saved || completed.metadata?.snapshotId !== saved.id || completed.metadata?.sha256 !== saved.sha256) + throw new Error("remote_workspace_snapshot_missing"); + replayedCommittedSave = true; + return { snapshotId: saved.id, generation: saved.generation, sha256: saved.sha256, + savedAt: completed.finishedAt.toISOString(), unchanged: true as const }; + } + } const existingUpload = snapshot.status === "uploaded" ? await getWorkspaceSnapshot(input.db, { companyId: input.workspace.companyId, workspaceId: input.workspace.id, snapshotId: snapshot.id }) : null; if (existingUpload) timing.reusedUpload = true; @@ -885,9 +909,14 @@ export async function saveRemoteWorkspace(input: { operationId: operation.id, timing }); await input.assertOwned(); const uploadStarted = performance.now(); - const saved = await uploadWorkspaceSnapshot({ db: input.db, storage: input.storage ?? getStorageService(), snapshotId: snapshot.id, + const storage = input.storage ?? getStorageService(); + const storageTiming = timing.storage ??= newWorkspaceSnapshotStorageTiming(); + const unchanged = captured && readSaveWork(operation.metadata)?.unknown === false + ? await findUnchangedWorkspaceSnapshot({ db: input.db, storage, snapshot, manifestDigest: captured.manifestDigest, + archiveSha256: captured.sha256, archiveBytes: captured.byteSize, timing: storageTiming }) : null; + const saved = unchanged ?? await uploadWorkspaceSnapshot({ db: input.db, storage, snapshotId: snapshot.id, companyId: input.workspace.companyId, workspaceId: input.workspace.id, archivePath: captured?.archivePath ?? "", - manifestDigest: existingUpload?.manifestDigest ?? captured!.manifestDigest, timing: timing.storage ??= newWorkspaceSnapshotStorageTiming() }); + manifestDigest: existingUpload?.manifestDigest ?? captured!.manifestDigest, timing: storageTiming }); timing.uploadMs = Math.max(0, Math.round(performance.now() - uploadStarted)); await input.assertOwned(); const publishStarted = performance.now(); @@ -903,15 +932,39 @@ export async function saveRemoteWorkspace(input: { await lockWorkspaceRunParents(tx, input.workspace.companyId, input.runId, operation.issueId); const workspace = await currentWorkspace(tx, input.workspace.companyId, input.workspace.id); const current = readRemoteWorkspaceAuthority(workspace.metadata); - if (!current || current.epoch !== expected.epoch || current.writerRunId !== input.runId || current.snapshotId !== snapshot.parentSnapshotId) { - if (current?.epoch === expected.epoch && current.snapshotId === saved.id) return; + if (unchanged) { + const [published] = await tx.select().from(workspaceOperations).where(and(eq(workspaceOperations.id, operation.id), + eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id), + eq(workspaceOperations.heartbeatRunId, input.runId))).for("update"); + if (published?.status === "succeeded" && published.metadata?.reusedSnapshotId === saved.id) return; + } + if (!current || current.epoch !== expected.epoch || current.writerRunId !== input.runId || current.snapshotId !== snapshot.parentSnapshotId + || unchanged && !sameExportLocation(current.location, expected.location!)) { + if (!unchanged && current?.epoch === expected.epoch && current.snapshotId === saved.id) return; throw new Error("remote_workspace_publication_fenced"); } + if (unchanged) { + const [parent] = await tx.select({ id: executionWorkspaceSnapshots.id }).from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.id, saved.id), eq(executionWorkspaceSnapshots.companyId, workspace.companyId), + eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.status, "committed"), + eq(executionWorkspaceSnapshots.manifestDigest, saved.manifestDigest), eq(executionWorkspaceSnapshots.objectKey, saved.objectKey), + eq(executionWorkspaceSnapshots.sha256, saved.sha256), eq(executionWorkspaceSnapshots.byteSize, saved.byteSize), + eq(executionWorkspaceSnapshots.generation, saved.generation))).for("update"); + if (!parent) throw new Error("remote_workspace_snapshot_publication_changed"); + } const authority: RemoteWorkspaceAuthority = { ...current, snapshotId: saved.id, dirty: boundary !== "final", writerRunId: boundary !== "final" ? input.runId : null }; - const [committed] = await tx.update(executionWorkspaceSnapshots).set({ status: "committed", updatedAt: at }).where(and( - eq(executionWorkspaceSnapshots.id, saved.id), eq(executionWorkspaceSnapshots.companyId, workspace.companyId), - eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.status, "uploaded"))).returning({ id: executionWorkspaceSnapshots.id }); + const [committed] = unchanged + ? await tx.update(executionWorkspaceSnapshots).set({ status: "deleted", deleteAfter: null, updatedAt: at }).where(and( + eq(executionWorkspaceSnapshots.id, snapshot.id), eq(executionWorkspaceSnapshots.companyId, workspace.companyId), + eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.status, "pending"), + isNull(executionWorkspaceSnapshots.objectKey), isNull(executionWorkspaceSnapshots.sha256), + isNull(executionWorkspaceSnapshots.byteSize), isNull(executionWorkspaceSnapshots.key), isNull(executionWorkspaceSnapshots.manifestDigest), + sql`coalesce(${executionWorkspaceSnapshots.metadata}->'uploads', '[]'::jsonb) = '[]'::jsonb` + )).returning({ id: executionWorkspaceSnapshots.id }) + : await tx.update(executionWorkspaceSnapshots).set({ status: "committed", updatedAt: at }).where(and( + eq(executionWorkspaceSnapshots.id, saved.id), eq(executionWorkspaceSnapshots.companyId, workspace.companyId), + eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.status, "uploaded"))).returning({ id: executionWorkspaceSnapshots.id }); if (!committed) throw new Error("remote_workspace_snapshot_publication_changed"); await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); if (boundary === "final") { @@ -932,18 +985,20 @@ export async function saveRemoteWorkspace(input: { remoteWorkspaceSnapshotRepositories: { snapshotId: saved.id, repositories }, }, updatedAt: at }).where(eq(executionWorkspaces.id, workspace.id)); await tx.update(workspaceOperations).set({ status: "succeeded", finishedAt: at, updatedAt: at, - metadata: sql`coalesce(${workspaceOperations.metadata}, '{}'::jsonb) || ${JSON.stringify({ snapshotId: saved.id, sha256: saved.sha256 })}::jsonb` }) + metadata: sql`coalesce(${workspaceOperations.metadata}, '{}'::jsonb) || ${JSON.stringify({ snapshotId: saved.id, sha256: saved.sha256, + ...(unchanged ? { reusedSnapshotId: saved.id } : {}) })}::jsonb` }) .where(eq(workspaceOperations.id, operation.id)); await persistRunReference(tx, workspace.companyId, input.runId, reference(workspace.id, authority, boundary === "final" ? "saved" : "prepared")); await persistActivity(tx as unknown as Db, { companyId: workspace.companyId, actorType: "system", actorId: "remote-workspaces", - action: "workspace.snapshot_saved", entityType: "execution_workspace", entityId: workspace.id, runId: input.runId, + action: unchanged ? "workspace.snapshot_skipped_unchanged" : "workspace.snapshot_saved", entityType: "execution_workspace", entityId: workspace.id, runId: input.runId, details: { snapshotId: saved.id, generation: saved.generation, epoch: current.epoch, boundary } }); - if (snapshot.parentSnapshotId) await tx.update(executionWorkspaceSnapshots).set({ status: "superseded", + if (snapshot.parentSnapshotId && !unchanged) await tx.update(executionWorkspaceSnapshots).set({ status: "superseded", deleteAfter: new Date(at.getTime() + 7 * 86400_000), updatedAt: at }).where(and(eq(executionWorkspaceSnapshots.id, snapshot.parentSnapshotId), eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id))); }); timing.publishMs = Math.max(0, Math.round(performance.now() - publishStarted)); - return { snapshotId: saved.id, generation: saved.generation, sha256: saved.sha256, savedAt: at.toISOString() }; + return { snapshotId: saved.id, generation: saved.generation, sha256: saved.sha256, savedAt: at.toISOString(), + ...(unchanged ? { unchanged: true as const } : {}) }; } catch (error) { failure = remoteWorkspaceSaveFailure(error); throw error; diff --git a/server/src/services/workspace-recovery-receipt.ts b/server/src/services/workspace-recovery-receipt.ts index c0365f4eaf5..a79334c7dc7 100644 --- a/server/src/services/workspace-recovery-receipt.ts +++ b/server/src/services/workspace-recovery-receipt.ts @@ -6,6 +6,7 @@ import { readNativeWorkspaceSyncReference, type NativeWorkspaceSyncReference } f import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js"; import { readSnapshot } from "./workspace-recovery-snapshot.js"; import { exhaustedSystemRecoveryAction, readSystemRecovery } from "./system-recovery-state.js"; +import { verifiedWorkspaceSnapshotReuse } from "./remote-workspace-snapshot-reuse-receipt.js"; export const workspaceRecoverySchema = z.object({ schema: z.literal("paperclip.workspace-recovery.v1"), @@ -161,9 +162,10 @@ export async function verifiedWorkspaceRepairReceipt(db: Db, run: Pick>'recoveryActionId'`)) .innerJoin(executionWorkspaceSnapshots, and(eq(executionWorkspaceSnapshots.companyId, run.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, remote.workspaceId), - eq(executionWorkspaceSnapshots.id, snapshotId), eq(executionWorkspaceSnapshots.sourceRunId, run.id), - eq(executionWorkspaceSnapshots.epoch, remote.epoch), inArray(executionWorkspaceSnapshots.status, ["committed", "superseded"]))) + eq(executionWorkspaceSnapshots.id, snapshotId), inArray(executionWorkspaceSnapshots.status, ["committed", "superseded"]))) .where(and(eq(workspaceOperations.companyId, run.companyId), eq(workspaceOperations.issueId, issueId), eq(workspaceOperations.executionWorkspaceId, remote.workspaceId), eq(workspaceOperations.heartbeatRunId, run.id), eq(workspaceOperations.phase, "workspace_recovery"), eq(workspaceOperations.status, "succeeded"), @@ -207,7 +207,11 @@ export async function immutableRemoteWorkspaceRecoveryAttachment(db: Db, run: Pi const proof = saved?.metadata, state = readWorkspaceRecovery(saved?.evidence ?? null); if (!saved || !proof || state?.runId !== run.id || state.status !== "recovered" || proof.storage !== "remote_workspace" || proof.leaseId !== remote.leaseId || proof.providerLeaseId !== remote.providerLeaseId - || proof.epoch !== remote.epoch || proof.sha256 !== saved.sha256 || proof.generation !== saved.generation || !saved.sha256) return null; + || proof.epoch !== remote.epoch || proof.sha256 !== saved.snapshot.sha256 + || proof.generation !== saved.snapshot.generation || !saved.snapshot.sha256) return null; + if ((saved.snapshot.epoch !== remote.epoch || saved.snapshot.sourceRunId !== run.id) + && !await verifiedWorkspaceSnapshotReuse(db, { companyId: run.companyId, issueId, runId: run.id, reference: remote, + snapshot: saved.snapshot })) return null; return { kind: "immutable_remote_workspace_attachment" as const, operationId: saved.operationId, - workspaceId: remote.workspaceId, snapshotId, sha256: saved.sha256, generation: saved.generation }; + workspaceId: remote.workspaceId, snapshotId, sha256: saved.snapshot.sha256, generation: saved.snapshot.generation }; }