From 1b7fe33432016b3b27ccdfbf7b8944da4b40726d Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:03:48 -0600 Subject: [PATCH] fix(workspaces): preserve import verification across process restarts (#20311) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Online import preserves old local files before the web server can use a disposable filesystem. > - A parent import must authenticate saved child archives before it can omit their directories. > - The operation journal preserves import identity, but completed child checks existed only in process memory. > - A container restart therefore repeated downloads, decryption and restoration that had already succeeded. > - This pull request saves a versioned verification receipt on each immutable child snapshot. > - A replacement process can reuse those checks while it reacquires locks and verifies current local files. ## Linked Issues or Issue Description Refs #19854 for verification reuse within a live source fence. Refs #19873 for cohort import. Refs #20298 for the separate remote-hold admission fix. Searched related workspace verification and restart PRs; none provides durable child verification. **What happened?** Replacing a server container interrupts a long online import. The original operation and snapshot survive, but a later invocation downloads, decrypts and restores completed child archives again. The current cache ends with the source-lock owner. **Expected behavior** Completed immutable checks survive process and scratch loss. Resume keeps the same operation and snapshot, reuses only correctly bound verification, and publishes once. Current local files and Git state must still be checked under fresh source locks. **Steps to reproduce** 1. Import two nested child workspaces, then start the parent import. 2. Kill the importer after a child verification finishes. Remove its scratch directory. 3. Resume in a new process. Before this fix, the completed child is downloaded and restored again. **Paperclip version or commit** The regression fails on the previous migration verifier. The final branch includes master at `189fefb99183953be37f0d1084275fbb0f35d6e2`. **Deployment mode** Managed Linux server with PostgreSQL and explicit S3 storage configuration. ## What Changed - Add optional opaque storage namespace identity. S3 binds it to the explicit endpoint, region, bucket, normalized prefix and addressing style. Implicit endpoints retain full verification. - Save a versioned child-snapshot receipt only after canonical crypto, manifest and file restoration succeed and a second HEAD confirms the same strong ETag and size. - Bind reuse to the complete snapshot reference, key material, child lineage, source operation and storage namespace. Require a fresh strong HEAD and canonical validation that the current secret provider can unwrap the saved snapshot key. - Keep current directory/Git capture and source-owner checks on every invocation. Read-only verifier callers do not write checkpoints. - Merge checkpoint metadata under a short row lock without changing parent journal hashes, snapshot retention or update timestamps. Bound reads and writes. Lock timeouts, statement timeouts, deadlocks and serialization conflicts are safe cache misses. Other errors remain fatal. - Extend real PostgreSQL/Git/archive tests with process death, scratch removal, resume, changed bindings and single publication. ## Verification - Baseline RED: two import invocations read each child twice. With durable receipts, each child is read once across both invocations. - Real process regression passed on the final rebased head: SIGKILL the owned importer after one checkpoint, delete all scratch, and resume in a new Node process. It retains the same operation and snapshot, reads each child once across both processes, and records one publication. - 53 focused cases passed in 104.86 seconds on the final production bytes before rebase. They cover existing nested/cohort behavior, namespace identity, changed source/reference/key/lineage/ETag, unknown proof versions, weak validators and bounded row contention. - After clean rebase and stronger scratch/snapshot assertions, the real process case passed again in 18.54 seconds. Server TypeScript, module boundaries and diff checks passed on final head `9da68ee7ce4a778201ef85e21a11c49eeea7bc16`. Required CI supplies final-head broad coverage. - The initial restricted-network dependency setup failed; unchanged preflight passed with normal network access. Initial TypeScript failures in test pipe assertions were corrected. No failed attempt is counted as a pass. - Review found a missing current-master-key check. A real restart regression with all child checkpoints complete reproduced unsafe publication after switching to a different valid master key. Follow-up `7ea92c2abd9ffdb2e7dd81cecf62c8fc7a6e74a5` delegates to the canonical snapshot-key owner before both durable and in-memory cache hits. - The fixed process regressions passed 2/2. Ten key-authority and existing identity/source guard cases passed. The wrong-key process now refuses with zero additional child payload GETs, no publication, and unchanged original journal/snapshot rows. Server typecheck, module boundaries and diff checks passed on the follow-up. - Review also found that transient database failures could make the optional checkpoint cache stop a valid import. Read and save now fall back only for SQLSTATE `55P03`, `57014`, `40P01`, and `40001`. Connection loss, identity changes, crypto failures, programming errors and cancellation remain fatal. Reads use a read-only transaction with one-second lock and eight-second statement limits. - The checkpoint database-fault regression failed two cases before the fix. All four recoverable import cases then passed. The final selected run passed 21 cases, including fatal-error and cancellation controls, real row contention, both process-restart cases and key/identity guards. Server typecheck and boundaries passed. Fault injection is limited to the checkpoint database boundary; the importer, PostgreSQL, Git, archives, storage and publication remain real. - Latest master was merged without rewriting published history. Six real import/process-restart cases passed again on final head `8c8e4213a680ebb5c8d6864416bc3a4f271e1e76` in 38.03 seconds. [Required CI](https://github.com/paperclipai/paperclip/actions/runs/42658248222) finished with 54 successful checks and four intentional skips; Snyk passed. [Fresh review](https://github.com/paperclipai/paperclip/pull/20311#issuecomment-6745354381) is 5/5 with all four threads addressed and resolved. Hosted verification after normal release delivery remains pending. ## Risks - A stale receipt could skip authentication of a different object. Complete reference/lineage/namespace binding, fresh strong HEAD and row-locked save validation prevent that reuse. Missing or unsupported hints trigger full verification. - Mutable local files can change between attempts. Their source locks and fresh captures remain required; a checkpoint never certifies current local bytes. - Checkpoint persistence can contend with other snapshot owners. It has one-second lock and eight-second statement limits; the four classified transient SQL errors retain the normal fully verified path. Other database errors stop the import so lost source ownership cannot be hidden. - No schema, archive format, public API or saved-run identity change. Older snapshots without receipts remain readable and get full verification. Existing data and recovery obligations stay intact. ## Model Used OpenAI Codex, GPT-6.1 Sol (`gpt-6.1-sol`), with reasoning, repository tools and code execution. Context-window size is not exposed by the runtime. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --- .../2026-11-07-online-git-source-fence.md | 17 ++ .../storage-immutable-namespace.test.ts | 24 ++ .../remote-workspace-migration-tree.ts | 25 +- .../services/remote-workspace-migration.ts | 2 +- ...mote-workspace-nested-verification.test.ts | 261 +++++++++++++++++- .../remote-workspace-snapshot-storage.ts | 4 + ...emote-workspace-verification-checkpoint.ts | 83 ++++++ server/src/storage/s3-provider.ts | 4 + server/src/storage/service.ts | 1 + server/src/storage/types.ts | 2 + 10 files changed, 416 insertions(+), 7 deletions(-) create mode 100644 server/src/__tests__/storage-immutable-namespace.test.ts create mode 100644 server/src/services/remote-workspace-verification-checkpoint.ts diff --git a/doc/plans/2026-11-07-online-git-source-fence.md b/doc/plans/2026-11-07-online-git-source-fence.md index 24eb5deec1f..434287d7276 100644 --- a/doc/plans/2026-11-07-online-git-source-fence.md +++ b/doc/plans/2026-11-07-online-git-source-fence.md @@ -100,3 +100,20 @@ serialized remote function, compare archive bytes and digests, remove every original source, restore uncommitted/staged work, and run fsck. No hosted import, hold, cancellation, provider call, or shared-cache Git command is part of these local tests. + +The online migration owner can checkpoint completed immutable nested-snapshot +verification in the child snapshot ledger. This optional versioned proof binds +the complete encrypted reference, child lineage, source operation and a known +S3 endpoint/region/bucket/prefix/style namespace. A new process can reuse it only +after fresh strong-ETag/length, current ledger/source-fence checks and canonical +metadata-only unwrapping of each child key under the current master key. This +key check also runs for in-memory hits; it returns no secret or payload. The proof +is published only after canonical decryption, manifest/file restore and a stable +post-verification HEAD. Missing or unknown proof versions, implicit provider +endpoints and weak or changed validators require full verification. Existing +records need no migration. No plaintext scratch, parent journal CAS, authority, +retention deadline or local/Git source proof is reused. Every resumed invocation +reacquires the source locks and captures current local bytes again. Read-only +verifier callers do not opt into checkpoint publication. +A change that strengthens immutable archive verification must advance the proof +version so earlier verification cannot bypass the new checks. diff --git a/server/src/__tests__/storage-immutable-namespace.test.ts b/server/src/__tests__/storage-immutable-namespace.test.ts new file mode 100644 index 00000000000..32984799766 --- /dev/null +++ b/server/src/__tests__/storage-immutable-namespace.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from "vitest"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; +import { createStorageService } from "../storage/service.js"; + +const config = { bucket: "fixture", region: "us-east-1", endpoint: "https://storage.example.test", prefix: "root", forcePathStyle: true }; +describe("immutable object namespace", () => { + it("propagates the opaque resource identity without credential material", () => { + const provider = createS3StorageProvider(config), service = createStorageService(provider); + expect(provider.immutableObjectNamespace).toMatch(/^[a-f0-9]{64}$/); + expect(service.immutableObjectNamespace).toBe(provider.immutableObjectNamespace); + expect(createS3StorageProvider({ ...config, endpoint: "https://storage.example.test/", prefix: "/root/" }).immutableObjectNamespace) + .toBe(provider.immutableObjectNamespace); + }); + it.each([ + { bucket: "other" }, { region: "us-west-2" }, { endpoint: "https://other.example.test" }, + { prefix: "other" }, { forcePathStyle: false }, + ])("separates a changed resource location %j", change => { + expect(createS3StorageProvider({ ...config, ...change }).immutableObjectNamespace) + .not.toBe(createS3StorageProvider(config).immutableObjectNamespace); + }); + it("does not attest an endpoint selected implicitly by SDK profiles or environment", () => { + expect(createS3StorageProvider({ bucket: "fixture", region: "us-east-1" }).immutableObjectNamespace).toBeUndefined(); + }); +}); diff --git a/server/src/services/remote-workspace-migration-tree.ts b/server/src/services/remote-workspace-migration-tree.ts index 76521ab43aa..b047795db51 100644 --- a/server/src/services/remote-workspace-migration-tree.ts +++ b/server/src/services/remote-workspace-migration-tree.ts @@ -12,7 +12,8 @@ import { localWorkspaceSourcePaths } from "./execution-workspace-source-roots.js import { offlineMigrationExecutionWhere } from "./offline-migration-execution.js"; import type { StorageService } from "../storage/types.js"; import { captureRemoteWorkspaceDirectory, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; -import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, type WorkspaceSnapshotRef } from "./remote-workspace-snapshot-storage.js"; +import { assertWorkspaceSnapshotKeyReadable, getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, type WorkspaceSnapshotRef } from "./remote-workspace-snapshot-storage.js"; +import { nestedWorkspaceVerificationCheckpoint } from "./remote-workspace-verification-checkpoint.js"; type ReadDb = Pick; type Workspace = typeof executionWorkspaces.$inferSelect; @@ -304,8 +305,9 @@ export async function inspectImportedWorkspaceSourcePaths(db: ReadDb, workspace: * the parent archive. The snapshot owner retains the resulting dependencies. */ export function createNestedWorkspaceImportVerifier(input: { db: Db; storage: StorageService; companyId: string; sourceLock: SourceTransaction; - projectIds: readonly string[]; issueIds: readonly string[]; signal?: AbortSignal; + projectIds: readonly string[]; issueIds: readonly string[]; signal?: AbortSignal; persistVerification?: boolean; }) { + const checkpoint = input.persistVerification ? nestedWorkspaceVerificationCheckpoint(input.db, input.storage.immutableObjectNamespace, input.signal) : undefined; const stop = new AbortController(), signal = input.signal ? AbortSignal.any([input.signal, stop.signal]) : stop.signal; const proofs = new Map }>(); let closed = false, current: Promise | undefined; @@ -329,15 +331,28 @@ export function createNestedWorkspaceImportVerifier(input: { && head.etag.replace(/^"|"$/g, "").trim() && !/^W\//i.test(head.etag) ? head.etag : null; const key = createHash("sha256").update(JSON.stringify([reference, children, operationId])).digest("hex"); const previous = proofs.get(key); - if (previous && etag === previous.etag - && await previous.checked) return; + if (previous && etag === previous.etag && await previous.checked) { + await assertWorkspaceSnapshotKeyReadable(reference); + return; + } proofs.delete(key); + if (etag && await checkpoint?.read({ reference, children, operationId, etag })) { + await assertWorkspaceSnapshotKeyReadable(reference); + await assertOwner(); + proofs.set(key, { etag, checked: Promise.resolve(true) }); + return; + } const checked = (async () => { await loadWorkspaceSnapshotArchive({ storage: input.storage, reference, targetPath: archivePath, signal }); await restoreRemoteWorkspaceArchive({ archivePath, targetDir: restored, expectedManifestDigest: reference.manifestDigest, signal }); if (!etag) return false; const after = await objectHead(reference); - return after !== null && after.exists && after.etag === etag && after.contentLength === reference.byteSize; + const stable = after !== null && after.exists && after.etag === etag && after.contentLength === reference.byteSize; + if (stable && checkpoint) { + await assertOwner(); + await checkpoint.save({ reference, children, operationId, etag }); + } + return stable; })(); const proof = etag ? { etag, checked } : null; if (proof) proofs.set(key, proof); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index 4ce163e8c71..e166f001567 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -418,7 +418,7 @@ async function lockMigrationRecovery(lock: MigrationTransaction, plans: Migratio function createMigrationChildVerifier(input: MigrationInput, onlineSource: OnlineWorkspaceSource | undefined, sourceLock: MigrationTransaction) { if (onlineSource?.kind !== "git") return undefined; const verifier = createNestedWorkspaceImportVerifier({ db: input.db, storage: input.storage, companyId: input.companyId, - sourceLock, projectIds: onlineSource.projectIds, issueIds: onlineSource.issueIds, signal: input.signal }); + sourceLock, projectIds: onlineSource.projectIds, issueIds: onlineSource.issueIds, signal: input.signal, persistVerification: true }); let pending = Promise.resolve(); return { prepare(options: Parameters[0]) { diff --git a/server/src/services/remote-workspace-nested-verification.test.ts b/server/src/services/remote-workspace-nested-verification.test.ts index 15ee5a9e5b9..7e3494a7730 100644 --- a/server/src/services/remote-workspace-nested-verification.test.ts +++ b/server/src/services/remote-workspace-nested-verification.test.ts @@ -1,4 +1,5 @@ -import { execFile } from "node:child_process"; +import { execFile, spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; import { createHash, randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import os from "node:os"; @@ -16,6 +17,9 @@ import { createS3StorageProvider } from "../storage/s3-provider.js"; import { createStorageService } from "../storage/service.js"; import type { StorageService } from "../storage/types.js"; import { migrateRemoteWorkspace } from "./remote-workspace-migration.js"; +import { getWorkspaceSnapshot } from "./remote-workspace-snapshot-storage.js"; +import { nestedWorkspaceVerificationCheckpoint } from "./remote-workspace-verification-checkpoint.js"; +import * as checkpointOwner from "./remote-workspace-verification-checkpoint.js"; import * as workspaceArchive from "./remote-workspace-snapshot-archive.js"; import { createNestedWorkspaceImportVerifier } from "./remote-workspace-migration-tree.js"; import { inspectOnlineWorkspaceSource } from "./online-workspace-source.js"; @@ -33,6 +37,7 @@ describe.skipIf(!support.supported)("bounded nested workspace verification", () let database: Awaited>, db: ReturnType, home: string, storage: StorageService, server: Server; let headEtags: Map; let headFailures: Map boolean>; + let storageNamespace: string | undefined; beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("nested-verification-"); db = createDb(database.connectionString, { maxConnections: 8 }); }, 90_000); afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); beforeEach(async () => { @@ -61,6 +66,7 @@ describe.skipIf(!support.supported)("bounded nested workspace verification", () const heads = createStorageService(createS3StorageProvider({ bucket: "fixture", region: "us-east-1", endpoint: `http://127.0.0.1:${address.port}`, forcePathStyle: true })); vi.spyOn(storage, "headObject").mockImplementation(heads.headObject.bind(heads)); + storageNamespace = heads.immutableObjectNamespace; }); afterEach(async () => { vi.restoreAllMocks(); await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots); @@ -113,6 +119,257 @@ describe.skipIf(!support.supported)("bounded nested workspace verification", () } const parent = (f: Awaited>, signal = AbortSignal.timeout(50_000)) => migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: f.parent.id, apply: true, signal }); + function checkpointFailure(phase: "read" | "save", failure: Error, beforeFailure?: () => void) { + const original = checkpointOwner.nestedWorkspaceVerificationCheckpoint; + let faults = 0; + const decorate = (target: Db): Db => new Proxy(target, { get(object, key) { + if (key === (phase === "read" ? "select" : "update")) return () => { faults++; beforeFailure?.(); throw failure; }; + if (key === "transaction") return (callback: (tx: Db) => Promise, options: unknown) => + object.transaction(tx => callback(decorate(tx as unknown as Db)), options as never); + const value = Reflect.get(object, key, object); return typeof value === "function" ? value.bind(object) : value; + } }); + vi.spyOn(checkpointOwner, "nestedWorkspaceVerificationCheckpoint").mockImplementation((source, namespace, signal) => + original(decorate(source), namespace, signal)); + return () => faults; + } + it.each([{ phase: "read", code: "57014" }, { phase: "save", code: "57014" }, + { phase: "save", code: "40P01" }, { phase: "read", code: "40001" }])( + "keeps import resumable after checkpoint $phase $code failure", async ({ phase, code }) => { + const f = await imported(), get = storage.getObject.bind(storage), counts = new Map(); + storage = { ...storage, immutableObjectNamespace: storageNamespace }; + vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { + if (f.keys.has(args[1])) counts.set(args[1], (counts.get(args[1]) ?? 0) + 1); + return get(...args); + }); + if (phase === "read") { + const capture = workspaceArchive.captureRemoteWorkspaceDirectory; + const stop = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory").mockImplementation(async input => { + if (input.cwd === f.root) throw new Error("simulated_container_restart"); + return capture(input); + }); + await expect(parent(f)).rejects.toThrow("simulated_container_restart"); stop.mockRestore(); + expect((await db.select().from(executionWorkspaceSnapshots)).filter(row => row.metadata?.nestedImportVerification)).toHaveLength(f.keys.size); + counts.clear(); + } + const pgError = Object.assign(new Error("injected_checkpoint_database_failure"), { code }); + const failure = code === "40001" ? new Error("checkpoint_query_failed", { cause: pgError }) : pgError; + const faults = checkpointFailure(phase as "read" | "save", failure); + await expect(parent(f)).resolves.toMatchObject({ applied: true }); + expect(faults()).toBeGreaterThan(0); expect([...counts.values()]).toEqual([1, 1]); + if (phase === "save") expect((await db.select().from(executionWorkspaceSnapshots)).filter(row => row.metadata?.nestedImportVerification)).toEqual([]); + const operations = await db.select().from(workspaceOperations).where(eq(workspaceOperations.executionWorkspaceId, f.parent.id)); + expect(operations).toHaveLength(1); expect(operations[0].status).toBe("succeeded"); + expect((await db.select().from(activityLog)).filter(row => row.action === "workspace.remote_storage_migrated" && row.entityId === f.parent.id)).toHaveLength(1); + for (const child of f.leaves) expect(await fs.readFile(path.join(child.cwd!, "untracked.txt"), "utf8")).toBe(`${child.name} untracked work\n`); + }, 60_000); + it.each([{ phase: "read", code: "42P01", message: "checkpoint_table_missing" }, + { phase: "save", code: "08006", message: "checkpoint_connection_lost" }, + { phase: "save", code: null, message: "nested_workspace_snapshot_changed" }, + { phase: "read", code: null, message: "unexpected_checkpoint_state" }])( + "refuses checkpoint $phase failure $code without hiding the cause", async ({ phase, code, message }) => { + const f = await imported(); storage = { ...storage, immutableObjectNamespace: storageNamespace }; + const failure = Object.assign(message === "unexpected_checkpoint_state" ? new TypeError(message) : new Error(message), code ? { code } : {}); + const faults = checkpointFailure(phase as "read" | "save", failure); + await expect(parent(f)).rejects.toBe(failure); expect(faults()).toBeGreaterThan(0); + expect(readRemoteWorkspaceAuthority((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.parent.id)))[0].metadata)).toBeNull(); + expect((await db.select().from(activityLog)).filter(row => row.action === "workspace.remote_storage_migrated" && row.entityId === f.parent.id)).toHaveLength(0); + for (const child of f.leaves) expect(await fs.readFile(path.join(child.cwd!, "untracked.txt"), "utf8")).toBe(`${child.name} untracked work\n`); + }, 60_000); + it("preserves cancellation over a retryable checkpoint database error", async () => { + const f = await imported(), controller = new AbortController(), stopped = new Error("cancelled_during_checkpoint_read"); + storage = { ...storage, immutableObjectNamespace: storageNamespace }; + const reference = (await getWorkspaceSnapshot(db, { companyId: f.companyId, workspaceId: f.snapshots[0].executionWorkspaceId, snapshotId: f.snapshots[0].id }))!; + const original = checkpointOwner.nestedWorkspaceVerificationCheckpoint; + checkpointFailure("read", Object.assign(new Error("query_cancelled"), { code: "57014" }), () => controller.abort(stopped)); + const checkpoint = checkpointOwner.nestedWorkspaceVerificationCheckpoint(db, storageNamespace, controller.signal)!; + await expect(checkpoint.read({ reference, children: [], operationId: f.snapshots[0].sourceOperationId, etag: '"fixture"' })).rejects.toBe(stopped); + expect(original(db, storageNamespace, controller.signal)).toBeDefined(); + }, 60_000); + it("treats checkpoint row contention as a bounded optional cache miss", async () => { + const f = await imported(), snapshot = f.snapshots[0]; + const reference = (await getWorkspaceSnapshot(db, { companyId: f.companyId, workspaceId: snapshot.executionWorkspaceId, snapshotId: snapshot.id }))!; + const head = await storage.headObject(f.companyId, reference.objectKey), checkpoint = nestedWorkspaceVerificationCheckpoint(db, storageNamespace)!; + const binding = { reference, children: [], operationId: snapshot.sourceOperationId, etag: head.etag! }; + await db.transaction(async tx => { + await tx.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, snapshot.id)).for("update"); + const started = performance.now(); + await expect(checkpoint.save(binding)).resolves.toBeUndefined(); + expect(performance.now() - started).toBeLessThan(2500); + expect(await checkpoint.read(binding)).toBe(false); + }); + expect((await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, snapshot.id)))[0]).toEqual(snapshot); + }, 10_000); + it.skipIf(process.platform === "win32").each(["partial", "changed-master-key"] as const)("resumes a killed import with %s checkpoints and the current key authority", async mode => { + const f = await imported(), repoRoot = fileURLToPath(new URL("../../../", import.meta.url)); + const address = server.address(); if (!address || typeof address === "string") throw new Error("fixture_address_missing"); + const keys = [...f.keys], endpoint = `http://127.0.0.1:${address.port}`, temporary = path.join(home, "process-scratch"); + await fs.mkdir(temporary); + const body = String.raw` + const {createDb,closeRegisteredClients}=await import('@paperclipai/db'); + const {createLocalDiskStorageProvider}=await import('./src/storage/local-disk-provider.ts'); + const {createS3StorageProvider}=await import('./src/storage/s3-provider.ts'); + const {createStorageService}=await import('./src/storage/service.ts'); + const {migrateRemoteWorkspace}=await import('./src/services/remote-workspace-migration.ts'); + const input=JSON.parse(process.env.IMPORT_FIXTURE); + const db=createDb(process.env.DATABASE_URL,{maxConnections:8}); + const local=createStorageService(createLocalDiskStorageProvider(input.objects)); + const heads=createStorageService(createS3StorageProvider({bucket:'fixture',region:'us-east-1',endpoint:input.endpoint,forcePathStyle:true})); + const storage={...local,immutableObjectNamespace:heads.immutableObjectNamespace,headObject:heads.headObject.bind(heads), + async getObject(companyId,key,options){ + if(input.pauseKey===key)await new Promise(()=>{}); + if(input.keys.includes(key))process.send?.({read:key}); + return local.getObject(companyId,key,options); + },async putFile(value){ + if(input.pauseParentPut)await new Promise(()=>{}); + return local.putFile(value); + }}; + try{const result=await migrateRemoteWorkspace({db,storage,companyId:input.companyId,workspaceId:input.workspaceId, + apply:true,signal:AbortSignal.timeout(30000)});process.send?.({result});} + catch(error){process.send?.({failed:true,message:error.message});process.exitCode=1;} + finally{await closeRegisteredClients();process.exit(process.exitCode??0);} + `; + const launch = (pauseKey: string | null, pauseParentPut = false, masterKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY) => { + const reads: string[] = [], results: unknown[] = [], failures: string[] = []; + const child = spawn(process.execPath, ["--import", path.join(repoRoot, "cli/node_modules/tsx/dist/loader.mjs"), + "--input-type=module", "--eval", body], { cwd: path.join(repoRoot, "server"), detached: true, + env: { ...process.env, PAPERCLIP_SECRETS_MASTER_KEY: masterKey, TMPDIR: temporary, DATABASE_URL: database.connectionString, IMPORT_FIXTURE: JSON.stringify({ + companyId: f.companyId, workspaceId: f.parent.id, objects: path.join(home, "objects"), endpoint, keys, pauseKey, pauseParentPut }) }, + stdio: ["ignore", "pipe", "pipe", "ipc"] }); + let stderr = ""; child.stderr!.on("data", bytes => { stderr += bytes; }); + child.stdout!.resume(); child.on("message", (message: { read?: string; result?: unknown; failed?: boolean; message?: string }) => { + if (message.read) reads.push(message.read); if (message.result) results.push(message.result); + if (message.failed && message.message) failures.push(message.message); + }); + const closed = new Promise<{ code: number | null; signal: string | null }>((resolve, reject) => { + child.once("error", reject); child.once("close", (code, signal) => resolve({ code, signal })); + }); + return { child, reads, results, failures, closed, error: () => stderr }; + }; + const first = mode === "partial" ? launch(keys[1]) : launch(null, true); + try { + await Promise.race([vi.waitFor(async () => { + const [snapshot] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, f.snapshots[0].id)); + const completed = (await db.select().from(executionWorkspaceSnapshots)).filter(row => row.metadata?.nestedImportVerification); + expect(completed).toHaveLength(mode === "partial" ? 1 : keys.length); expect(snapshot).toBeDefined(); + }, { timeout: 15_000, interval: 25 }), first.closed.then(result => { throw new Error(`import_ended_before_checkpoint:${result.code}:${first.error()}`); })]); + process.kill(-first.child.pid!, "SIGKILL"); + expect(await first.closed).toMatchObject({ signal: "SIGKILL" }); + } finally { + if (first.child.exitCode === null && first.child.signalCode === null) { process.kill(-first.child.pid!, "SIGKILL"); await first.closed; } + } + const [before] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.executionWorkspaceId, f.parent.id)); + expect(before.status).toBe("running"); + const [beforeSnapshot] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, f.parent.id)); + await fs.rm(temporary, { recursive: true, force: true }); await fs.mkdir(temporary); + const snapshotsBefore = await db.select().from(executionWorkspaceSnapshots); + const second = launch(null, false, mode === "changed-master-key" ? "62".repeat(32) : undefined), result = await second.closed; + if (mode === "changed-master-key") { + expect(result, second.error()).toMatchObject({ code: 1, signal: null }); + expect(second.failures).toEqual([expect.stringContaining("Secret decryption failed")]); + expect(second.reads).toEqual([]); expect(second.results).toEqual([]); + expect((await db.select().from(workspaceOperations).where(eq(workspaceOperations.executionWorkspaceId, f.parent.id)))[0]).toEqual(before); + expect(await db.select().from(executionWorkspaceSnapshots)).toEqual(snapshotsBefore); + expect(readRemoteWorkspaceAuthority((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.parent.id)))[0].metadata)).toBeNull(); + expect((await db.select().from(activityLog)).filter(row => row.action === "workspace.remote_storage_migrated" && row.entityId === f.parent.id)).toHaveLength(0); + return; + } + expect(result, second.error()).toMatchObject({ code: 0, signal: null }); + expect(second.results).toEqual([expect.objectContaining({ applied: true, snapshotId: beforeSnapshot.id })]); + expect([...first.reads, ...second.reads].sort()).toEqual([...keys].sort()); + const operations = await db.select().from(workspaceOperations).where(eq(workspaceOperations.executionWorkspaceId, f.parent.id)); + expect(operations).toHaveLength(1); expect(operations[0]).toMatchObject({ id: before.id, status: "succeeded" }); + expect((await db.select().from(activityLog)).filter(row => row.action === "workspace.remote_storage_migrated" && row.entityId === f.parent.id)).toHaveLength(1); + }, 60_000); + it("rechecks the current master key before reusing an in-memory child proof", async () => { + const f = await imported(), get = storage.getObject.bind(storage), counts = new Map(); + vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { + if (f.keys.has(args[1])) counts.set(args[1], (counts.get(args[1]) ?? 0) + 1); + return get(...args); + }); + const capture = workspaceArchive.captureRemoteWorkspaceDirectory; let parentCaptures = 0; + vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory").mockImplementation(async input => { + const result = await capture(input); + if (input.cwd === f.root && ++parentCaptures === 2) vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "62".repeat(32)); + return result; + }); + await expect(parent(f)).rejects.toThrow("Secret decryption failed"); + expect(parentCaptures).toBe(2); expect([...counts.values()]).toEqual([1, 1]); + expect(readRemoteWorkspaceAuthority((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.parent.id)))[0].metadata)).toBeNull(); + expect((await db.select().from(activityLog)).filter(row => row.action === "workspace.remote_storage_migrated" && row.entityId === f.parent.id)).toHaveLength(0); + }, 60_000); + it("resumes completed immutable child checks after losing the process and scratch", async () => { + const f = await imported(), get = storage.getObject.bind(storage), counts = new Map(); + storage = { ...storage, immutableObjectNamespace: storageNamespace }; + vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { + if (f.keys.has(args[1])) counts.set(args[1], (counts.get(args[1]) ?? 0) + 1); + return get(...args); + }); + const capture = workspaceArchive.captureRemoteWorkspaceDirectory; + const interrupted = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory").mockImplementation(async input => { + if (input.cwd === f.root) throw new Error("simulated_container_restart"); + return capture(input); + }); + await expect(parent(f)).rejects.toThrow("simulated_container_restart"); + const [operation] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.executionWorkspaceId, f.parent.id)); + expect(operation.status).toBe("running"); + expect([...counts.values()]).toEqual([1, 1]); + interrupted.mockRestore(); + await expect(parent(f)).resolves.toMatchObject({ applied: true }); + expect([...counts.values()]).toEqual([1, 1]); + await expect(parent(f)).resolves.toMatchObject({ applied: true, replayed: true }); + const operations = await db.select().from(workspaceOperations).where(eq(workspaceOperations.executionWorkspaceId, f.parent.id)); + expect(operations).toHaveLength(1); expect(operations[0].id).toBe(operation.id); expect(operations[0].status).toBe("succeeded"); + const publication = await db.select().from(activityLog); + expect(publication.filter(row => row.action === "workspace.remote_storage_migrated" && row.entityId === f.parent.id)).toHaveLength(1); + }, 60_000); + it.each(["local-source", "etag", "namespace", "reference", "key", "lineage", "proof-version", "weak"] as const)( + "revalidates a changed %s after restarting from an immutable checkpoint", async change => { + const f = await imported(), get = storage.getObject.bind(storage), counts = new Map(), snapshot = f.snapshots[0]; + storage = { ...storage, immutableObjectNamespace: storageNamespace }; + const reads = async (...args: Parameters) => { + if (f.keys.has(args[1])) counts.set(args[1], (counts.get(args[1]) ?? 0) + 1); + return get(...args); + }; + vi.spyOn(storage, "getObject").mockImplementation(reads); + const capture = workspaceArchive.captureRemoteWorkspaceDirectory; + const stop = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory").mockImplementation(async input => { + if (input.cwd === f.root) throw new Error("simulated_container_restart"); + return capture(input); + }); + await expect(parent(f)).rejects.toThrow("simulated_container_restart"); stop.mockRestore(); + const [saved] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, snapshot.id)); + expect(saved.metadata?.nestedImportVerification).toMatchObject({ schema: "paperclip.workspace-snapshot-verification.v1" }); + expect(saved.updatedAt).toEqual(snapshot.updatedAt); expect(saved.deleteAfter).toEqual(snapshot.deleteAfter); + if (change === "local-source") await fs.writeFile(path.join(f.leaves[0].cwd!, "new-work.txt"), "new work survives restart\n"); + if (change === "etag") headEtags.set(snapshot.objectKey!, '"replaced-object-version"'); + if (change === "weak") headEtags.set(snapshot.objectKey!, 'W/"not-a-strong-validator"'); + if (change === "namespace") storage = { ...storage, immutableObjectNamespace: "97".repeat(32) }; + if (change === "reference") await db.update(executionWorkspaceSnapshots).set({ sha256: "00".repeat(32) }) + .where(eq(executionWorkspaceSnapshots.id, snapshot.id)); + if (change === "key") await db.update(executionWorkspaceSnapshots).set({ key: { ...snapshot.key, + tag: Buffer.alloc(16).toString("base64") } }).where(eq(executionWorkspaceSnapshots.id, snapshot.id)); + if (change === "lineage") await db.update(executionWorkspaceSnapshots).set({ metadata: { ...saved.metadata, + importedWorkspaceChildren: [{ workspaceId: randomUUID(), snapshotId: randomUUID(), operationId: randomUUID(), + relativePath: ".paperclip/worktrees/unproved", manifestDigest: "98".repeat(32) }] } }) + .where(eq(executionWorkspaceSnapshots.id, snapshot.id)); + if (change === "proof-version") await db.update(executionWorkspaceSnapshots).set({ metadata: { ...saved.metadata, + nestedImportVerification: { ...(saved.metadata!.nestedImportVerification as Record), schema: "paperclip.workspace-snapshot-verification.v99" } } }) + .where(eq(executionWorkspaceSnapshots.id, snapshot.id)); + if (["local-source", "reference", "key", "lineage"].includes(change)) { + await expect(parent(f)).rejects.toThrow(change === "local-source" ? "nested_workspace_source_changed" + : change === "reference" ? "object_hash_mismatch" : change === "key" ? "Secret decryption failed" : "nested_workspace_coverage_invalid"); + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.parent.id)); + expect(readRemoteWorkspaceAuthority(workspace.metadata)).toBeNull(); + if (change === "local-source") { + expect([...counts.values()]).toEqual([1, 1]); + expect(await fs.readFile(path.join(f.leaves[0].cwd!, "new-work.txt"), "utf8")).toBe("new work survives restart\n"); + } + } else { + await expect(parent(f)).resolves.toMatchObject({ applied: true }); + expect(counts.get(snapshot.objectKey!)).toBe(change === "weak" ? 3 : 2); + if (change === "namespace") expect([...counts.values()]).toEqual([2, 2]); + } + }, 60_000); it("overlaps two independent child readbacks and reuses only authenticated immutable proof", async () => { const f = await imported(), get = storage.getObject.bind(storage), counts = new Map(); const restore = vi.spyOn(workspaceArchive, "restoreRemoteWorkspaceArchive"), capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); @@ -236,6 +493,7 @@ describe.skipIf(!support.supported)("bounded nested workspace verification", () }, 60_000); it("shares immutable proof across declared aliases inside one complete source fence", async () => { const f = await imported(), get = storage.getObject.bind(storage), counts = new Map(); + storage = { ...storage, immutableObjectNamespace: storageNamespace }; const [issue] = await db.insert(issues).values({ companyId: f.companyId, projectId: f.projectId, title: "Parent alias" }).returning(); const [alias] = await db.insert(executionWorkspaces).values({ companyId: f.companyId, projectId: f.projectId, sourceIssueId: issue.id, name: "Parent alias", mode: "shared_workspace", strategyType: "project_primary", providerType: "local_fs", @@ -259,5 +517,6 @@ describe.skipIf(!support.supported)("bounded nested workspace verification", () } finally { await verifier.close(); } }); expect([...counts.values()]).toEqual([1, 1]); + expect((await db.select().from(executionWorkspaceSnapshots)).every(row => !row.metadata?.nestedImportVerification)).toBe(true); }, 60_000); }); diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index 4c7ec81fd22..8044c6aa914 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -360,6 +360,10 @@ async function snapshotKey(ref: WorkspaceSnapshotRef) { return wrapped; } +export async function assertWorkspaceSnapshotKeyReadable(reference: WorkspaceSnapshotRef): Promise { + await snapshotKey(reference); +} + /** Authentication finishes in private scratch before any plaintext is published. */ export async function loadWorkspaceSnapshotArchive(input: { storage: StorageService; reference: WorkspaceSnapshotRef; targetPath: string; signal?: AbortSignal; temporaryParent?: string; diff --git a/server/src/services/remote-workspace-verification-checkpoint.ts b/server/src/services/remote-workspace-verification-checkpoint.ts new file mode 100644 index 00000000000..03dbbee74f4 --- /dev/null +++ b/server/src/services/remote-workspace-verification-checkpoint.ts @@ -0,0 +1,83 @@ +import { createHash } from "node:crypto"; +import { isDeepStrictEqual } from "node:util"; +import { and, eq, sql } from "drizzle-orm"; +import { executionWorkspaceSnapshots, type Db } from "@paperclipai/db"; +import type { ImportedWorkspaceChild } from "./remote-workspace-migration-tree.js"; +import type { WorkspaceSnapshotRef } from "./remote-workspace-snapshot-storage.js"; + +const SCHEMA = "paperclip.workspace-snapshot-verification.v1"; +const KEY = "nestedImportVerification"; +type Snapshot = typeof executionWorkspaceSnapshots.$inferSelect; +type Binding = { reference: WorkspaceSnapshotRef; children: ImportedWorkspaceChild[]; operationId: string; etag: string }; + +function canonical(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]`; + if (value && typeof value === "object") return `{${Object.entries(value).sort(([a], [b]) => a.localeCompare(b)) + .map(([key, item]) => `${JSON.stringify(key)}:${canonical(item)}`).join(",")}}`; + return JSON.stringify(value) ?? "null"; +} +function fingerprint(namespace: string, input: Binding) { + return createHash("sha256").update(canonical([SCHEMA, namespace, input.reference, input.children, input.operationId])).digest("hex"); +} +function matches(row: Snapshot | undefined, input: Binding) { + const ref = input.reference; + return row && ["committed", "superseded"].includes(row.status) && row.sourceOperationId === input.operationId + && row.companyId === ref.companyId && row.executionWorkspaceId === ref.workspaceId && row.id === ref.id + && row.generation === ref.generation && row.objectKey === ref.objectKey && row.sha256 === ref.sha256 + && row.byteSize === ref.byteSize && row.manifestDigest === ref.manifestDigest && isDeepStrictEqual(row.key, ref.key) + && isDeepStrictEqual(row.metadata?.importedWorkspaceChildren ?? [], input.children); +} +function scope(ref: WorkspaceSnapshotRef) { + return and(eq(executionWorkspaceSnapshots.companyId, ref.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, ref.workspaceId), + eq(executionWorkspaceSnapshots.id, ref.id)); +} + +function transientCheckpointFailure(error: unknown): boolean { + if (!error || typeof error !== "object") return false; + const failure = error as { code?: unknown; cause?: { code?: unknown } }; + const code = typeof failure.code === "string" ? failure.code : failure.cause?.code; + return typeof code === "string" && ["55P03", "57014", "40P01", "40001"].includes(code); +} + +export function nestedWorkspaceVerificationCheckpoint(db: Db, namespace: string | undefined, signal?: AbortSignal) { + if (!namespace || !/^[a-f0-9]{64}$/.test(namespace)) return undefined; + return { + async read(input: Binding): Promise { + signal?.throwIfAborted(); + try { return await db.transaction(async tx => { + await tx.execute(sql`set local lock_timeout='1s'`); + await tx.execute(sql`set local statement_timeout='8s'`); + const [row] = await tx.select().from(executionWorkspaceSnapshots).where(scope(input.reference)); + signal?.throwIfAborted(); + if (!matches(row, input)) return false; + const raw = row.metadata?.[KEY]; + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return false; + const proof = raw as Record; + return Object.keys(proof).every(key => ["schema", "namespace", "fingerprint", "etag", "completedAt"].includes(key)) + && proof.schema === SCHEMA && proof.namespace === namespace && proof.fingerprint === fingerprint(namespace, input) + && proof.etag === input.etag && typeof proof.completedAt === "string" && Number.isFinite(Date.parse(proof.completedAt)); + }, { accessMode: "read only" }); } catch (error) { + signal?.throwIfAborted(); + if (!transientCheckpointFailure(error)) throw error; + return false; + } + }, + async save(input: Binding): Promise { + signal?.throwIfAborted(); + try { await db.transaction(async tx => { + await tx.execute(sql`set local lock_timeout='1s'`); + await tx.execute(sql`set local statement_timeout='8s'`); + const [row] = await tx.select().from(executionWorkspaceSnapshots).where(scope(input.reference)).for("update"); + signal?.throwIfAborted(); + if (!matches(row, input)) throw new Error("nested_workspace_snapshot_changed"); + const proof = { schema: SCHEMA, namespace, fingerprint: fingerprint(namespace, input), etag: input.etag, + completedAt: new Date().toISOString() }; + await tx.update(executionWorkspaceSnapshots).set({ metadata: { ...row.metadata, [KEY]: proof } }).where(scope(input.reference)); + signal?.throwIfAborted(); + }); } catch (error) { + signal?.throwIfAborted(); + if (!transientCheckpointFailure(error)) throw error; + } + }, + }; +} diff --git a/server/src/storage/s3-provider.ts b/server/src/storage/s3-provider.ts index a87d279982c..9e553280d7d 100644 --- a/server/src/storage/s3-provider.ts +++ b/server/src/storage/s3-provider.ts @@ -9,6 +9,7 @@ import { } from "@aws-sdk/client-s3"; import { assertSourceFileSize, fileRangeMd5, putS3Multipart, putS3MultipartFile } from "./s3-multipart.js"; import { createReadStream } from "node:fs"; +import { createHash } from "node:crypto"; import { recordS3BodyFailure, withS3FailureDiagnostics } from "./s3-failure.js"; import { addAbortSignal, Readable } from "node:stream"; import { retryS3RequestTimeouts } from "./s3-request-timeout-retry.js"; @@ -121,6 +122,9 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid return { id: "s3", + immutableObjectNamespace: config.endpoint ? createHash("sha256").update(JSON.stringify(["s3", + new URL(config.endpoint).href, region, bucket, prefix, Boolean(config.forcePathStyle), + ])).digest("hex") : undefined, async putObjectConditional(input) { const output = await client.send(new PutObjectCommand({ diff --git a/server/src/storage/service.ts b/server/src/storage/service.ts index b742bef8d66..9203903d0d5 100644 --- a/server/src/storage/service.ts +++ b/server/src/storage/service.ts @@ -107,6 +107,7 @@ function assertPutFileInput(input: PutFileInput): void { export function createStorageService(provider: StorageProvider): StorageService { return { provider: provider.id, + ...(provider.immutableObjectNamespace ? { immutableObjectNamespace: provider.immutableObjectNamespace } : {}), async putFile(input: PutFileInput): Promise { assertPutFileInput(input); diff --git a/server/src/storage/types.ts b/server/src/storage/types.ts index 11cb9449f41..cec0dcdb473 100644 --- a/server/src/storage/types.ts +++ b/server/src/storage/types.ts @@ -61,6 +61,7 @@ export interface HeadObjectResult { export interface StorageProvider { id: StorageProviderId; + readonly immutableObjectNamespace?: string; putObject(input: PutObjectInput): Promise; getObject(input: GetObjectInput): Promise; headObject(input: GetObjectInput): Promise; @@ -100,6 +101,7 @@ export interface PutFileResult { export interface StorageService { provider: StorageProviderId; + readonly immutableObjectNamespace?: string; putFile(input: PutFileInput): Promise; getObject(companyId: string, objectKey: string, options?: Pick): Promise; headObject(companyId: string, objectKey: string, options?: Pick): Promise;