From 1d4f18e023cd3b80e533d35703834c4e84daad70 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 10 Nov 2026 08:28:21 -0600 Subject: [PATCH] perf(workspaces): copy legacy snapshots once Capture files once with one archive body checksum. Trust upload acknowledgements, reuse journaled uploads, and retain imported children by database reference. Remove migration readback, extraction, recapture, and recursive verification caches. Co-Authored-By: Paperclip --- .../2026-11-07-workspace-cohort-capture.md | 46 +- .../remote-workspace-cohort-migration.test.ts | 98 ++-- .../remote-workspace-migration-tree.ts | 131 +---- .../remote-workspace-migration.test.ts | 219 ++++---- .../services/remote-workspace-migration.ts | 66 ++- ...mote-workspace-nested-verification.test.ts | 481 ++---------------- .../remote-workspace-snapshot-archive.test.ts | 45 ++ .../remote-workspace-snapshot-archive.ts | 70 ++- .../remote-workspace-snapshot-storage.ts | 39 +- ...emote-workspace-verification-checkpoint.ts | 83 --- 10 files changed, 438 insertions(+), 840 deletions(-) delete mode 100644 server/src/services/remote-workspace-verification-checkpoint.ts diff --git a/doc/plans/2026-11-07-workspace-cohort-capture.md b/doc/plans/2026-11-07-workspace-cohort-capture.md index ab731ba6031..c674af36576 100644 --- a/doc/plans/2026-11-07-workspace-cohort-capture.md +++ b/doc/plans/2026-11-07-workspace-cohort-capture.md @@ -35,22 +35,25 @@ The batch acquires an independent complete source fence, all component lifecycle locks, and the union of required physical copyback locks. It refuses to borrow an enclosing transaction because a savepoint cannot confirm the outer commit. -Reservation runs serially on the original pool. Capture, encryption, upload, -authenticated readback, restore, and recapture run concurrently on that pool. +Reservation runs serially on the original pool. Each member makes one archive, +checksums the complete archive body while copying its files, encrypts it, and +uploads it. Migration trusts a successful upload. It does not hash each source +file, download the upload, extract a test copy, or capture the source again. +Encryption computes the ciphertext checksum as it writes, without a separate +read of the encrypted file. These member operations run concurrently. Under the complete fence, preflight compares every selected workspace fingerprint with the aggregate source proof's full preimage before the first reservation. Each reservation repeats its member check before writing a journal. Earlier source changes refuse the attempt; a fresh attempt can replan when no journal was reserved. Source changes after a partial reservation remain blocked by the saved aggregate proof. -One nested-child verifier belongs to the complete cohort fence. Its prepare -calls are queued; each call retains the existing two-child bound and completed -authenticated immutable proof. Every member still rechecks current local bytes, -Git metadata, database references and recovery bindings. The verifier closes -only after all member work joins, before the fence can leave. +One nested-child selector belongs to the complete cohort fence. Its prepare +calls are queued. It uses committed snapshot and operation records to retain +already imported children as dependencies. It does not download, restore, or +recapture those children. Publication still checks their database bindings. +The selector closes after all member work joins, before the fence can leave. The caller must budget for the number of logical captures, even though there is -only one Node process. Include up to two nested-child verification workers in -addition to the configured member concurrency. The first hosted qualification +only one Node process. The first hosted qualification uses two same-component leaves without imported children. Eight captures can use much more scratch space, CPU, storage bandwidth, and database capacity than one capture. Start with a two-member canary and measure the actual host before increasing concurrency. @@ -71,9 +74,10 @@ eligibility or snapshot work. Publication repeats that check. The owner does not reset or reopen those journals. A single-workspace import refuses a cohort journal. -After capture, the owner transaction locks and freshly validates every member, -source, journal, recovery binding, and uploaded snapshot before it writes any -workspace authority. It then publishes members serially. The outer commit must +After upload, the owner transaction checks every member's ownership, journal, +recovery binding, and snapshot record before it writes workspace authority. +It does not reconstruct the data to prove the copy. It publishes members +serially. The outer commit must succeed before the batch returns success or publishes live activity events. The originals remain intact. @@ -84,6 +88,24 @@ reservation. A changed or incomplete membership is rejected. Fresh physical source checks still run after a process or volume handover; the durable logical fingerprint does not treat an inode number as lasting authority. +An acknowledged upload is reused directly. If the previous process lost its +upload reply, a journaled upload found by HEAD is reused. Neither case rebuilds +or downloads the archive. An explicit upload failure remains a failure. + +## Archive compatibility + +Migration uses the existing PCWS1 container with a `bodySha256` manifest field +instead of per-file hashes. The manifest digest binds that checksum. Readers +check the complete body during extraction. Existing per-file archives remain +readable. Deploy the new reader before running this migration: older releases +cannot read archive-checksummed snapshots. Ordinary run captures keep their +existing per-file mode. Object loading still authenticates and decrypts the +complete encrypted archive when a real restore is needed. + +This deliberately trades migration readback and repeat source comparisons for +speed. It relies on the existing source/writer fences and the storage upload +acknowledgement. The original source is retained for recovery. + An uncertain commit must first be observed through the canonical inventory and journals. An entirely remote set is an idempotent replay only when its selected snapshots and succeeded journals prove the same complete cohort membership. diff --git a/server/src/services/remote-workspace-cohort-migration.test.ts b/server/src/services/remote-workspace-cohort-migration.test.ts index 702fd15062f..c21e2dbbe9b 100644 --- a/server/src/services/remote-workspace-cohort-migration.test.ts +++ b/server/src/services/remote-workspace-cohort-migration.test.ts @@ -1,5 +1,5 @@ import { execFile } from "node:child_process"; -import { createHash, randomUUID } from "node:crypto"; +import { randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -107,6 +107,8 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => } it("captures two shared-Git leaves concurrently and restores both after outer commit", async () => { const f = await fixture(), held = await recoverable(f), workspaceIds = f.leaves.map(row => row.id), put = storage.putFile.bind(storage); + const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"), restore = vi.spyOn(workspaceArchive, "restoreRemoteWorkspaceArchive"); + const get = vi.spyOn(storage, "getObject"), head = vi.spyOn(storage, "headObject"); let entered = 0, active = 0, maximum = 0, release!: () => void; const barrier = new Promise(resolve => { release = resolve; }), timer = setTimeout(() => release(), 2000); vi.spyOn(storage, "putFile").mockImplementation(async input => { @@ -132,6 +134,9 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => } finally { clearTimeout(timer); } expect(maximum).toBe(2); await expect(run(f)).resolves.toMatchObject({ applied: true, replayed: true }); + expect(capture).toHaveBeenCalledTimes(2); + expect(capture.mock.calls.every(([input]) => input.integrity === "archive")).toBe(true); + expect(restore).not.toHaveBeenCalled(); expect(get).not.toHaveBeenCalled(); expect(head).not.toHaveBeenCalled(); await expect(run(f, { workspaceIds: [f.leaves[0].id] })).rejects.toThrow("workspace_cohort_membership_changed"); expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, held.runId)))[0]).toEqual(held.owner); const archives = []; @@ -151,7 +156,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => expect(await git(targetDir, "log", "-1", "--format=%s")).toBe("Original history"); } }, 60_000); - it("releases completed parent payload before a held peer while preserving the final child mutation check", async () => { + it("releases completed parent payload before a held peer and retains the committed child reference without rechecking its bytes", async () => { const f = await fixture(), parent = f.leaves[0], cwd = path.join(parent.cwd!, ".paperclip", "worktrees", "nested"); await git(f.root, "worktree", "add", "-qb", "nested", cwd); const [childIssue, aliasIssue] = await db.insert(issues).values([ @@ -170,18 +175,22 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => await fs.writeFile(path.join(cwd, "untracked.txt"), "nested dirty bytes\n"); await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: child.id, apply: true, signal: AbortSignal.timeout(50_000) }); + const [childSnapshot] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, child.id)); + const [childWorkspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, child.id)); const readOnly = path.join(parent.cwd!, "read-only"); await fs.mkdir(readOnly); await fs.writeFile(path.join(readOnly, "saved.txt"), "read-only parent payload\n"); await fs.chmod(readOnly, 0o555); const capture = workspaceArchive.captureRemoteWorkspaceDirectory; let firstScratch = "", waiting = false, release!: () => void; const held = new Promise(resolve => { release = resolve; }); - vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory").mockImplementation(async input => { + const captured = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory").mockImplementation(async input => { if (input.cwd === parent.cwd && path.basename(input.archivePath) === "workspace.pcws") { if (!firstScratch) firstScratch = path.dirname(input.archivePath); else { waiting = true; await held; } } return capture(input); }); + const restore = vi.spyOn(workspaceArchive, "restoreRemoteWorkspaceArchive"); + const get = vi.spyOn(storage, "getObject"), head = vi.spyOn(storage, "headObject"); const beforeActivity = await db.select().from(activityLog); const pending = run(f, { workspaceIds: [parent.id, alias.id], concurrency: 1 }).then(value => ({ value }), error => ({ error })); let result: Awaited | undefined; @@ -191,21 +200,30 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => for (const name of ["workspace.pcws", "verified.pcws", "current.pcws", "restored"]) expect.soft(await fs.lstat(path.join(firstScratch, name)).catch(error => { if (error.code === "ENOENT") return null; throw error; })).toBeNull(); const covered = (await fs.readdir(firstScratch)).filter(name => name.startsWith("covered-")); - expect(covered.length).toBeGreaterThan(0); - for (const area of covered) for (const name of ["saved.pcws", "current.pcws", "restored"]) - expect(await fs.lstat(path.join(firstScratch, area, name))).toBeDefined(); + expect(covered).toEqual([]); expect((await db.select().from(executionWorkspaces).where(inArray(executionWorkspaces.id, [parent.id, alias.id]))) .every(row => !readRemoteWorkspaceAuthority(row.metadata))).toBe(true); await fs.writeFile(path.join(cwd, "untracked.txt"), "changed after first payload cleanup\n"); } finally { release(); result = await pending; await fs.chmod(readOnly, 0o755); } - expect(result).toMatchObject({ error: expect.objectContaining({ message: "nested_workspace_source_changed" }) }); + expect(result).toMatchObject({ value: { applied: true } }); expect((await db.select().from(executionWorkspaces).where(inArray(executionWorkspaces.id, [parent.id, alias.id]))) - .every(row => !readRemoteWorkspaceAuthority(row.metadata))).toBe(true); - expect(await db.select().from(activityLog)).toEqual(beforeActivity); + .every(row => readRemoteWorkspaceAuthority(row.metadata))).toBe(true); + expect(await db.select().from(activityLog)).toHaveLength(beforeActivity.length + 2); + expect(captured).toHaveBeenCalledTimes(2); + expect(captured.mock.calls.every(([input]) => input.cwd === parent.cwd && input.integrity === "archive")).toBe(true); + expect(restore).not.toHaveBeenCalled(); expect(get).not.toHaveBeenCalled(); expect(head).not.toHaveBeenCalled(); + expect((await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, childSnapshot.id)))[0]).toEqual(childSnapshot); + expect((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, child.id)))[0]).toEqual(childWorkspace); + for (const member of [parent, alias]) { + const [snapshot] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, member.id)); + expect(snapshot.metadata?.importedWorkspaceChildren).toEqual([expect.objectContaining({ workspaceId: child.id, + snapshotId: childSnapshot.id, manifestDigest: childSnapshot.manifestDigest })]); + expect(await fs.readFile(path.join(parent.cwd!, "saved.txt"), "utf8")).toBe(`${parent.name} dirty tracked work\n`); + } expect(await fs.readFile(path.join(cwd, "untracked.txt"), "utf8")).toBe("changed after first payload cleanup\n"); await expect(fs.lstat(firstScratch)).rejects.toMatchObject({ code: "ENOENT" }); }, 60_000); - it.each(["unchanged", "changed"] as const)("shares the merged child verifier across parallel aliases under the complete cohort fence (%s child bytes)", async mode => { + it.each(["unchanged", "changed"] as const)("retains committed child metadata across parallel aliases without downloads or recaptures (%s child bytes)", async mode => { const f = await fixture(), parent = f.leaves[0], cwd = path.join(parent.cwd!, ".paperclip", "worktrees", "nested"); await git(f.root, "worktree", "add", "-qb", "nested", cwd); const [childIssue, aliasIssue] = await db.insert(issues).values([ @@ -225,36 +243,33 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: child.id, apply: true, signal: AbortSignal.timeout(50_000) }); const [snapshot] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, child.id)); - const get = storage.getObject.bind(storage), head = storage.headObject.bind(storage); - let reads = 0; - vi.spyOn(storage, "headObject").mockImplementation(async (...args) => { - const result = await head(...args); - if (!result.exists) return result; - const bytes = await fs.readFile(path.join(home, "objects", args[1])); - return { ...result, etag: `"${createHash("sha256").update(bytes).digest("hex")}"` }; - }); - vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { - if (args[1] === snapshot.objectKey) reads++; - return get(...args); - }); + const [childWorkspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, child.id)); + const get = vi.spyOn(storage, "getObject"), head = vi.spyOn(storage, "headObject"); const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); + const restore = vi.spyOn(workspaceArchive, "restoreRemoteWorkspaceArchive"); const beforeActivity = await db.select().from(activityLog), put = storage.putFile.bind(storage); let uploads = 0; - if (mode === "changed") vi.spyOn(storage, "putFile").mockImplementation(async input => { + vi.spyOn(storage, "putFile").mockImplementation(async input => { const result = await put(input); - if (++uploads === 2) await fs.writeFile(path.join(cwd, "untracked.txt"), "changed during parent upload\n"); + if (++uploads === 2 && mode === "changed") await fs.writeFile(path.join(cwd, "untracked.txt"), "changed during parent upload\n"); return result; }); - const pending = run(f, { workspaceIds: [parent.id, alias.id] }); - if (mode === "unchanged") await expect(pending).resolves.toMatchObject({ applied: true }); - else { - await expect(pending).rejects.toThrow("nested_workspace_source_changed"); - const selected = await db.select().from(executionWorkspaces).where(inArray(executionWorkspaces.id, [parent.id, alias.id])); - expect(selected.every(row => !readRemoteWorkspaceAuthority(row.metadata))).toBe(true); - expect(await db.select().from(activityLog)).toEqual(beforeActivity); + await expect(run(f, { workspaceIds: [parent.id, alias.id] })).resolves.toMatchObject({ applied: true }); + const selected = await db.select().from(executionWorkspaces).where(inArray(executionWorkspaces.id, [parent.id, alias.id])); + expect(selected.every(row => readRemoteWorkspaceAuthority(row.metadata))).toBe(true); + expect(await db.select().from(activityLog)).toHaveLength(beforeActivity.length + 2); + expect(uploads).toBe(2); + expect(get).not.toHaveBeenCalled(); expect(head).not.toHaveBeenCalled(); expect(restore).not.toHaveBeenCalled(); + expect(capture).toHaveBeenCalledTimes(2); + expect(capture.mock.calls.every(([input]) => input.cwd === parent.cwd && input.integrity === "archive")).toBe(true); + expect((await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, snapshot.id)))[0]).toEqual(snapshot); + expect((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, child.id)))[0]).toEqual(childWorkspace); + for (const member of selected) { + const [saved] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, member.id)); + expect(saved.metadata?.importedWorkspaceChildren).toEqual([expect.objectContaining({ workspaceId: child.id, + snapshotId: snapshot.id, manifestDigest: snapshot.manifestDigest })]); + expect(await fs.readFile(path.join(parent.cwd!, "saved.txt"), "utf8")).toBe(`${parent.name} dirty tracked work\n`); } - expect(reads).toBe(1); - expect(capture.mock.calls.filter(([input]) => input.cwd === cwd && path.basename(input.archivePath) === "current.pcws")).toHaveLength(2); expect(await fs.readFile(path.join(cwd, "untracked.txt"), "utf8")).toBe(mode === "unchanged" ? "nested dirty bytes\n" : "changed during parent upload\n"); }, 60_000); it.each(["foreign", "journal", "finished"])("refuses a changed %s operation before any authority write", async mode => { @@ -417,6 +432,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => const f = await fixture(), selected = await aliasMembers(f, f.parent, 10), workspaceIds = selected.map(row => row.id); for (const child of f.leaves) await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: child.id, apply: true, signal: AbortSignal.timeout(50_000) }); + const childSnapshots = await db.select().from(executionWorkspaceSnapshots); await fs.writeFile(path.join(f.root, "root-untracked.txt"), "parent retained work\n"); const inspect = onlineWorkspaceSource.inspectOnlineWorkspaceSource, transaction = db.transaction.bind(db), put = storage.putFile.bind(storage); let readers = 0, maximumReaders = 0, reservations = 0, maximumReservations = 0, uploads = 0, maximumUploads = 0; @@ -429,6 +445,8 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => try { return await transaction(...args); } finally { reservations--; } }); const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); + const restore = vi.spyOn(workspaceArchive, "restoreRemoteWorkspaceArchive"); + const get = vi.spyOn(storage, "getObject"), head = vi.spyOn(storage, "headObject"); let entered = 0, release!: () => void, interrupt = mode === "resume"; let timer: ReturnType | undefined; const barrier = new Promise(done => { release = done; }); @@ -459,14 +477,24 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () => expect(readers).toBe(0); expect(reservations).toBe(0); expect(uploads).toBe(0); const operations = (await db.select().from(workspaceOperations)).filter(row => workspaceIds.includes(row.executionWorkspaceId!)); expect(operations).toHaveLength(10); expect(operations.every(row => row.status === "succeeded")).toBe(true); - if (mode === "fresh") expect(capture.mock.calls.filter(([input]) => f.leaves.some(child => input.cwd === child.cwd) - && path.basename(input.archivePath) === "current.pcws")).toHaveLength(4); + expect(capture).toHaveBeenCalledTimes(mode === "fresh" ? 10 : 11); + expect(capture.mock.calls.every(([input]) => input.cwd === f.root && input.integrity === "archive")).toBe(true); + expect(get).not.toHaveBeenCalled(); expect(restore).not.toHaveBeenCalled(); + expect(head.mock.calls.filter(([, key]) => childSnapshots.some(snapshot => snapshot.objectKey === key))).toEqual([]); + if (mode === "fresh") expect(head).not.toHaveBeenCalled(); else { expect(operations.map(row => row.id).sort()).toEqual(savedOperations.map(row => row.id).sort()); expect(operations.every(row => JSON.stringify(row.metadata!.cohortMigration) === JSON.stringify(savedOperations[0].metadata!.cohortMigration))).toBe(true); expect((await db.select().from(executionWorkspaceSnapshots)).filter(row => savedSnapshotIds.includes(row.id))).toHaveLength(2); } await expect(run(f, { workspaceIds })).resolves.toMatchObject({ applied: true, replayed: true }); + expect(capture).toHaveBeenCalledTimes(mode === "fresh" ? 10 : 11); + expect((await db.select().from(executionWorkspaceSnapshots)).filter(row => childSnapshots.some(child => child.id === row.id))) + .toEqual(childSnapshots); + for (const child of f.leaves) { + expect(await fs.readFile(path.join(child.cwd!, "saved.txt"), "utf8")).toBe(`${child.name} dirty tracked work\n`); + expect(await fs.readFile(path.join(child.cwd!, "untracked.txt"), "utf8")).toBe(`${child.name} untracked work\n`); + } await fs.rename(f.root, f.root + "-original"); for (const workspace of selected) { const [row] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace.id)); diff --git a/server/src/services/remote-workspace-migration-tree.ts b/server/src/services/remote-workspace-migration-tree.ts index 03e2f4fa19b..6a080775e0c 100644 --- a/server/src/services/remote-workspace-migration-tree.ts +++ b/server/src/services/remote-workspace-migration-tree.ts @@ -1,5 +1,4 @@ import fs from "node:fs/promises"; -import { createHash } from "node:crypto"; import path from "node:path"; import { isDeepStrictEqual } from "node:util"; import { and, asc, eq, gt, inArray, ne, sql } from "drizzle-orm"; @@ -11,9 +10,6 @@ import { readWorkspaceGitSourceRoots } from "./workspace-git-source-roots.js"; 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 { 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; @@ -22,11 +18,6 @@ type NestedVerificationInput = { db: Db; storage: StorageService; workspace: Workspace; scratch: string; signal?: AbortSignal; recordedChildren?: ImportedWorkspaceChild[]; gitSourceRoots?: string[]; }; -type NestedVerification = { - abort(error: unknown): void; - immutable(reference: WorkspaceSnapshotRef, children: ImportedWorkspaceChild[], operationId: string, - archivePath: string, restored: string): Promise; -}; export type ImportedWorkspaceChild = { workspaceId: string; snapshotId: string; operationId: string; relativePath: string; manifestDigest: string; }; @@ -302,66 +293,19 @@ export async function inspectImportedWorkspaceSourcePaths(db: ReadDb, workspace: })); } -/** Runtime-created nested task folders can be omitted only after an exact - * committed import covers their current bytes. Ordinary user folders stay in - * 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; persistVerification?: boolean; + projectIds: readonly string[]; issueIds: readonly string[]; signal?: AbortSignal; }) { - 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; - const objectHead = async (reference: WorkspaceSnapshotRef) => { - try { return await input.storage.headObject(reference.companyId, reference.objectKey, { signal }); } - catch { signal.throwIfAborted(); return null; } - }; const assertOwner = async () => { if (closed) throw new Error("nested_workspace_verifier_closed"); - signal.throwIfAborted(); + input.signal?.throwIfAborted(); await withWorkspaceSourceLocks(input.db, { companyId: input.companyId, projectIds: input.projectIds, issueIds: input.issueIds }, async lock => { if (lock !== input.sourceLock) throw new Error("nested_workspace_source_fence_changed"); await lock.execute(sql`select 1`); }, true); }; - const verification: NestedVerification = { - abort: error => stop.abort(error), - immutable: async (reference, children, operationId, archivePath, restored) => { - const head = await objectHead(reference); - const etag = head?.exists && head.contentLength === reference.byteSize && typeof head.etag === "string" - && 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) { - 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); - 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); - try { if (!await checked && proofs.get(key) === proof) proofs.delete(key); } - catch (error) { if (proofs.get(key) === proof) proofs.delete(key); throw error; } - }, - }; return { assertOwner, prepare(options: Omit): Promise { @@ -371,33 +315,27 @@ export function createNestedWorkspaceImportVerifier(input: { if (options.workspace.companyId !== input.companyId) throw new Error("workspace_source_company_mismatch"); const sourceRoot = rootOf(options.workspace); if (sourceRoot) await assertCurrentWorkspaceSourceHeld(sourceRoot); - const result = await prepareNestedWorkspaceImportsWithVerification({ ...options, db: input.db, storage: input.storage, signal }, verification); + const result = await prepareNestedWorkspaceImports({ ...options, db: input.db, storage: input.storage, signal: input.signal }); await assertOwner(); return result; - })().catch(error => { stop.abort(error); throw error; }); + })(); current = work; return work.finally(() => { current = undefined; }); }, async close() { - closed = true; stop.abort(new Error("nested_workspace_verifier_closed")); + closed = true; if (current) await Promise.allSettled([current]); - proofs.clear(); }, }; } -export function prepareNestedWorkspaceImports(input: NestedVerificationInput): Promise { - return prepareNestedWorkspaceImportsWithVerification(input); -} - -async function prepareNestedWorkspaceImportsWithVerification(input: NestedVerificationInput, verification?: NestedVerification): Promise { +export async function prepareNestedWorkspaceImports(input: NestedVerificationInput): Promise { const sourceRoot = rootOf(input.workspace); if (!sourceRoot || input.recordedChildren && !input.recordedChildren.length) return []; const root = await physicalPath(sourceRoot); const candidates = (await importedWorkspaceCandidates(input.db, input.workspace)).filter(row => !input.recordedChildren || input.recordedChildren.some(child => child.operationId === row.operation.id)); const covered: ImportedWorkspaceChild[] = []; - const selected: typeof candidates = []; for (const { operation, snapshot, sourcePath } of candidates) { input.signal?.throwIfAborted(); const relativePath = path.relative(root, sourcePath).split(path.sep).join("/"); @@ -405,64 +343,11 @@ async function prepareNestedWorkspaceImportsWithVerification(input: NestedVerifi const entry = await fs.lstat(sourcePath).catch((error: NodeJS.ErrnoException) => { if (error.code === "ENOENT") return null; throw error; }); if (!entry) { if (input.recordedChildren) throw new Error("nested_workspace_source_changed"); continue; } if (!entry.isDirectory() || entry.isSymbolicLink()) throw new Error("nested_workspace_source_changed"); - if (!verification) await verifyCoveredWorkspace({ ...input, sourcePath, snapshot, operationId: operation.id, - expectedDigest: operation.metadata?.manifestDigest, depth: 0 }); - else selected.push({ operation, snapshot, sourcePath }); + if (snapshot.manifestDigest !== operation.metadata?.manifestDigest) throw new Error("nested_workspace_snapshot_changed"); + validateImportedWorkspaceChildren(snapshot.metadata?.importedWorkspaceChildren ?? []); covered.push({ workspaceId: snapshot.executionWorkspaceId, snapshotId: snapshot.id, operationId: operation.id, relativePath, manifestDigest: snapshot.manifestDigest! }); } - if (verification) { - let next = 0, failure: unknown; - const worker = async () => { - while (!input.signal?.aborted && next < selected.length) { - const { operation, snapshot, sourcePath } = selected[next++]; - try { await verifyCoveredWorkspace({ ...input, sourcePath, snapshot, operationId: operation.id, - expectedDigest: operation.metadata?.manifestDigest, depth: 0 }, verification); } - catch (error) { failure ??= error; verification.abort(error); } - } - }; - const peers = await Promise.allSettled(Array.from({ length: Math.min(2, selected.length) }, worker)); - for (const peer of peers) if (peer.status === "rejected") failure ??= peer.reason; - if (failure) throw failure; - input.signal?.throwIfAborted(); - } if (input.recordedChildren && !isDeepStrictEqual(covered, input.recordedChildren)) throw new Error("nested_workspace_scope_changed"); return covered; } - -async function verifyCoveredWorkspace(input: { - db: Db; storage: StorageService; sourcePath: string; snapshot: typeof executionWorkspaceSnapshots.$inferSelect; - operationId: string; expectedDigest: unknown; scratch: string; signal?: AbortSignal; depth: number; gitSourceRoots?: string[]; -}, verification?: NestedVerification) { - input.signal?.throwIfAborted(); - if (input.depth > 32) throw new Error("nested_workspace_coverage_invalid"); - const { snapshot } = input; - const reference = await getWorkspaceSnapshot(input.db, { companyId: snapshot.companyId, - workspaceId: snapshot.executionWorkspaceId, snapshotId: snapshot.id }); - if (!reference || reference.manifestDigest !== input.expectedDigest || snapshot.sourceOperationId !== input.operationId) - throw new Error("nested_workspace_snapshot_changed"); - const children = validateImportedWorkspaceChildren(snapshot.metadata?.importedWorkspaceChildren ?? []); - const area = await fs.mkdtemp(path.join(input.scratch, "covered-")); - for (const child of children) { - const [saved] = await input.db.select({ operation: workspaceOperations, snapshot: executionWorkspaceSnapshots }) - .from(executionWorkspaceSnapshots).innerJoin(workspaceOperations, and( - eq(workspaceOperations.companyId, executionWorkspaceSnapshots.companyId), eq(workspaceOperations.id, executionWorkspaceSnapshots.sourceOperationId), - )).where(and(eq(executionWorkspaceSnapshots.companyId, snapshot.companyId), eq(executionWorkspaceSnapshots.id, child.snapshotId), - eq(executionWorkspaceSnapshots.executionWorkspaceId, child.workspaceId), eq(workspaceOperations.id, child.operationId), - eq(workspaceOperations.status, "succeeded"), eq(workspaceOperations.phase, "remote_workspace_migration"), - inArray(executionWorkspaceSnapshots.status, ["committed", "superseded"]))); - const sourcePath = path.join(input.sourcePath, child.relativePath); - if (!saved?.operation.cwd || await physicalPath(saved.operation.cwd) !== sourcePath) throw new Error("nested_workspace_coverage_invalid"); - await verifyCoveredWorkspace({ ...input, sourcePath, snapshot: saved.snapshot, operationId: child.operationId, - expectedDigest: child.manifestDigest, scratch: area, depth: input.depth + 1 }, verification); - } - const archivePath = path.join(area, "saved.pcws"), restored = path.join(area, "restored"); - if (verification) await verification.immutable(reference, children, input.operationId, archivePath, restored); - else { - await loadWorkspaceSnapshotArchive({ storage: input.storage, reference, targetPath: archivePath, signal: input.signal }); - await restoreRemoteWorkspaceArchive({ archivePath, targetDir: restored, expectedManifestDigest: reference.manifestDigest, signal: input.signal }); - } - const current = await captureRemoteWorkspaceDirectory({ cwd: input.sourcePath, archivePath: path.join(area, "current.pcws"), - exclude: children.map(child => child.relativePath), gitSourceRoots: input.gitSourceRoots, signal: input.signal }); - if (current.manifestDigest !== reference.manifestDigest) throw new Error("nested_workspace_source_changed"); -} diff --git a/server/src/services/remote-workspace-migration.test.ts b/server/src/services/remote-workspace-migration.test.ts index cb288ae3160..44314972af8 100644 --- a/server/src/services/remote-workspace-migration.test.ts +++ b/server/src/services/remote-workspace-migration.test.ts @@ -61,6 +61,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { beforeEach(async () => { vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "61".repeat(32)); root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-migration-test-")); storage = createStorageService(createLocalDiskStorageProvider(path.join(root, "objects"))); }); afterEach(async () => { + vi.restoreAllMocks(); await Promise.all(servers.splice(0).map(server => new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())))); vi.unstubAllEnvs(); await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots); await db.delete(workspaceRuntimeServices); await db.delete(executionWorkspaceRuntimeLeases); @@ -98,6 +99,42 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { } const queuedBackends = async () => db.execute(sql`select pid from pg_stat_activity where datname=current_database() and wait_event_type='Lock' and query like 'select pg_advisory_xact_lock%'`); + it.each(["single", "cohort"] as const)("copies once without a readback during %s migration", async kind => { + const f = await queuedFixture(); + const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); + const get = vi.spyOn(storage, "getObject").mockRejectedValue(new Error("migration must not download")); + const put = vi.spyOn(storage, "putFile"); + try { + const migrate = () => kind === "single" ? migrateRemoteWorkspace({ db, storage, ...f, apply: true }) + : migrateRemoteWorkspaceCohort({ db, storage, companyId: f.companyId, workspaceIds: [f.workspaceId], + apply: true, signal: AbortSignal.timeout(10_000) }); + expect(await migrate()).toMatchObject({ applied: true }); + expect(await migrate()).toMatchObject({ applied: true, replayed: true }); + expect(capture).toHaveBeenCalledTimes(1); + expect(capture.mock.calls[0][0]).toMatchObject({ integrity: "archive" }); + expect(put).toHaveBeenCalledTimes(1); expect(get).not.toHaveBeenCalled(); + get.mockRestore(); + const authority = readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)!; + const ref = await getWorkspaceSnapshot(db, { ...f, snapshotId: authority.snapshotId! }); + await fs.rm(f.cwd, { recursive: true }); + const archivePath = path.join(root, "cold.pcws"), targetDir = path.join(root, "cold"); + await loadWorkspaceSnapshotArchive({ storage, reference: ref!, targetPath: archivePath }); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref!.manifestDigest }); + expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe("original saved work\n"); + } finally { capture.mockRestore(); get.mockRestore(); put.mockRestore(); } + }); + it("reuses an uploaded archive after its acknowledgement is lost without another capture or download", async () => { + const f = await queuedFixture(), upload = storage.putFile.bind(storage); + const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); + const get = vi.spyOn(storage, "getObject").mockRejectedValue(new Error("migration must not download")); + const put = vi.spyOn(storage, "putFile").mockImplementation(async input => { await upload(input); throw new Error("lost acknowledgement"); }); + try { + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("lost acknowledgement"); + put.mockImplementation(upload); + expect(await migrateRemoteWorkspace({ db, storage, ...f, apply: true })).toMatchObject({ applied: true }); + expect(capture).toHaveBeenCalledTimes(1); expect(put).toHaveBeenCalledTimes(1); expect(get).not.toHaveBeenCalled(); + } finally { capture.mockRestore(); get.mockRestore(); put.mockRestore(); } + }); it.each(["single", "cohort"] as const)("commits a queued %s import after the finite source writer releases", async kind => { const f = await queuedFixture(), writer = await sourceWriter(f.companyId, [f.issueId]); let settled = false; @@ -572,7 +609,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); }); - it.each(["changed", "corrupt", "late-change", "abort"])("keeps the parent local when nested coverage is unsafe (%s)", async mode => { + it.each(["changed", "corrupt", "late-change", "abort"])("trusts committed nested references while retaining cancellation (%s)", async mode => { const { parent, child } = await nestedFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); await migrateRemoteWorkspace({ db, storage, ...child, apply: true, writersStopped: true }); const authority = readRemoteWorkspaceAuthority((await workspace(child.workspaceId)).metadata)!; @@ -586,8 +623,22 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { if (mode === "abort") controller.abort(); else await fs.writeFile(path.join(child.cwd, "child.txt"), "changed during upload"); } return result; }; } - await expect(migrateRemoteWorkspace({ db, storage, ...parent, apply: true, writersStopped: true, signal: controller.signal })).rejects.toThrow(); - expect(readRemoteWorkspaceAuthority((await workspace(parent.workspaceId)).metadata)).toBeNull(); + const get = vi.spyOn(storage, "getObject"), capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); + const migration = migrateRemoteWorkspace({ db, storage, ...parent, apply: true, writersStopped: true, signal: controller.signal }); + if (mode === "abort") { + await expect(migration).rejects.toThrow(); + expect(readRemoteWorkspaceAuthority((await workspace(parent.workspaceId)).metadata)).toBeNull(); + } else { + await expect(migration).resolves.toMatchObject({ applied: true }); + const parentAuthority = readRemoteWorkspaceAuthority((await workspace(parent.workspaceId)).metadata)!; + const [parentReference] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, parentAuthority.snapshotId!)); + expect(parentReference.metadata?.importedWorkspaceChildren).toEqual([expect.objectContaining({ workspaceId: child.workspaceId, + snapshotId: reference!.id, manifestDigest: reference!.manifestDigest })]); + } + expect(get).not.toHaveBeenCalled(); + expect(capture).toHaveBeenCalledTimes(1); + expect(capture.mock.calls[0][0]).toMatchObject({ cwd: parent.cwd, integrity: "archive" }); + expect(readRemoteWorkspaceAuthority((await workspace(child.workspaceId)).metadata)).toEqual(authority); expect(await fs.readFile(path.join(parent.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); if (mode === "abort") { await migrateRemoteWorkspace({ db, storage, ...parent, apply: true, writersStopped: true }); @@ -606,7 +657,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { expect(await migrateWorkspaceTrees(db, { storage, signal: new AbortController().signal, phase: () => {} })).toBe(0); expect(readRemoteWorkspaceAuthority((await workspace(child.workspaceId)).metadata)).toEqual(childAuthority); }); - it("checks transitive nested coverage before omitting a child tree", async () => { + it("retains transitive committed nested references without recapturing changed child bytes", async () => { const { parent, child } = await nestedFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); const id = randomUUID(), cwd = path.join(child.cwd, ".paperclip", "worktrees", "grandchild"); await fs.mkdir(cwd, { recursive: true }); await fs.writeFile(path.join(cwd, "deep.txt"), "deep work"); @@ -614,10 +665,17 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { await db.insert(executionWorkspaces).values({ id, companyId: parent.companyId, sourceIssueId: issue!.id, name: "Grandchild", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd }); await migrateRemoteWorkspace({ db, storage, companyId: parent.companyId, workspaceId: id, apply: true, writersStopped: true }); + const grandchildAuthority = readRemoteWorkspaceAuthority((await workspace(id)).metadata)!; await migrateRemoteWorkspace({ db, storage, ...child, apply: true, writersStopped: true }); + const childAuthority = readRemoteWorkspaceAuthority((await workspace(child.workspaceId)).metadata)!; await fs.writeFile(path.join(cwd, "deep.txt"), "new deep work"); - await expect(migrateRemoteWorkspace({ db, storage, ...parent, apply: true, writersStopped: true })).rejects.toThrow("nested_workspace_source_changed"); - expect(readRemoteWorkspaceAuthority((await workspace(parent.workspaceId)).metadata)).toBeNull(); + const get = vi.spyOn(storage, "getObject"), capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); + await expect(migrateRemoteWorkspace({ db, storage, ...parent, apply: true, writersStopped: true })).resolves.toMatchObject({ applied: true }); + expect(readRemoteWorkspaceAuthority((await workspace(parent.workspaceId)).metadata)?.snapshotId).toBeTruthy(); + expect(readRemoteWorkspaceAuthority((await workspace(child.workspaceId)).metadata)).toEqual(childAuthority); + expect(readRemoteWorkspaceAuthority((await workspace(id)).metadata)).toEqual(grandchildAuthority); + expect(await fs.readFile(path.join(cwd, "deep.txt"), "utf8")).toBe("new deep work"); + expect(get).not.toHaveBeenCalled(); expect(capture).toHaveBeenCalledTimes(1); }); async function gitArchiveFixture(repoUrl = "https://example.invalid/repository.git", home = root) { const f = await seed(), projectId = randomUUID(), projectWorkspaceId = randomUUID(), branchName = "saved-task"; @@ -748,43 +806,22 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { expect(uploaded).toBe(true); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(f.task); }); - it("retains archived intent payload through the final saved relativePath proof", async () => { - const f = await archivedRepositoryFixture(), put = storage.putFile.bind(storage); - let changed = false, restored = "", finalProof = 0; - storage.putFile = async input => { - const result = await put(input); - if (!changed) { changed = true; await db.update(executionWorkspaceRepositories).set({ relativePath: "untracked.txt" }) - .where(eq(executionWorkspaceRepositories.id, f.repository.id)); } - return result; - }; - const restore = workspaceArchive.restoreRemoteWorkspaceArchive, lstat = fs.lstat.bind(fs); - const restoreSpy = vi.spyOn(workspaceArchive, "restoreRemoteWorkspaceArchive").mockImplementation(async input => { - const result = await restore(input); restored = input.targetDir; return result; - }); - const lstatSpy = vi.spyOn(fs, "lstat").mockImplementation(async (...args) => { - if (restored && String(args[0]) === path.join(restored, "untracked.txt")) { - finalProof++; - for (const name of ["workspace.pcws", "verified.pcws", "current.pcws"]) - expect((await lstat(path.join(path.dirname(restored), name))).isFile()).toBe(true); - expect((await lstat(restored)).isDirectory()).toBe(true); - } - return lstat(...args); - }); - try { - await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("repository_preparation_unresolved"); - expect(changed).toBe(true); expect(finalProof).toBeGreaterThan(0); - expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); - expect((await db.select().from(executionWorkspaceRepositories).where(eq(executionWorkspaceRepositories.id, f.repository.id)))[0]) - .toEqual({ ...f.repository, relativePath: "untracked.txt" }); - expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(f.task); - await expect(fs.lstat(path.dirname(restored))).rejects.toMatchObject({ code: "ENOENT" }); - } finally { lstatSpy.mockRestore(); restoreSpy.mockRestore(); } + it("preserves archived repository intent metadata without restoring an uploaded archive", async () => { + const f = await archivedRepositoryFixture(), restore = vi.spyOn(workspaceArchive, "restoreRemoteWorkspaceArchive"); + const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true }); + expect(restore).not.toHaveBeenCalled(); expect(capture).toHaveBeenCalledTimes(1); + expect(capture.mock.calls[0][0]).toMatchObject({ integrity: "archive" }); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy(); + expect((await db.select().from(executionWorkspaceRepositories).where(eq(executionWorkspaceRepositories.id, f.repository.id)))[0]).toEqual(f.repository); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(f.task); + expect((await workspace(f.workspaceId)).metadata?.gitCacheRestoration).toEqual(f.before.metadata?.gitCacheRestoration); }); it("rechecks the current archived repository intent path at publication", async () => { const f = await archivedRepositoryFixture(), put = storage.putFile.bind(storage); let uploaded = false; storage.putFile = async input => { const saved = await put(input); uploaded = true; await db.update(executionWorkspaceRepositories).set({ relativePath: "untracked.txt" }).where(eq(executionWorkspaceRepositories.id, f.repository.id)); return saved; }; - await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("repository_preparation_unresolved"); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_workspace_migration_source_changed"); expect(uploaded).toBe(true); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(f.task); expect((await workspace(f.workspaceId)).metadata?.gitCacheRestoration).toEqual(f.before.metadata?.gitCacheRestoration); @@ -805,7 +842,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { const pending = await pendingRepository(f); const [repository] = await db.update(executionWorkspaceRepositories).set({ relativePath: "untracked.txt" }).where(eq(executionWorkspaceRepositories.id, pending.id)).returning(); await fs.rm(f.base, { recursive: true }); - await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("repository_preparation_unresolved"); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_workspace_migration_source_changed"); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); expect((await db.select().from(executionWorkspaceSnapshots))[0]).toMatchObject({ id: snapshot.id, status: "uploaded" }); expect((await db.select().from(executionWorkspaceRepositories).where(eq(executionWorkspaceRepositories.id, repository.id)))[0]).toEqual(repository); @@ -1270,11 +1307,11 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { return; } if (state === "missing-base-changed-retry") { - const get = storage.getObject.bind(storage); let expire = true; - storage.getObject = async (...args) => { - const result = await get(...args); - if (expire && args[1].includes("/workspace-snapshots/")) { - expire = false; + const transaction = db.transaction.bind(db), get = vi.spyOn(storage, "getObject"); let expired = false; + const fault = vi.spyOn(db, "transaction").mockImplementation(async (action, config) => { + const result = await transaction(action, config); + if (!expired && result && typeof result === "object" && "status" in result && result.status === "uploaded") { + expired = true; const row = await workspace(f.workspaceId); const recipe = row.metadata!.gitCacheRestoration as Record; await db.update(executionWorkspaces).set({ metadata: { ...row.metadata, gitCacheRestoration: { @@ -1282,8 +1319,11 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { } } }).where(eq(executionWorkspaces.id, f.workspaceId)); } return result; - }; - await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_workspace_migration_publication_changed"); + }); + try { + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_workspace_migration_publication_changed"); + expect(expired).toBe(true); expect(get).not.toHaveBeenCalled(); + } finally { fault.mockRestore(); } expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); expect((await workspace(f.workspaceId)).status).toBe("archived"); expect((await db.select().from(executionWorkspaceSnapshots))[0]?.status).toBe("uploaded"); @@ -1473,7 +1513,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { expect(await db.select().from(workspaceOperations)).toHaveLength(0); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); }); - it("verifies saved files, switches authority once, and retains the original folder", async () => { + it("captures saved files, switches authority once, and retains the original folder", async () => { const f = await seed(); expect(await migrateRemoteWorkspace({ db, storage, ...f, apply: true })).toMatchObject({ applied: true, alreadyMigrated: true }); const row = await workspace(f.workspaceId), authority = readRemoteWorkspaceAuthority(row.metadata); @@ -1485,36 +1525,33 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { expect(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(1); expect(await db.select().from(activityLog)).toHaveLength(1); }); - it("keeps the existing download path for Windows imports", async () => { + it("imports on Windows without downloading the uploaded archive", async () => { const f = await seed(), get = vi.spyOn(storage, "getObject"), platform = vi.spyOn(process, "platform", "get").mockReturnValue("win32"); try { await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true }); - expect(get).toHaveBeenCalledTimes(2); + expect(get).not.toHaveBeenCalled(); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy(); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); } finally { platform.mockRestore(); } }); - it.each([false, true])("removes plaintext verification files with read-only folders after import (source changed: %s)", async sourceChanged => { + it.each([false, true])("removes capture scratch and preserves read-only source folders after import (source changed: %s)", async sourceChanged => { const f = await seed(), protectedDir = path.join(f.cwd, "protected"), scratchRoot = path.join(root, "scratch"); await fs.mkdir(protectedDir); await fs.mkdir(scratchRoot); await fs.writeFile(path.join(protectedDir, "saved.txt"), "read-only saved work\n", { mode: 0o444 }); await fs.chmod(protectedDir, 0o555); vi.stubEnv("TMPDIR", scratchRoot); - const get = storage.getObject.bind(storage); let reads = 0; - if (sourceChanged) storage.getObject = async (...args) => { - if (++reads === 1) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n"); - return get(...args); + const put = storage.putFile.bind(storage), get = vi.spyOn(storage, "getObject"); + let uploads = 0; + storage.putFile = async input => { + const result = await put(input); uploads++; + if (sourceChanged) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n"); + return result; }; try { - const migration = migrateRemoteWorkspace({ db, storage, ...f, apply: true }); - if (sourceChanged) { - await expect(migration).rejects.toThrow("remote_workspace_migration_source_changed"); - expect(reads).toBe(1); - expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); - } else { - await expect(migration).resolves.toMatchObject({ applied: true, alreadyMigrated: true }); - expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy(); - } + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true }); + expect(uploads).toBe(1); expect(get).not.toHaveBeenCalled(); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy(); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe(sourceChanged ? "new local work\n" : "original saved work\n"); expect(await fs.readdir(scratchRoot)).toEqual([]); expect((await fs.stat(protectedDir)).mode & 0o777).toBe(0o555); expect((await fs.stat(path.join(protectedDir, "saved.txt"))).mode & 0o777).toBe(0o444); @@ -1533,14 +1570,22 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { expect(await migrateRemoteWorkspace({ db, storage, ...f, apply: true })).toMatchObject({ applied: true }); const operations = await db.select().from(workspaceOperations); expect(operations).toHaveLength(1); expect(operations[0]?.id).toBe(before.operationId); }); - it("still rejects local source edits after an interrupted upload", async () => { + it("reuses the saved upload after local source edits while preserving the edited original", async () => { const f = await seed(), put = storage.putFile.bind(storage); let fail = true; storage.putFile = async input => { const saved = await put(input); if (fail) { fail = false; throw new Error("upload reply lost"); } return saved; }; await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("upload reply lost"); await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local edits\n"); - await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_workspace_snapshot_operation_content_changed"); - expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"), get = vi.spyOn(storage, "getObject"); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true }); + expect(capture).not.toHaveBeenCalled(); expect(get).not.toHaveBeenCalled(); + const authority = readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)!; + expect(authority.snapshotId).toBeTruthy(); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("new local edits\n"); + const ref = (await getWorkspaceSnapshot(db, { ...f, snapshotId: authority.snapshotId! }))!; + const archivePath = path.join(root, "saved-before-edit.pcws"), targetDir = path.join(root, "saved-before-edit"); + await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: archivePath }); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref.manifestDigest }); + expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe("original saved work\n"); }); async function nestedGitWrapperFixture(relative = ".review/repository") { const original = await seed(); @@ -1800,7 +1845,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { await fs.writeFile(path.join(nested, ".git", "objects", "info", "alternates"), objects + "\n"); return result; }); - await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_workspace_migration_source_changed"); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("online_workspace_source_scope_changed"); const after = await inspectOnlineWorkspaceSource(db, await workspace(f.workspaceId)); expect(after).toMatchObject({ kind: "git", root: f.cwd }); expect(after!.gitSourceFingerprints).not.toEqual(before!.gitSourceFingerprints); @@ -1896,36 +1941,40 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); }); - it("rechecks a service that becomes unresolved during snapshot verification", async () => { - const f = await seed(), get = storage.getObject.bind(storage); let reads = 0; - storage.getObject = async (...args) => { - if (++reads === 1) await db.insert(workspaceRuntimeServices).values({ id: randomUUID(), companyId: f.companyId, + it("rechecks a service that becomes unresolved during upload", async () => { + const f = await seed(), put = storage.putFile.bind(storage); let uploads = 0; + storage.putFile = async input => { + const saved = await put(input); + if (++uploads === 1) await db.insert(workspaceRuntimeServices).values({ id: randomUUID(), companyId: f.companyId, executionWorkspaceId: f.workspaceId, scopeType: "execution_workspace", scopeId: f.workspaceId, serviceName: "Late synthetic preview", status: "running", lifecycle: "shared", provider: "local_process", cwd: f.cwd }); - return get(...args); + return saved; }; await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("workspace_runtime_service_unsettled"); - expect(reads).toBe(1); + expect(uploads).toBe(1); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); }); - it("rechecks a writer that appears while the restored copy is being verified", async () => { - const f = await seed(), get = storage.getObject.bind(storage); let reads = 0; - storage.getObject = async (...args) => { - if (++reads === 1) await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId, + it("rechecks a writer that appears during upload", async () => { + const f = await seed(), put = storage.putFile.bind(storage); let uploads = 0; + storage.putFile = async input => { + const saved = await put(input); + if (++uploads === 1) await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId, phase: "workspace_command", status: "running" }); - return get(...args); + return saved; }; await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("workspace_operation_unresolved"); - expect(reads).toBe(1); + expect(uploads).toBe(1); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); }); - it("preserves a local edit made during verification and refuses activation", async () => { - const f = await seed(), get = storage.getObject.bind(storage); let reads = 0; - storage.getObject = async (...args) => { if (++reads === 1) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n"); return get(...args); }; - await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("source_changed"); - expect(reads).toBe(1); - expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + it("preserves local edits made during upload and activates the saved capture without rereading", async () => { + const f = await seed(), put = storage.putFile.bind(storage), get = vi.spyOn(storage, "getObject"); + const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); let uploads = 0; + storage.putFile = async input => { const saved = await put(input); uploads++; + await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n"); return saved; }; + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true }); + expect(uploads).toBe(1); expect(get).not.toHaveBeenCalled(); expect(capture).toHaveBeenCalledTimes(1); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy(); expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("new local work\n"); }); }); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index 58116e90966..7c5c4dc0a52 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -15,8 +15,8 @@ import { workspaceExportCauses } from "./workspace-recovery-source.js"; import type { StorageService } from "../storage/types.js"; import { persistActivity, publishActivity } from "./activity-log.js"; import { tryAcquireExecutionWorkspaceLifecycleLock, withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js"; -import { captureRemoteWorkspaceDirectory, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; -import { beginWorkspaceSnapshot, loadWorkspaceSnapshotArchive, recoverWorkspaceSnapshotArchive, uploadWorkspaceSnapshot } from "./remote-workspace-snapshot-storage.js"; +import { captureRemoteWorkspaceDirectory } from "./remote-workspace-snapshot-archive.js"; +import { beginWorkspaceSnapshot, recoverWorkspaceSnapshotUpload, uploadWorkspaceSnapshot } from "./remote-workspace-snapshot-storage.js"; import { notifyRemoteWorkspaceSnapshots } from "./remote-workspace-snapshot-retention.js"; import { materializeTaskDirectorySnapshot, removeTaskDirectoryTree } from "./isolated-task-directory-snapshot.js"; import { loadTaskDirectoryRestorePoint, mayBeIsolatedTaskDirectoryWorkspace, readTaskDirectoryRetention, validateTaskDirectoryRestorePointRef } from "./isolated-task-directory-retention.js"; @@ -65,6 +65,8 @@ const gitSourceProof = (source: ArchivedGitMigrationSource | null) => source ? { } : null; type RepositoryRequest = typeof executionWorkspaceRepositories.$inferSelect; +const repositoryIntentBindings = (rows: RepositoryRequest[]) => rows.map(({ id, relativePath }) => ({ id, relativePath })) + .sort((a, b) => a.id.localeCompare(b.id)); function isUnmaterializedRepositoryCandidate(repository: RepositoryRequest) { const unprepared = repository.state === "pending" && !repository.failureCode @@ -411,6 +413,9 @@ async function reserveMigration(input: MigrationInput, onlineSource?: OnlineWork if (canonical(recoverySourcesResumeIdentity(operation.metadata?.legacyRecoverySources ?? [])) !== canonical(recoverySourcesResumeIdentity(current.legacyRecoverySources))) throw new Error("remote_workspace_migration_source_changed"); if (canonical(operation.metadata?.onlineRecoveryHolds ?? []) !== canonical(current.onlineRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed"); + if (Object.hasOwn(operation.metadata ?? {}, "archivedRepositoryIntents") + && canonical(operation.metadata!.archivedRepositoryIntents) !== canonical(repositoryIntentBindings(current.archivedRepositoryIntents))) + throw new Error("remote_workspace_migration_source_changed"); return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds, archivedRepositoryIntents: current.archivedRepositoryIntents }; } let migrationSupersession; @@ -424,6 +429,7 @@ async function reserveMigration(input: MigrationInput, onlineSource?: OnlineWork const [operation] = await tx.insert(workspaceOperations).values({ ...(cohort ? { id: cohort.members.find(member => member.workspaceId === input.workspaceId)!.operationId } : {}), companyId: workspace.companyId, executionWorkspaceId: workspace.id, issueId: workspace.sourceIssueId, cwd: workspace.cwd ?? workspace.providerRef, phase: "remote_workspace_migration", status: "running", metadata: { schema: SCHEMA, sourceFingerprint, source, sourceIssueBinding: current.issueBinding, + archivedRepositoryIntents: repositoryIntentBindings(current.archivedRepositoryIntents), ...(cohort ? { cohortMigration: cohort } : {}), ...(migrationSupersession ? { migrationSupersession } : {}), ...(nativeSessionContinuity ? { nativeSessionContinuity } : {}), @@ -463,7 +469,7 @@ async function lockMigrationRecovery(lock: MigrationTransaction, plans: Migratio function createMigrationChildVerifier(input: MigrationInput, onlineSource: OnlineWorkspaceSource | undefined, sourceLock: MigrationTransaction, cohort = false) { 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, persistVerification: true }); + sourceLock, projectIds: onlineSource.projectIds, issueIds: onlineSource.issueIds, signal: input.signal }); type Options = Parameters[0]; const initial = new Map>(), final = new Map>(); let pending = Promise.resolve(); @@ -512,24 +518,31 @@ async function captureMigration(input: MigrationInput, reserved: ReservedMigrati input.signal?.throwIfAborted(); const retained = reserved.retainedDirectory?.restorePoint; const archivePath = path.join(scratch, "workspace.pcws"); - const recovered = reserved.archivedGitSource ? await recoverWorkspaceSnapshotArchive({ + const scope = object(operation.metadata?.nestedWorkspaceScope); + if (operation.metadata?.nestedWorkspaceScope !== undefined && (scope.version !== 1 || !["pending", "fixed"].includes(String(scope.state)))) + throw new Error("nested_workspace_scope_invalid"); + const recovered = await recoverWorkspaceSnapshotUpload({ db: input.db, storage: input.storage, companyId: input.companyId, workspaceId: input.workspaceId, - snapshotId: snapshot.id, targetPath: archivePath, signal: input.signal, - }) : null; - const sourceRoot = recovered ? null : retained ? await materializeTaskDirectorySnapshot({ + snapshotId: snapshot.id, signal: input.signal, + }); + if (recovered) { + if (scope.state === "pending") throw new Error("nested_workspace_scope_invalid"); + const ref = await uploadWorkspaceSnapshot({ ...input, snapshotId: snapshot.id, archivePath, + manifestDigest: recovered.manifestDigest, verification: "upload" }); + return { operation, ref, nested: scope.state === "fixed" ? validateImportedWorkspaceChildren(scope.children) : [], scratch }; + } + const sourceRoot = retained ? await materializeTaskDirectorySnapshot({ payload: await loadTaskDirectoryRestorePoint({ storage: input.storage, workspace: reserved.workspace, ref: retained, recoverMissingEmpty: true, signal: input.signal }), target: path.join(scratch, "retained-source"), expectedDigest: retained.digest, }) : reserved.archivedGitSource ? await materializeArchivedGitMigrationSource({ db: input.db, storage: input.storage, workspace: reserved.workspace, source: reserved.archivedGitSource, scratch }) : reserved.workspace.cwd ?? reserved.workspace.providerRef!; - const scope = object(operation.metadata?.nestedWorkspaceScope); - if (operation.metadata?.nestedWorkspaceScope !== undefined && (scope.version !== 1 || !["pending", "fixed"].includes(String(scope.state)))) - throw new Error("nested_workspace_scope_invalid"); // Journals predating scoped captures included every child. Preserve that // archive identity even if a later batch has already imported the leaves. const recordedChildren = scope.state === "fixed" ? validateImportedWorkspaceChildren(scope.children) : scope.state === "pending" ? undefined : []; const nested = (input.writersStopped || onlineSource?.kind === "git") && !retained && !reserved.archivedGitSource - ? await (childVerifier?.prepare ?? prepareNestedWorkspaceImports)({ ...input, workspace: reserved.workspace, scratch, recordedChildren, gitSourceRoots: onlineSource?.gitSourceRoots }) : []; + ? await (childVerifier?.prepare ?? prepareNestedWorkspaceImports)({ ...input, workspace: reserved.workspace, scratch, recordedChildren, + gitSourceRoots: onlineSource?.gitSourceRoots }) : []; if (scope.state === "pending") { const [saved] = await input.db.update(workspaceOperations).set({ metadata: { ...operation.metadata, nestedWorkspaceScope: { version: 1, state: "fixed", children: nested } }, updatedAt: new Date() }) @@ -540,34 +553,18 @@ async function captureMigration(input: MigrationInput, reserved: ReservedMigrati } if (sourceRoot) await assertArchivedRepositoryIntents(sourceRoot, reserved.archivedRepositoryIntents); const exclude = nested.map(child => child.relativePath); - const captured = recovered ?? await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath, exclude, requireNoGit: onlineSource?.kind === "ordinary", gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal }); - const restoredArchive = path.join(scratch, "verified.pcws"), restored = path.join(scratch, "restored"); - const verifiedArchivePath = process.platform === "win32" ? undefined : restoredArchive; + const captured = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath, exclude, requireNoGit: onlineSource?.kind === "ordinary", + gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal, integrity: "archive" }); const ref = await uploadWorkspaceSnapshot({ db: input.db, storage: input.storage, snapshotId: snapshot.id, companyId: input.companyId, workspaceId: input.workspaceId, archivePath, manifestDigest: captured.manifestDigest, signal: input.signal, - verifiedArchivePath, + verification: "upload", // Private Git reconstruction can change administrative bytes between // attempts. Its journaled logical recipe is checked at reservation and // again under source locks before authority publication below. ...(reserved.archivedGitSource ? { unpublishedRecapture: { sourceOperationId: operation.id } } : {}), }); - if (!verifiedArchivePath) await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, - targetPath: restoredArchive, signal: input.signal }); - await restoreRemoteWorkspaceArchive({ archivePath: restoredArchive, targetDir: restored, expectedManifestDigest: ref.manifestDigest, signal: input.signal }); - await assertArchivedRepositoryIntents(restored, reserved.archivedRepositoryIntents); - if (!recovered) { - const currentArchive = path.join(scratch, "current.pcws"); - const current = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath: currentArchive, exclude, requireNoGit: onlineSource?.kind === "ordinary", gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal }); - if (current.manifestDigest !== captured.manifestDigest) throw new Error("remote_workspace_migration_source_changed"); - } - if (nested.length && !childVerifier?.deferFinal) { - const verifyScratch = path.join(scratch, "verify-nested"); await fs.mkdir(verifyScratch); - const verified = await (childVerifier?.prepare ?? prepareNestedWorkspaceImports)({ ...input, workspace: reserved.workspace, - scratch: verifyScratch, recordedChildren: nested, gitSourceRoots: onlineSource?.gitSourceRoots }); - if (canonical(verified) !== canonical(nested)) throw new Error("nested_workspace_source_changed"); - } input.signal?.throwIfAborted(); - return { operation, ref, nested, scratch, restored }; + return { operation, ref, nested, scratch }; } catch (error) { await removeTaskDirectoryTree(scratch); throw error; } } type CapturedMigration = Awaited>; @@ -618,7 +615,8 @@ async function validateMigrationPublication(tx: MigrationTransaction, input: Mig if (Boolean(validation.retainedDirectory) !== Boolean(reserved.retainedDirectory)) throw new Error("remote_workspace_migration_source_changed"); if (canonical(gitSourceProof(validation.archivedGitSource)) !== canonical(gitSourceProof(reserved.archivedGitSource))) throw new Error("remote_workspace_migration_source_changed"); if (validation.reasons.length) throw new Error(`remote_workspace_migration_blocked:${validation.reasons.join(",")}`); - await assertArchivedRepositoryIntents(captured.restored, validation.archivedRepositoryIntents); + if (canonical(repositoryIntentBindings(validation.archivedRepositoryIntents)) !== canonical(repositoryIntentBindings(reserved.archivedRepositoryIntents))) + throw new Error("remote_workspace_migration_source_changed"); if (reserved.archivedGitSource) await lockArchivedGitMigrationSources(tx as unknown as Db, workspace, reserved.archivedGitSource); const [uploaded] = await tx.select().from(executionWorkspaceSnapshots).where(and(eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.id, ref.id))).for("update"); @@ -901,9 +899,7 @@ export async function migrateRemoteWorkspaceCohort(input: { const captured = await captureMigration(plan.input, plan.reserved, plan.onlineSource, childVerifier); captures[index] = captured; if (plan.onlineSource && !plan.reserved.archivedRepositoryIntents.length) { - for (const name of ["workspace.pcws", "verified.pcws", "current.pcws"]) - await fs.rm(path.join(captured.scratch, name), { force: true }); - await removeTaskDirectoryTree(captured.restored); + await fs.rm(path.join(captured.scratch, "workspace.pcws"), { force: true }); } } catch (error) { failure ??= error; stop.abort(error); } diff --git a/server/src/services/remote-workspace-nested-verification.test.ts b/server/src/services/remote-workspace-nested-verification.test.ts index 809b42f3f3e..0e47f7cb3d6 100644 --- a/server/src/services/remote-workspace-nested-verification.test.ts +++ b/server/src/services/remote-workspace-nested-verification.test.ts @@ -1,77 +1,41 @@ -import { execFile, spawn } from "node:child_process"; -import { fileURLToPath } from "node:url"; -import { createHash, randomUUID } from "node:crypto"; +import { execFile } from "node:child_process"; +import { randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { createServer, type Server } from "node:http"; import { promisify } from "node:util"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { activityLog, companies, createDb, executionWorkspaceSnapshots, executionWorkspaces, issues, projects, workspaceOperations, type Db } from "@paperclipai/db"; +import { activityLog, companies, createDb, executionWorkspaceSnapshots, executionWorkspaces, issues, projects, workspaceOperations } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { resolveManagedProjectWorkspaceDir } from "../home-paths.js"; import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; -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"; -import { tryAcquireExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js"; -import { lockOnlineLegacyWorkspaceRecoverySources } from "./online-legacy-workspace-recovery.js"; -import { IsolatedRepositorySourceBusy, withWorkspaceSourceLocks } from "./workspace-source-lock.js"; +import { withWorkspaceSourceLocks } from "./workspace-source-lock.js"; const support = await getEmbeddedPostgresTestSupport(), execute = promisify(execFile); const git = async (cwd: string, ...args: string[]) => (await execute("git", ["-c", `core.hooksPath=${os.devNull}`, "-C", cwd, ...args], { env: { ...process.env, GIT_CONFIG_NOSYSTEM: "1", GIT_CONFIG_GLOBAL: os.devNull, GIT_TERMINAL_PROMPT: "0", - GIT_AUTHOR_NAME: "Nested verification", GIT_AUTHOR_EMAIL: "fixture@example.test", GIT_COMMITTER_NAME: "Nested verification", GIT_COMMITTER_EMAIL: "fixture@example.test" }, + GIT_AUTHOR_NAME: "Nested import", GIT_AUTHOR_EMAIL: "fixture@example.test", GIT_COMMITTER_NAME: "Nested import", GIT_COMMITTER_EMAIL: "fixture@example.test" }, })).stdout.trim(); -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; +describe.skipIf(!support.supported)("nested workspace migration references", () => { + let database: Awaited>, db: ReturnType, home: string, storage: StorageService; 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 () => { home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-nested-verification-"))); vi.stubEnv("PAPERCLIP_HOME", home); vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "61".repeat(32)); - vi.stubEnv("AWS_ACCESS_KEY_ID", "fixture"); vi.stubEnv("AWS_SECRET_ACCESS_KEY", "fixture"); - vi.stubEnv("AWS_SESSION_TOKEN", "fixture"); vi.stubEnv("AWS_EC2_METADATA_DISABLED", "true"); storage = createStorageService(createLocalDiskStorageProvider(path.join(home, "objects"))); - headEtags = new Map(); - headFailures = new Map(); - server = createServer((request, response) => { - void (async () => { - if (request.method !== "HEAD") { response.writeHead(405).end(); return; } - const key = decodeURIComponent(new URL(request.url!, "http://localhost").pathname).slice("/fixture/".length); - const root = path.join(home, "objects"), file = path.resolve(root, key); - if (!file.startsWith(root + path.sep)) { response.writeHead(400).end(); return; } - const bytes = await fs.readFile(file).catch(() => null); - if (!bytes) { response.writeHead(404).end(); return; } - if (headFailures.get(key)?.()) { response.writeHead(503).end(); return; } - const etag = headEtags.has(key) ? headEtags.get(key) : `"${createHash("sha256").update(bytes).digest("hex")}"`; - response.writeHead(200, { ...(etag == null ? {} : { ETag: etag }), "Content-Length": bytes.length }).end(); - })().catch(() => { response.writeHead(500).end(); }); - }); - await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); - const address = server.address(); if (!address || typeof address === "string") throw new Error("fixture_address_missing"); - 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); await db.delete(issues); await db.delete(executionWorkspaces); await db.delete(projects); await db.delete(companies); - server?.closeAllConnections(); await new Promise(resolve => server?.close(() => resolve())); vi.unstubAllEnvs(); await fs.rm(home, { recursive: true, force: true }); }); async function fixture() { @@ -117,376 +81,45 @@ describe.skipIf(!support.supported)("bounded nested workspace verification", () const snapshots = await db.select().from(executionWorkspaceSnapshots); return { ...f, snapshots, descendants, keys: new Set(snapshots.map(snapshot => snapshot.objectKey!)) }; } - 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); - }}; - process.send?.({ready:true}); - 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[] = []; - let ready = false; - 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; ready?: boolean }) => { - if (message.ready) ready = true; - 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, ready: () => ready, error: () => stderr }; - }; - const first = mode === "partial" ? launch(keys[1]) : launch(null, true); - try { - const prematureExit = first.closed.then(result => { throw new Error(`import_ended_before_checkpoint:${result.code}:${first.error()}`); }); - await Promise.race([vi.waitFor(() => { - expect(first.ready(), `import_startup_not_ready:${first.error()}`).toBe(true); - }, { timeout: 30_000, interval: 25 }), prematureExit]); - 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 }), prematureExit]); - 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"); - let pending = 0, maximum = 0, entered = 0, release!: () => void; - const both = new Promise(resolve => { release = resolve; }), timer = setTimeout(release, 2000); - vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { - if (!f.keys.has(args[1])) return get(...args); - counts.set(args[1], (counts.get(args[1]) ?? 0) + 1); maximum = Math.max(maximum, ++pending); - if (++entered === 2) release(); - try { await both; return await get(...args); } finally { pending--; } - }); - try { await expect(parent(f)).resolves.toMatchObject({ applied: true }); } finally { clearTimeout(timer); } - expect.soft(maximum).toBe(2); expect.soft([...counts.values()]).toEqual([1, 1]); - expect.soft(restore.mock.calls.filter(([args]) => path.basename(args.archivePath) === "saved.pcws")).toHaveLength(2); - expect(capture.mock.calls.filter(([args]) => args.archivePath.includes("covered-") && path.basename(args.archivePath) === "current.pcws")).toHaveLength(4); - expect(readRemoteWorkspaceAuthority((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.parent.id)))[0].metadata)).not.toBeNull(); - 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("bounds both ancestor branches and their grandchildren without recursive permit deadlock", async () => { - const f = await imported(true), get = storage.getObject.bind(storage), counts = new Map(); - let active = 0, maximum = 0; - vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { - if (!f.keys.has(args[1])) return get(...args); - counts.set(args[1], (counts.get(args[1]) ?? 0) + 1); maximum = Math.max(maximum, ++active); - try { const value = await get(...args); value.stream.once("close", () => { active--; }); return value; } - catch (error) { active--; throw error; } - }); - await expect(parent(f)).resolves.toMatchObject({ applied: true }); - expect(maximum).toBeLessThanOrEqual(2); expect([...counts.values()]).toEqual([1, 1, 1, 1]); expect(active).toBe(0); - for (const child of f.descendants) expect(await fs.readFile(path.join(child.cwd!, "untracked.txt"), "utf8")).toBe(child.name); - }, 60_000); - it.each([null, '""', 'W/"weak"'])("uses full authenticated verification when the provider ETag is %s", async etag => { - const f = await imported(), get = storage.getObject.bind(storage), counts = new Map(); - for (const key of f.keys) headEtags.set(key, etag); - 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); - }); - await expect(parent(f)).resolves.toMatchObject({ applied: true }); expect([...counts.values()]).toEqual([2, 2]); - }, 60_000); - it.each(["before", "after"] as const)("keeps full verification available when HEAD fails %s the readback", async phase => { - const f = await imported(), get = storage.getObject.bind(storage), key = f.snapshots[0].objectKey!; let reads = 0; - headFailures.set(key, () => reads === (phase === "before" ? 0 : 1)); - vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { - if (args[1] === key) reads++; - return get(...args); - }); - await expect(parent(f)).resolves.toMatchObject({ applied: true }); expect(reads).toBe(2); - }, 60_000); - it("preserves caller cancellation during a failed HEAD without starting payload readback", async () => { - const f = await imported(), get = storage.getObject.bind(storage), controller = new AbortController(); let reads = 0; - for (const key of f.keys) headFailures.set(key, () => { controller.abort(new Error("caller_stopped_during_head")); return true; }); - vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { - if (f.keys.has(args[1])) reads++; - return get(...args); - }); - await expect(parent(f, controller.signal)).rejects.toThrow(); expect(reads).toBe(0); - expect(readRemoteWorkspaceAuthority((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.parent.id)))[0].metadata)).toBeNull(); - }, 60_000); - it.each(["etag", "reference", "ciphertext", "local-source"] as const)("rechecks a changed %s after the parent capture", async change => { - const f = await imported(), put = storage.putFile.bind(storage), get = storage.getObject.bind(storage), counts = new Map(); - const snapshot = f.snapshots[0]; - 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 parent = (f: Awaited>) => + migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: f.parent.id, apply: true, signal: AbortSignal.timeout(20_000) }); + it("retains transitive committed children without downloading, restoring, or recapturing them", async () => { + const f = await imported(true), get = vi.spyOn(storage, "getObject"), head = vi.spyOn(storage, "headObject"); + const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"), restore = vi.spyOn(workspaceArchive, "restoreRemoteWorkspaceArchive"); + expect(await parent(f)).toMatchObject({ applied: true }); + expect(get).not.toHaveBeenCalled(); expect(head).not.toHaveBeenCalled(); expect(restore).not.toHaveBeenCalled(); + expect(capture).toHaveBeenCalledTimes(1); + const [snapshot] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, f.parent.id)); + expect(snapshot.metadata?.importedWorkspaceChildren).toHaveLength(2); + expect((snapshot.metadata!.importedWorkspaceChildren as Array<{ workspaceId: string }>).map(child => child.workspaceId).sort()) + .toEqual(f.leaves.map(child => child.id).sort()); + expect((await db.select().from(executionWorkspaceSnapshots)).every(row => !row.metadata?.nestedImportVerification)).toBe(true); + for (const descendant of f.descendants) expect(await fs.readFile(path.join(descendant.cwd!, "untracked.txt"), "utf8")).toBe(descendant.name); + }); + it.each(["local edit", "missing object"])("trusts a committed child's recorded upload after a %s", async mode => { + const f = await imported(); + if (mode === "local edit") await fs.writeFile(path.join(f.leaves[0].cwd!, "untracked.txt"), "later local bytes\n"); + else await storage.deleteObject(f.companyId, f.snapshots[0].objectKey!); + const get = vi.spyOn(storage, "getObject"), head = vi.spyOn(storage, "headObject"); + expect(await parent(f)).toMatchObject({ applied: true }); + expect(get).not.toHaveBeenCalled(); expect(head).not.toHaveBeenCalled(); + const [snapshot] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, f.parent.id)); + expect(snapshot.metadata?.importedWorkspaceChildren).toHaveLength(2); + if (mode === "local edit") expect(await fs.readFile(path.join(f.leaves[0].cwd!, "untracked.txt"), "utf8")).toBe("later local bytes\n"); + }); + it("still refuses a child snapshot binding changed during the parent upload", async () => { + const f = await imported(), put = storage.putFile.bind(storage); vi.spyOn(storage, "putFile").mockImplementation(async input => { const result = await put(input); - if (change === "etag") headEtags.set(snapshot.objectKey!, '"replacement-version"'); - if (change === "reference") await db.update(executionWorkspaceSnapshots).set({ sha256: "00".repeat(32) }).where(eq(executionWorkspaceSnapshots.id, snapshot.id)); - if (change === "ciphertext") { - const file = path.join(home, "objects", snapshot.objectKey!), bytes = await fs.readFile(file); bytes[0] ^= 1; await fs.writeFile(file, bytes); - } - if (change === "local-source") await fs.writeFile(path.join(f.leaves[0].cwd!, "new-work.txt"), "newer work must remain local\n"); + await db.update(executionWorkspaceSnapshots).set({ manifestDigest: "f".repeat(64) }).where(eq(executionWorkspaceSnapshots.id, f.snapshots[0].id)); return result; }); - if (change === "etag") { - await expect(parent(f)).resolves.toMatchObject({ applied: true }); expect(counts.get(snapshot.objectKey!)).toBe(2); - } else { - await expect(parent(f)).rejects.toThrow(change === "local-source" ? "nested_workspace_source_changed" : "object_hash_mismatch"); - expect(readRemoteWorkspaceAuthority((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.parent.id)))[0].metadata)).toBeNull(); - if (change === "local-source") expect([...counts.values()]).toEqual([1, 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(["failure", "cancel"] as const)("joins active grandchild peers and retains the source fence on %s", async mode => { - const f = await imported(true), get = storage.getObject.bind(storage), controller = new AbortController(); - let entered = 0, announce!: () => void, release!: () => void; - const both = new Promise(resolve => { announce = resolve; }), peer = new Promise(resolve => { release = resolve; }); - vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { - if (!f.keys.has(args[1])) return get(...args); - const index = ++entered; - if (index === 2) { if (mode === "cancel") controller.abort(new Error("caller_stop")); announce(); } - await both; - if (index === 1 && mode === "failure") throw new Error("child_readback_failed"); - await peer; throw new Error("peer_finished_after_abort"); - }); - let settled = false; - const pending = parent(f, AbortSignal.any([controller.signal, AbortSignal.timeout(50_000)])) - .then(value => ({ value }), error => ({ error })).finally(() => { settled = true; }); - const other = createDb(database.connectionString, { maxConnections: 1 }); - try { - await both; await new Promise(resolve => setTimeout(resolve, 25)); expect(settled).toBe(false); expect(entered).toBe(2); - await expect(withWorkspaceSourceLocks(other, { companyId: f.companyId, projectIds: [f.projectId] }, async () => true, true)) - .rejects.toBeInstanceOf(IsolatedRepositorySourceBusy); - } finally { release(); await other.$client.end({ timeout: 1 }); } - expect(await pending).toHaveProperty("error"); - expect(readRemoteWorkspaceAuthority((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.parent.id)))[0].metadata)).toBeNull(); - }, 60_000); - it("refuses a verifier inherited from a different actual source-owner transaction", async () => { + await expect(parent(f)).rejects.toThrow("online_workspace_source_scope_changed"); + const [row] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.parent.id)); + expect(readRemoteWorkspaceAuthority(row.metadata)).toBeNull(); + expect(await fs.readFile(path.join(f.root, "saved.txt"), "utf8")).toBe("committed base\n"); + }); + it("refuses a selector inherited from another source-owner transaction", async () => { const f = await imported(); let verifier!: ReturnType; await withWorkspaceSourceLocks(db, { companyId: f.companyId, projectIds: [f.projectId] }, async sourceLock => { verifier = createNestedWorkspaceImportVerifier({ db, storage, companyId: f.companyId, projectIds: [f.projectId], issueIds: [], sourceLock }); @@ -497,33 +130,5 @@ describe.skipIf(!support.supported)("bounded nested workspace verification", () }); } finally { await verifier.close(); } await expect(verifier.prepare({ workspace: f.parent, scratch: home, recordedChildren: [] })).rejects.toThrow("nested_workspace_verifier_unavailable"); - }, 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", - cwd: f.root, providerRef: f.root, repoUrl: f.parent.repoUrl, branchName: "main" }).returning(); - const source = await inspectOnlineWorkspaceSource(db, f.parent); expect(source).not.toBeNull(); - 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); - }); - await withWorkspaceSourceLocks(db, { companyId: f.companyId, projectIds: source!.projectIds, issueIds: source!.issueIds }, async sourceLock => { - for (const workspaceId of source!.workspaceIds) expect(await tryAcquireExecutionWorkspaceLifecycleLock(sourceLock, workspaceId)).toBe(true); - expect(await lockOnlineLegacyWorkspaceRecoverySources(sourceLock as unknown as Db, { companyId: f.companyId, - sources: source!.companions.flatMap(proof => proof.sources), holds: source!.companions.flatMap(proof => proof.holds) })).toBe(true); - const verifier = createNestedWorkspaceImportVerifier({ db, storage, companyId: f.companyId, sourceLock, - projectIds: source!.projectIds, issueIds: source!.issueIds }); - try { - const first = path.join(home, "first-alias"), second = path.join(home, "second-alias"); await fs.mkdir(first); await fs.mkdir(second); - const children = await verifier.prepare({ workspace: f.parent, scratch: first, gitSourceRoots: source!.gitSourceRoots }); - expect(await verifier.prepare({ workspace: alias, scratch: second, recordedChildren: children, - gitSourceRoots: source!.gitSourceRoots })).toEqual(children); - } 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-archive.test.ts b/server/src/services/remote-workspace-snapshot-archive.test.ts index 3401d7e1aa7..ae9e63fd47e 100644 --- a/server/src/services/remote-workspace-snapshot-archive.test.ts +++ b/server/src/services/remote-workspace-snapshot-archive.test.ts @@ -71,6 +71,51 @@ describe("portable remote workspace archives", () => { const promise = new Promise(done => { resolve = done; }); return { promise, resolve }; } + it("copies each source file once and restores an archive with one body checksum", async () => { + const cwd = await fs.realpath(await source()), file = path.join(cwd, "saved.txt"); + await fs.writeFile(file, "saved work\n"); + await fs.writeFile(path.join(cwd, "empty.txt"), ""); + await fs.chmod(file, 0o750); + await fs.symlink("saved.txt", path.join(cwd, "link")); + const opens: string[] = []; + archiveIo.afterOpen = async target => { if (target.startsWith(cwd + path.sep)) opens.push(target); }; + const archivePath = path.join(root, "once.pcws"); + const snapshot = { archivePath, ...await captureRemoteWorkspaceDirectory({ cwd, archivePath, integrity: "archive" }) }; + expect(opens.sort()).toEqual([path.join(cwd, "empty.txt"), file].sort()); + const bytes = await fs.readFile(archivePath), manifestSize = bytes.readUInt32BE(6); + const manifest = JSON.parse(bytes.subarray(10, 10 + manifestSize).toString()); + expect(manifest.bodySha256).toBe(createHash("sha256").update(bytes.subarray(10 + manifestSize)).digest("hex")); + expect(manifest.entries.filter((entry: { type: string }) => entry.type === "file") + .every((entry: { sha256?: string }) => entry.sha256 === undefined)).toBe(true); + expect(snapshot.sha256).toBe(createHash("sha256").update(bytes).digest("hex")); + await fs.rm(cwd, { recursive: true }); + const restored = await restore(snapshot); + expect(await fs.readFile(path.join(restored, "saved.txt"), "utf8")).toBe("saved work\n"); + expect((await fs.stat(path.join(restored, "saved.txt"))).mode & 0o777).toBe(0o750); + expect(await fs.readlink(path.join(restored, "link"))).toBe("saved.txt"); + bytes[bytes.length - 1] ^= 1; + const corrupt = path.join(root, "corrupt.pcws"); await fs.writeFile(corrupt, bytes); + await expect(restoreRemoteWorkspaceArchive({ archivePath: corrupt, targetDir: path.join(root, "corrupt"), + expectedManifestDigest: snapshot.manifestDigest })).rejects.toThrow("archive_hash_mismatch"); + await expect(fs.access(path.join(root, "corrupt"))).rejects.toMatchObject({ code: "ENOENT" }); + }); + it("resumes staged and unstaged Git work from one archive-checksummed capture", async () => { + const base = await source(); await git(base, "init", "-q", "-b", "main"); + await fs.writeFile(path.join(base, "work.txt"), "committed\n"); await git(base, "add", "."); await git(base, "commit", "-qm", "base"); + const cwd = path.join(root, "linked"); await git(base, "worktree", "add", "-qb", "selected", cwd); + await fs.writeFile(path.join(cwd, "work.txt"), "staged\n"); await git(cwd, "add", "."); + await fs.writeFile(path.join(cwd, "work.txt"), "unstaged\n"); + await fs.writeFile(path.join(cwd, "new.txt"), "untracked\n"); + const archivePath = path.join(root, "git-once.pcws"); + const snapshot = { archivePath, ...await captureRemoteWorkspaceDirectory({ cwd, archivePath, + integrity: "archive", gitSourceRoots: [await fs.realpath(root)] }) }; + await fs.rm(base, { recursive: true }); await fs.rm(cwd, { recursive: true }); + const restored = await restore(snapshot); + expect(await git(restored, "show", ":work.txt")).toBe("staged"); + expect(await fs.readFile(path.join(restored, "work.txt"), "utf8")).toBe("unstaged\n"); + expect(await fs.readFile(path.join(restored, "new.txt"), "utf8")).toBe("untracked\n"); + await git(restored, "fsck", "--connectivity-only", "--no-dangling"); + }); it.each([".", "nested"])("requires a shared Git fence before reading included %s Git storage", async location => { const cwd = await source(), repository = path.join(cwd, location); await fs.mkdir(repository, { recursive: true }); diff --git a/server/src/services/remote-workspace-snapshot-archive.ts b/server/src/services/remote-workspace-snapshot-archive.ts index 3e729aed92a..aab6f5db7ce 100644 --- a/server/src/services/remote-workspace-snapshot-archive.ts +++ b/server/src/services/remote-workspace-snapshot-archive.ts @@ -17,6 +17,7 @@ export async function workspaceArchive(input: { signal?: AbortSignal; maxEntries?: number; maxManifestBytes?: number; + integrity?: "files" | "archive"; }): Promise<{ schema: "paperclip.remote-workspace-archive.v1"; manifestDigest: string; byteSize: number; sha256: string; entries: number }> { const fs = await import("node:fs/promises"); const { createReadStream } = await import("node:fs"); @@ -26,6 +27,7 @@ export async function workspaceArchive(input: { const child = await import("node:child_process"); const schema = "paperclip.remote-workspace-archive.v1" as const; const magic = Buffer.from("PCWS1\n"); + const archiveIntegrity = input.integrity === "archive"; type Entry = { path: string; type: "directory" | "file" | "symlink"; mode?: number; size?: number; sha256?: string; target?: string }; const fail = (reason: string): never => { throw new Error(`remote_workspace_archive_${reason}`); }; const check = () => input.signal?.throwIfAborted(); @@ -248,8 +250,20 @@ export async function workspaceArchive(input: { || await git(repo, ["ls-files", "--unmerged", "-z"])) fail("git_operation_in_progress: finish or abort the Git operation before saving the workspace"); return { objects: [{ object, type: "commit" }], metadata: [{ name: "REBASE_HEAD", sha256: entry.sha256 }] }; }; - const readStableFile = async (source: string, consume?: (bytes: Buffer) => Promise) => { + const readStableFile = async (source: string, consume?: (bytes: Buffer) => Promise, hashContents = true) => { check(); + if (!hashContents) { + const handle = await fs.open(source, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW | fs.constants.O_NONBLOCK); + try { + const stat = await handle.stat(); + if (!stat.isFile() || await fs.realpath(source) !== source) fail("changed"); + let size = 0; + for await (const bytes of handle.createReadStream({ autoClose: false, signal: input.signal, highWaterMark: 1024 * 1024 })) { + check(); size += bytes.length; await consume?.(bytes); + } + return { sha256: "", size, mode: stat.mode & 0o777 }; + } finally { await handle.close(); } + } const original = await fs.lstat(source, { bigint: true }); const handle = await fs.open(source, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW | fs.constants.O_NONBLOCK); try { @@ -452,6 +466,10 @@ export async function workspaceArchive(input: { } else if (stat.isFile()) { const entry: Entry = { path: next, type: "file" }; entries.push(entry); + if (archiveIntegrity) { + entry.size = stat.size; entry.mode = stat.mode & 0o777; + continue; + } const read = readStableFile(source).then(value => { Object.assign(entry, value); }, error => { if (!failed) { failed = true; failure = error; } }); @@ -499,14 +517,14 @@ export async function workspaceArchive(input: { for (let index = 0; index < repositories.length; index++) { const relative = repositories[index]!, repo = path.join(root, relative), normalized = path.join(scratch, `repo-${index}.git`); const settings = await portableGitSettings(repo); - gitSignatures.set(relative, await signature(repo, settings)); + if (!archiveIntegrity) gitSignatures.set(relative, await signature(repo, settings)); const offline = await gitStateHash(repo, "shallow") !== null || settings.some(setting => /^remote\..*\.promisor$/u.test(setting.key) && setting.value === "true"); if (offline) { // A shallow local clone uses upload-pack. It drops index-only objects // and asks a filtered source for promised blobs it never downloaded. // Preserve its existing stores and promisor boundaries without fetch. - await verifyGitObjects(repo); + if (!archiveIntegrity) await verifyGitObjects(repo); const format = (await git(repo, ["rev-parse", "--show-object-format"])).trim(); if (!["sha1", "sha256"].includes(format)) fail("unsupported_git_setting"); await git(root, ["init", "--bare", "--template=", `--object-format=${format}`, normalized]); @@ -557,33 +575,45 @@ export async function workspaceArchive(input: { await git(restoredRepo, ["cat-file", "-e", `${object}^{${type}}`]); } await fs.rm(path.join(destination, "objects", "info", "alternates"), { force: true }); - await verifyGitObjects(restoredRepo); - if (await signature(repo) !== gitSignatures.get(relative)) fail("changed"); + if (!archiveIntegrity) { + await verifyGitObjects(restoredRepo); + if (await signature(repo) !== gitSignatures.get(relative)) fail("changed"); + } } const originals = new Map(before.map(entry => [entry.path, entry])); const entries = [...before, ...(await walk(tree)).filter(entry => !originals.has(entry.path))] .sort((a, b) => a.path < b.path ? -1 : a.path > b.path ? 1 : 0); - const manifest = { schema, exclusionPolicy: "paperclip.remote-workspace-exclusions.v1", excluded: [...excludedPaths].sort(), entries }, body = Buffer.from(JSON.stringify(manifest)); + const manifest = { schema, exclusionPolicy: "paperclip.remote-workspace-exclusions.v1", excluded: [...excludedPaths].sort(), entries, + ...(archiveIntegrity ? { bodySha256: "0".repeat(64) } : {}) }; + let body = Buffer.from(JSON.stringify(manifest)); if (body.length > 64 * 1024 * 1024) fail("manifest_too_large"); - const digest = manifestHash(manifest), length = Buffer.alloc(4); length.writeUInt32BE(body.length); + const length = Buffer.alloc(4); length.writeUInt32BE(body.length); const output = await fs.open(input.archivePath, "wx", 0o600); + const bodyDigest = archiveIntegrity ? crypto.createHash("sha256") : undefined; archiveCreated = true; try { await output.writeFile(Buffer.concat([magic, length, body])); for (const entry of entries) if (entry.type === "file") { - const observed = await readStableFile(path.join(originals.has(entry.path) ? root : tree, entry.path), bytes => output.writeFile(bytes)); - if (observed.size !== entry.size || observed.sha256 !== entry.sha256 || observed.mode !== entry.mode) fail("changed"); + const observed = await readStableFile(path.join(originals.has(entry.path) ? root : tree, entry.path), async bytes => { + bodyDigest?.update(bytes); await output.writeFile(bytes); + }, !archiveIntegrity); + if (observed.size !== entry.size || !archiveIntegrity && observed.sha256 !== entry.sha256 || observed.mode !== entry.mode) fail("changed"); + } + if (bodyDigest) { + manifest.bodySha256 = bodyDigest.digest("hex"); + body = Buffer.from(JSON.stringify(manifest)); + if ((await output.write(body, 0, body.length, magic.length + 4)).bytesWritten !== body.length) fail("changed"); } } finally { await output.close(); } directories.length = 0; - if (JSON.stringify(before) !== JSON.stringify(await walk(root, true))) fail("changed"); - for (const relative of repositories) { + if (!archiveIntegrity && JSON.stringify(before) !== JSON.stringify(await walk(root, true))) fail("changed"); + for (const relative of archiveIntegrity ? [] : repositories) { if (await signature(path.join(root, relative)) !== gitSignatures.get(relative)) fail("changed"); } if (input.manifestPath) await fs.writeFile(input.manifestPath, body, { flag: "wx", mode: 0o600 }); const content = await hashFile(input.archivePath); completed = true; - return { schema, manifestDigest: digest, ...content, entries: entries.length }; + return { schema, manifestDigest: manifestHash(manifest), ...content, entries: entries.length }; } finally { if (archiveCreated && !completed) await fs.rm(input.archivePath, { force: true }); await fs.rm(scratch, { recursive: true, force: true }); @@ -604,8 +634,10 @@ export async function workspaceArchive(input: { const bytes = Buffer.alloc(manifestSize); if ((await archive.read(bytes, 0, manifestSize, prefix.length)).bytesRead !== manifestSize) fail("truncated"); check(); - const manifest = JSON.parse(bytes.toString("utf8")) as { schema: string; entries: Entry[] }; + const manifest = JSON.parse(bytes.toString("utf8")) as { schema: string; entries: Entry[]; bodySha256?: string }; if (manifest.schema !== schema || !Array.isArray(manifest.entries) || manifestHash(manifest) !== input.expectedManifestDigest) fail("manifest_mismatch"); + if (manifest.bodySha256 !== undefined && !/^[a-f0-9]{64}$/.test(manifest.bodySha256)) fail("manifest_mismatch"); + const bodyDigest = manifest.bodySha256 === undefined ? undefined : crypto.createHash("sha256"); if (input.maxEntries !== undefined && manifest.entries.length > input.maxEntries) fail("entry_limit"); const kinds = new Map(); let bodySize = 0; for (const entry of manifest.entries) { @@ -614,7 +646,8 @@ export async function workspaceArchive(input: { const parent = path.posix.dirname(entry.path); if (parent !== "." && kinds.get(parent) !== "directory") fail("invalid_parent"); if (entry.type === "file") { - if (!Number.isSafeInteger(entry.size) || entry.size! < 0 || !/^[a-f0-9]{64}$/.test(entry.sha256 ?? "")) fail("invalid_file"); + if (!Number.isSafeInteger(entry.size) || entry.size! < 0 + || (bodyDigest ? entry.sha256 !== undefined : !/^[a-f0-9]{64}$/.test(entry.sha256 ?? ""))) fail("invalid_file"); bodySize += entry.size!; } else if (entry.type === "symlink") { if (!safeLink(entry.path, entry.target!)) fail("unsafe_symlink"); } else if (entry.type !== "directory") fail("invalid_type"); @@ -634,20 +667,21 @@ export async function workspaceArchive(input: { if (entry.type === "directory") { await fs.mkdir(target, { mode: 0o700 }); stagedDirectories.push(target); } else if (entry.type === "symlink") await fs.symlink(entry.target!, target); else { - const output = await fs.open(target, "wx", 0o600), digest = crypto.createHash("sha256"); + const output = await fs.open(target, "wx", 0o600), digest = bodyDigest ? undefined : crypto.createHash("sha256"); try { let remaining = entry.size!; while (remaining) { check(); const read = await archive.read(buffer, 0, Math.min(buffer.length, remaining), offset); if (!read.bytesRead) fail("truncated"); - const part = buffer.subarray(0, read.bytesRead); await output.writeFile(part); digest.update(part); + const part = buffer.subarray(0, read.bytesRead); await output.writeFile(part); (bodyDigest ?? digest!).update(part); offset += read.bytesRead; remaining -= read.bytesRead; } - if (digest.digest("hex") !== entry.sha256) fail("file_hash_mismatch"); + if (digest && digest.digest("hex") !== entry.sha256) fail("file_hash_mismatch"); } finally { await output.close(); } await fs.chmod(target, entry.mode!); } } + if (bodyDigest && bodyDigest.digest("hex") !== manifest.bodySha256) fail("archive_hash_mismatch"); for (const entry of manifest.entries.filter(entry => entry.type === "directory").reverse()) { check(); await fs.chmod(path.join(staging, entry.path), entry.mode!); } @@ -737,7 +771,7 @@ archive({...workerData,signal:controller.signal}).then(result=>parentPort.postMe } } -export function captureRemoteWorkspaceDirectory(input: { cwd: string; archivePath: string; manifestPath?: string; exclude?: string[]; requireNoGit?: boolean; gitSourceRoots?: string[]; signal?: AbortSignal }) { +export function captureRemoteWorkspaceDirectory(input: { cwd: string; archivePath: string; manifestPath?: string; exclude?: string[]; requireNoGit?: boolean; gitSourceRoots?: string[]; signal?: AbortSignal; integrity?: "files" | "archive" }) { return runArchiveWorker({ ...input, kind: "capture" }); } diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index 07cc627cdab..b554dfd52ac 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -130,16 +130,14 @@ export async function findUnchangedWorkspaceSnapshot(input: { }); } -export async function recoverWorkspaceSnapshotArchive(input: { +export async function recoverWorkspaceSnapshotUpload(input: { db: Db; storage: StorageService; snapshotId: string; companyId: string; workspaceId: string; - targetPath: string; signal?: AbortSignal; + signal?: AbortSignal; }): Promise { const [row] = await input.db.select().from(executionWorkspaceSnapshots).where(scope(input)); if (!row) failure("missing"); if (["uploaded", "committed", "superseded"].includes(row!.status)) { - const ref = reference(row!); - await loadWorkspaceSnapshotArchive({ ...input, reference: ref }); - return ref; + return reference(row!); } if (row!.status !== "pending") failure("not_writable"); const attempts = Array.isArray(row!.metadata?.uploads) ? row!.metadata!.uploads as UploadAttempt[] : []; @@ -149,12 +147,20 @@ export async function recoverWorkspaceSnapshotArchive(input: { workspaceId: row!.executionWorkspaceId, generation: row!.generation }; validateReference(ref); if (!(await input.storage.headObject(ref.companyId, ref.objectKey, { signal: input.signal })).exists) continue; - await loadWorkspaceSnapshotArchive({ ...input, reference: ref }); return ref; } return null; } +export async function recoverWorkspaceSnapshotArchive(input: { + db: Db; storage: StorageService; snapshotId: string; companyId: string; workspaceId: string; + targetPath: string; signal?: AbortSignal; +}): Promise { + const ref = await recoverWorkspaceSnapshotUpload(input); + if (ref) await loadWorkspaceSnapshotArchive({ ...input, reference: ref }); + return ref; +} + /** * Where a workspace snapshot save or restore spends its time. Each step is * also a `storage.snapshot.*` run phase, which reaches the run log when it runs @@ -196,6 +202,7 @@ type WorkspaceSnapshotUpload = { unpublishedRecapture?: { sourceOperationId: string }; /** Receives per-step durations and storage request counts. */ timing?: WorkspaceSnapshotStorageTiming; + verification?: "readback" | "upload"; }; /** Upload attempts are journaled before remote I/O. The caller owns publication. */ @@ -204,6 +211,8 @@ export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): P } async function uploadWorkspaceSnapshotTimed(input: WorkspaceSnapshotUpload): Promise { + if (input.verification === "upload") + return uploadVerifiedWorkspaceSnapshot(input, async () => {}, { acceptServerVerifiedUpload: false }); if (input.verifiedArchivePath === undefined) return uploadVerifiedWorkspaceSnapshot(input, ref => verifyObject(input.storage, ref, input.signal), { acceptServerVerifiedUpload: true }); input.signal?.throwIfAborted(); @@ -311,25 +320,33 @@ async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, v const scratch = await mkdtemp(path.join(os.tmpdir(), "pc-workspace-encrypt-")); try { const encrypted = path.join(scratch, "archive.encrypted"), dataKey = randomBytes(32), iv = randomBytes(12); + const ciphertextDigest = createHash("sha256"), prefix = Buffer.concat([MAGIC, iv]); + let ciphertextBytes = prefix.length; + ciphertextDigest.update(prefix); const cipher = createCipheriv("aes-256-gcm", dataKey, iv); cipher.setAAD(aad(identity)); const header = await open(encrypted, "wx", 0o600); - try { await header.writeFile(Buffer.concat([MAGIC, iv])); } finally { await header.close(); } + try { await header.writeFile(prefix); } finally { await header.close(); } // One read of the archive both digests and encrypts it. An unchanged file // identity across that read replaces the separate digest pass. const before = await lstat(input.archivePath); const incomingDigest = createHash("sha256"); let incomingBytes = 0; const observed = new Transform({ transform(bytes, _encoding, callback) { incomingDigest.update(bytes); incomingBytes += bytes.length; callback(null, bytes); } }); - await timedStep(input.timing, "encryptMs", "storage.snapshot.encrypt", () => pipeline(createReadStream(input.archivePath, { signal: input.signal, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), observed, cipher, createWriteStream(encrypted, { flags: "a", mode: 0o600, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), { signal: input.signal })); + const encryptedDigest = new Transform({ transform(bytes, _encoding, callback) { + ciphertextDigest.update(bytes); ciphertextBytes += bytes.length; callback(null, bytes); + } }); + await timedStep(input.timing, "encryptMs", "storage.snapshot.encrypt", () => pipeline(createReadStream(input.archivePath, { signal: input.signal, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), observed, cipher, encryptedDigest, createWriteStream(encrypted, { flags: "a", mode: 0o600, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), { signal: input.signal })); const after = await lstat(input.archivePath); if (!before.isFile() || before.isSymbolicLink() || incomingBytes !== before.size || after.dev !== before.dev || after.ino !== before.ino || after.size !== before.size || after.mtimeMs !== before.mtimeMs || after.ctimeMs !== before.ctimeMs) failure("archive_changed"); const plaintext = { sha256: incomingDigest.digest("hex"), byteSize: incomingBytes }; - const footer = await open(encrypted, "a"); try { await footer.writeFile(cipher.getAuthTag()); } finally { await footer.close(); } + const tag = cipher.getAuthTag(); ciphertextDigest.update(tag); ciphertextBytes += tag.length; + const footer = await open(encrypted, "a"); try { await footer.writeFile(tag); } finally { await footer.close(); } const { material } = await localEncryptedProvider.createVersion({ value: JSON.stringify({ schema: KEY_SCHEMA, ...identity, dataKey: dataKey.toString("base64"), plaintextSha256: plaintext.sha256, plaintextBytes: plaintext.byteSize }) }); dataKey.fill(0); - // Digest the encrypted file as stored: upload parts are read from these bytes. - const content = await timedStep(input.timing, "digestMs", "storage.snapshot.digest", () => digestFile(encrypted, input.signal)); + const content = await timedStep(input.timing, "digestMs", "storage.snapshot.digest", async () => ({ + sha256: ciphertextDigest.digest("hex"), byteSize: ciphertextBytes, + })); if (input.timing) input.timing.bytes = content.byteSize; const attempt: UploadAttempt = { objectKey: `${identity.companyId}/workspace-snapshots/${identity.workspaceId}/${identity.id}/${randomUUID()}.pcws`, ...content, key: material, manifestDigest: input.manifestDigest }; diff --git a/server/src/services/remote-workspace-verification-checkpoint.ts b/server/src/services/remote-workspace-verification-checkpoint.ts deleted file mode 100644 index 03dbbee74f4..00000000000 --- a/server/src/services/remote-workspace-verification-checkpoint.ts +++ /dev/null @@ -1,83 +0,0 @@ -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; - } - }, - }; -}