diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index bdaa876787a..054b8317510 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1997,6 +1997,13 @@ project with the same repository URL; publication locks and rechecks those sourc rows. Imports from mutable local folders still recapture and compare the current files; a saved upload does not permit later local edits to be discarded. +If every recorded snapshot upload is definitively absent, an unpublished +archived Git import can recapture the same validated logical source. Private Git +administrative bytes can differ between reconstructions. The original journal +and every upload attempt remain recorded; available prior uploads and concurrent +publication winners are reused. Unknown or corrupt objects block recapture. +The source recipe and ownership checks still gate authority publication. + When the verified Git source still exists, reconstruction also copies its local `info/exclude` rules into the private repository before applying a saved restore point. This keeps saved ignored files, including recovery receipts and bundles, diff --git a/server/src/services/remote-workspace-migration.test.ts b/server/src/services/remote-workspace-migration.test.ts index 4054a99e23d..1bf6b7fcfd1 100644 --- a/server/src/services/remote-workspace-migration.test.ts +++ b/server/src/services/remote-workspace-migration.test.ts @@ -727,7 +727,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { const address = server.address(); if (!address || typeof address === "string") throw new Error("test server not bound"); return { bare, repoUrl: `http://127.0.0.1:${address.port}/repository.git` }; } - it.each(["present", "root-history", "pruned", "retry", "source-excludes", "source-excludes-symlink", "missing-base", "missing-base-retry", "missing-base-uploaded-retry", "missing-base-changed-retry", "nested", "nested-linked", "nested-retry", "nested-source-changed-retry", "unapproved-nested", "missing", "expired", "corrupt"])("preserves archived Git edits or refuses an unavailable restore point (%s)", async state => { + it.each(["present", "root-history", "pruned", "retry", "missing-base-failed-upload-retry", "source-excludes", "source-excludes-symlink", "missing-base", "missing-base-retry", "missing-base-uploaded-retry", "missing-base-changed-retry", "nested", "nested-linked", "nested-retry", "nested-source-changed-retry", "unapproved-nested", "missing", "expired", "corrupt"])("preserves archived Git edits or refuses an unavailable restore point (%s)", async state => { const remote = state.startsWith("missing-base") ? await serveRepository() : null; const f = await gitArchiveFixture(remote?.repoUrl); if (remote) { await git(remote.bare, "fetch", f.base, "main:refs/heads/main"); await git(remote.bare, "update-server-info"); } @@ -838,6 +838,20 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { storage.putFile = async input => { const result = await put(input); if (fail) { fail = false; throw new Error("upload reply lost"); } return result; }; await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("upload reply lost"); } + let failedAttempt: Record | undefined; + if (state === "missing-base-failed-upload-retry") { + const put = storage.putFile.bind(storage); let fail = true; + storage.putFile = async input => { + if (fail && input.namespace === "workspace-snapshots") { fail = false; throw new Error("upload_part InternalError"); } + return put(input); + }; + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("upload_part InternalError"); + const [pending] = await db.select().from(executionWorkspaceSnapshots); + expect(pending).toMatchObject({ status: "pending", objectKey: null }); + failedAttempt = (pending!.metadata!.uploads as Record[])[0]!; + expect((await storage.headObject(f.companyId, failedAttempt.objectKey as string)).exists).toBe(false); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + } if (state === "missing-base-uploaded-retry") { const get = storage.getObject.bind(storage); let reads = 0; storage.getObject = async (...args) => { @@ -886,6 +900,16 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { return; } await migrateRemoteWorkspace({ db, storage, ...f, apply: true }); + if (failedAttempt) { + const [saved] = await db.select().from(executionWorkspaceSnapshots); + const attempts = saved!.metadata!.uploads as Record[]; + expect(saved!.status).toBe("committed"); + expect(attempts).toHaveLength(2); + expect(attempts[0]).toEqual(failedAttempt); + expect(attempts[1]!.manifestDigest).not.toBe(failedAttempt.manifestDigest); + expect(await db.select().from(workspaceOperations).where(eq(workspaceOperations.phase, "remote_workspace_migration"))) + .toEqual([expect.objectContaining({ id: saved!.sourceOperationId, status: "succeeded" })]); + } if (interrupted) { const operations = await db.select().from(workspaceOperations).where(eq(workspaceOperations.phase, "remote_workspace_migration")); expect(operations).toHaveLength(1); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index d5b63d8de58..954e5a3b04f 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -376,7 +376,12 @@ export async function migrateRemoteWorkspace(input: { const exclude = nested.map(child => child.relativePath); const captured = recovered ?? await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath, exclude, signal: input.signal }); 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 }); + companyId: input.companyId, workspaceId: input.workspaceId, archivePath, manifestDigest: captured.manifestDigest, signal: input.signal, + // 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 } } : {}), + }); const restoredArchive = path.join(scratch, "verified.pcws"), restored = path.join(scratch, "restored"); await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, targetPath: restoredArchive, signal: input.signal }); await restoreRemoteWorkspaceArchive({ archivePath: restoredArchive, targetDir: restored, expectedManifestDigest: ref.manifestDigest, signal: input.signal }); diff --git a/server/src/services/remote-workspace-snapshot-storage.test.ts b/server/src/services/remote-workspace-snapshot-storage.test.ts index 3dd2ddf04ab..a2d938f009b 100644 --- a/server/src/services/remote-workspace-snapshot-storage.test.ts +++ b/server/src/services/remote-workspace-snapshot-storage.test.ts @@ -75,6 +75,128 @@ suite("remote workspace snapshot storage", () => { expect(row?.status).toBe("pending"); expect((row?.metadata?.uploads as unknown[]).length).toBe(1); await expect(uploadWorkspaceSnapshot({ ...f.upload, manifestDigest: "b".repeat(64) })).rejects.toThrow("operation_content_changed"); }); + it("recaptures only an unavailable unpublished operation while retaining every attempt", async () => { + const f = await fixture(), put = storage.putFile.bind(storage); + storage.putFile = async () => { throw new Error("upload_part InternalError"); }; + await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("upload_part InternalError"); + const [pending] = await db.select().from(executionWorkspaceSnapshots); + const original = (pending!.metadata!.uploads as Record[])[0]!; + await fs.rm(f.archivePath); + await fs.writeFile(path.join(f.cwd, "saved.txt"), "equivalent logical source, different physical capture\n"); + const captured = await captureRemoteWorkspaceDirectory({ cwd: f.cwd, archivePath: f.archivePath }); + const retry = { ...f.upload, manifestDigest: captured.manifestDigest, unpublishedRecapture: { sourceOperationId: f.operationId } }; + await expect(uploadWorkspaceSnapshot({ ...retry, unpublishedRecapture: { sourceOperationId: randomUUID() } })).rejects.toThrow("operation_binding_changed"); + storage.putFile = put; + const ref = await uploadWorkspaceSnapshot(retry); + expect(ref.manifestDigest).toBe(captured.manifestDigest); + const [saved] = await db.select().from(executionWorkspaceSnapshots); + expect(saved!.metadata!.uploads).toEqual([original, expect.objectContaining({ objectKey: ref.objectKey, manifestDigest: captured.manifestDigest })]); + // The capability does not reopen a published snapshot. + await expect(uploadWorkspaceSnapshot({ ...retry, manifestDigest: f.captured.manifestDigest })).rejects.toThrow("operation_content_changed"); + }); + it("prefers an available prior capture over a different recapture", async () => { + const f = await fixture(), put = storage.putFile.bind(storage); + storage.putFile = async input => { await put(input); throw new Error("lost upload reply"); }; + await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("lost upload reply"); + await fs.rm(f.archivePath); + const ref = await uploadWorkspaceSnapshot({ ...f.upload, manifestDigest: "b".repeat(64), unpublishedRecapture: { sourceOperationId: f.operationId } }); + expect(ref.manifestDigest).toBe(f.captured.manifestDigest); + expect((await db.select().from(executionWorkspaceSnapshots))[0]!.metadata!.uploads).toHaveLength(1); + }); + it("rechecks prior objects after encryption before recording a recapture", async () => { + const f = await fixture(), put = storage.putFile.bind(storage), head = storage.headObject.bind(storage); + let savedInput!: Parameters[0]; + storage.putFile = async input => { + const parts: Buffer[] = []; + for await (const bytes of input.body) parts.push(Buffer.from(bytes)); + savedInput = { ...input, body: Buffer.concat(parts) }; + throw new Error("upload interrupted before persistence"); + }; + await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("upload interrupted before persistence"); + let heads = 0; + storage.headObject = async (...args) => { if (++heads === 2) await put(savedInput); return head(...args); }; + storage.putFile = put; + const ref = await uploadWorkspaceSnapshot({ ...f.upload, manifestDigest: "b".repeat(64), unpublishedRecapture: { sourceOperationId: f.operationId } }); + expect(ref.manifestDigest).toBe(f.captured.manifestDigest); + expect(heads).toBe(2); + expect((await db.select().from(executionWorkspaceSnapshots))[0]!.metadata!.uploads).toHaveLength(1); + }); + it("refuses stale recapture admission when another attempt enters the ledger", async () => { + const f = await fixture(), head = storage.headObject.bind(storage); + storage.putFile = async () => { throw new Error("upload unavailable"); }; + await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("upload unavailable"); + let ready!: () => void, release!: () => void; + const paused = new Promise(resolve => { ready = resolve; }), resume = new Promise(resolve => { release = resolve; }); + let heads = 0; + storage.headObject = async (...args) => { if (++heads === 1) { ready(); await resume; } return head(...args); }; + const stale = uploadWorkspaceSnapshot({ ...f.upload, manifestDigest: "b".repeat(64), unpublishedRecapture: { sourceOperationId: f.operationId } }); + try { + await paused; + await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("upload unavailable"); + release(); + await expect(stale).rejects.toThrow("upload_attempts_changed"); + const [saved] = await db.select().from(executionWorkspaceSnapshots); + expect(saved).toMatchObject({ status: "pending", objectKey: null }); + expect(saved!.metadata!.uploads).toHaveLength(2); + } finally { release(); await Promise.allSettled([stale]); } + }); + it.each(["unknown", "corrupt"])("refuses recapture when a prior object is %s", async state => { + const f = await fixture(), put = storage.putFile.bind(storage); + storage.putFile = async input => { await put(input); throw new Error("lost upload reply"); }; + await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("lost upload reply"); + if (state === "unknown") storage.headObject = async () => { throw new Error("HEAD unavailable"); }; + else { + const [pending] = await db.select().from(executionWorkspaceSnapshots); + const [attempt] = pending!.metadata!.uploads as { objectKey: string }[]; + await fs.writeFile(path.join(root, "objects", attempt!.objectKey), "corrupt object"); + } + await expect(uploadWorkspaceSnapshot({ ...f.upload, manifestDigest: "b".repeat(64), unpublishedRecapture: { sourceOperationId: f.operationId } })) + .rejects.toThrow(state === "unknown" ? "HEAD unavailable" : "object_hash_mismatch"); + const [saved] = await db.select().from(executionWorkspaceSnapshots); + expect(saved).toMatchObject({ status: "pending", objectKey: null }); + expect(saved!.metadata!.uploads).toHaveLength(1); + }); + it.each(["original", "recapture", "original-with-lost-reply"])("keeps a %s winner when another physical capture finishes late", async winner => { + const f = await fixture(), put = storage.putFile.bind(storage); + const deferred = () => { + let resolve!: () => void; + const promise = new Promise(ready => { resolve = ready; }); + return { resolve, promise }; + }; + const originalReady = deferred(), originalRelease = deferred(), recaptureReady = deferred(), recaptureRelease = deferred(); + let calls = 0; + storage.putFile = async input => { + const first = ++calls === 1; + if (first) { originalReady.resolve(); await originalRelease.promise; } + else { recaptureReady.resolve(); await recaptureRelease.promise; } + const saved = await put(input); + if (first && winner === "original-with-lost-reply") throw new Error("late original upload reply lost"); + return saved; + }; + const unpublishedRecapture = { sourceOperationId: f.operationId }; + const original = uploadWorkspaceSnapshot({ ...f.upload, unpublishedRecapture }); + let recapture: ReturnType | undefined; + try { + await originalReady.promise; + const archivePath = path.join(root, "recaptured.pcws"); + await fs.writeFile(path.join(f.cwd, "saved.txt"), "different physical capture\n"); + const captured = await captureRemoteWorkspaceDirectory({ cwd: f.cwd, archivePath }); + recapture = uploadWorkspaceSnapshot({ ...f.upload, archivePath, manifestDigest: captured.manifestDigest, unpublishedRecapture }); + await recaptureReady.promise; + (winner === "recapture" ? recaptureRelease : originalRelease).resolve(); + if (winner === "original-with-lost-reply") { + await expect(original).rejects.toThrow("late original upload reply lost"); + recaptureRelease.resolve(); + } + const saved = await (winner === "original" ? original : recapture); + expect(saved.manifestDigest).toBe(winner === "recapture" ? captured.manifestDigest : f.captured.manifestDigest); + originalRelease.resolve(); recaptureRelease.resolve(); + if (winner !== "original-with-lost-reply") expect(await original).toEqual(saved); + expect(await recapture).toEqual(saved); + expect(await getWorkspaceSnapshot(db, { ...f, snapshotId: f.row.id })).toEqual(saved); + expect((await db.select().from(executionWorkspaceSnapshots))[0]!.metadata!.uploads).toHaveLength(2); + } finally { originalRelease.resolve(); recaptureRelease.resolve(); await Promise.allSettled([original, recapture]); } + }); it("authenticates a lost-reply archive without publishing its pending upload", async () => { const f = await fixture(), put = storage.putFile.bind(storage); storage.putFile = async input => { await put(input); throw new Error("lost upload reply"); }; diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index 261253c3781..12c38b45bae 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -5,6 +5,7 @@ import os from "node:os"; import path from "node:path"; import { pipeline } from "node:stream/promises"; import { Transform } from "node:stream"; +import { isDeepStrictEqual } from "node:util"; import { and, eq } from "drizzle-orm"; import { executionWorkspaceSnapshots, type Db } from "@paperclipai/db"; import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js"; @@ -124,15 +125,27 @@ export async function recoverWorkspaceSnapshotArchive(input: { export async function uploadWorkspaceSnapshot(input: { db: Db; storage: StorageService; snapshotId: string; companyId: string; workspaceId: string; archivePath: string; manifestDigest: string; signal?: AbortSignal; + // Only an owner with an immutable logical source may replace an unavailable, + // unpublished capture. A recoverable or concurrently published attempt wins, + // and its returned manifest may differ from the new physical capture. + unpublishedRecapture?: { sourceOperationId: string }; }): Promise { const [row] = await input.db.select().from(executionWorkspaceSnapshots).where(scope(input)); if (!row) failure("missing"); + if (input.unpublishedRecapture && row!.sourceOperationId !== input.unpublishedRecapture.sourceOperationId) failure("operation_binding_changed"); if (["uploaded", "committed", "superseded"].includes(row!.status)) { const saved = reference(row!); if (saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed"); await verifyObject(input.storage, saved, input.signal); return saved; } if (row!.status !== "pending" || !/^[a-f0-9]{64}$/.test(input.manifestDigest)) failure("not_writable"); + const recapture = Boolean(input.unpublishedRecapture); + const requireUnpublished = (current: SnapshotRow) => { + if (current.sourceOperationId !== input.unpublishedRecapture!.sourceOperationId + || current.objectKey !== null || current.sha256 !== null || current.byteSize !== null + || current.key !== null || current.manifestDigest !== null) failure("not_unpublished"); + }; + if (recapture) requireUnpublished(row!); const identity = { id: row!.id, companyId: row!.companyId, workspaceId: row!.executionWorkspaceId, generation: row!.generation }; const publish = async (attempt: UploadAttempt) => { const updated = await input.db.transaction(async tx => { @@ -143,20 +156,27 @@ export async function uploadWorkspaceSnapshot(input: { const winning = updated ?? (await input.db.select().from(executionWorkspaceSnapshots).where(scope(input)))[0]; if (!winning || !["uploaded", "committed", "superseded"].includes(winning.status)) failure("publication_changed"); const saved = reference(winning!); - if (saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed"); + if (!recapture && saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed"); + if (saved.objectKey !== attempt.objectKey) await verifyObject(input.storage, saved, input.signal); return saved; }; // Recover a completed upload even when the previous process lost its reply // and its scratch directory. Failed attempts remain referenced for cleanup. - const attempts = Array.isArray(row!.metadata?.uploads) ? row!.metadata!.uploads as UploadAttempt[] : []; - for (const attempt of attempts) { - if (attempt.manifestDigest !== input.manifestDigest) failure("operation_content_changed"); - const ref = { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef; - validateReference(ref); - if ((await input.storage.headObject(ref.companyId, ref.objectKey, { signal: input.signal })).exists) { - await verifyObject(input.storage, ref, input.signal); return publish(attempt); + const availableAttempt = async (attempts: UploadAttempt[]) => { + for (const attempt of attempts) { + input.signal?.throwIfAborted(); + if (!recapture && attempt.manifestDigest !== input.manifestDigest) failure("operation_content_changed"); + const ref = { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef; + validateReference(ref); + if ((await input.storage.headObject(ref.companyId, ref.objectKey, { signal: input.signal })).exists) { + await verifyObject(input.storage, ref, input.signal); return attempt; + } } - } + return null; + }; + const attempts = Array.isArray(row!.metadata?.uploads) ? row!.metadata!.uploads as UploadAttempt[] : []; + const available = await availableAttempt(attempts); + if (available) return publish(available); 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); @@ -175,17 +195,28 @@ export async function uploadWorkspaceSnapshot(input: { const content = await digestFile(encrypted, input.signal); const attempt: UploadAttempt = { objectKey: `${identity.companyId}/workspace-snapshots/${identity.workspaceId}/${identity.id}/${randomUUID()}.pcws`, ...content, key: material, manifestDigest: input.manifestDigest }; + // Recheck after encryption, then bind admission to the unchanged ledger. + // Remote I/O stays outside the row lock. An older upload completing after + // this check can still win the pending-to-uploaded publication below. + if (recapture) { + const recovered = await availableAttempt(attempts); + if (recovered) return publish(recovered); + } const recorded = await input.db.transaction(async tx => { const [current] = await tx.select().from(executionWorkspaceSnapshots).where(scope(input)).for("update"); - if (!current || current.status !== "pending") return false; - const previous = Array.isArray(current.metadata?.uploads) ? current.metadata.uploads : []; + if (!current || current.status !== "pending") return null; + const previous = Array.isArray(current.metadata?.uploads) ? current.metadata.uploads as UploadAttempt[] : []; + if (recapture) { + requireUnpublished(current); + if (!isDeepStrictEqual(previous, attempts)) failure("upload_attempts_changed"); + } else if (previous.some(saved => saved.manifestDigest !== input.manifestDigest)) failure("operation_content_changed"); await notifyRemoteWorkspaceSnapshots(tx, `workspace:${input.workspaceId}`); await tx.update(executionWorkspaceSnapshots).set({ metadata: { ...current.metadata, uploads: [...previous, attempt] }, updatedAt: new Date() }).where(scope(input)); return true; }); if (!recorded) { const saved = await getWorkspaceSnapshot(input.db, input); - if (!saved || saved.manifestDigest !== input.manifestDigest) failure("publication_changed"); + if (!saved || (!recapture && saved.manifestDigest !== input.manifestDigest)) failure("publication_changed"); await verifyObject(input.storage, saved, input.signal); return saved; } // The object key is unique per attempt, so a late losing uploader cannot @@ -194,6 +225,10 @@ export async function uploadWorkspaceSnapshot(input: { originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted), byteSize: attempt.byteSize, sha256: attempt.sha256, signal: input.signal }); await verifyObject(input.storage, { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef, input.signal); + if (recapture) { + const recovered = await availableAttempt(attempts); + if (recovered) return publish(recovered); + } return await publish(attempt); } finally { await rm(scratch, { recursive: true, force: true }); } }