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;