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 }; }