fix(workspaces): keep single-pass archives compatible with existing readers

Compute format-required file digests during the only source copy and fill their fixed-length header fields afterward. Preserve PCWS1 bytes and old-reader support without pre-hashing source files.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-11-10 08:35:03 -06:00
1 parent 1d4f18e023
commit e876dbdc53
6 files changed
+54 -58

No files matched your search

@@ -36,9 +36,9 @@ 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. 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.
computes its checksums while copying files into it, encrypts it, and
uploads it. Migration trusts a successful upload. It does not pre-hash source
files, 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
@@ -94,13 +94,13 @@ 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.
Migration keeps the PCWS1 archive format. The file checksums required by existing
readers are computed during the single source copy, without a preliminary file
read. Their fixed-length fields are filled in the archive header after copying.
The complete archive is checksummed for storage. Existing readers, saved
snapshots, and interrupted upload journals remain compatible. Ordinary run
captures keep their existing 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
@@ -135,7 +135,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () =>
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(capture.mock.calls.every(([input]) => input.singlePass === true)).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);
@@ -210,7 +210,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () =>
.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(captured.mock.calls.every(([input]) => input.cwd === parent.cwd && input.singlePass === true)).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);
@@ -261,7 +261,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () =>
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(capture.mock.calls.every(([input]) => input.cwd === parent.cwd && input.singlePass === true)).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) {
@@ -478,7 +478,7 @@ describe.skipIf(!support.supported)("bounded workspace cohort migration", () =>
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);
expect(capture).toHaveBeenCalledTimes(mode === "fresh" ? 10 : 11);
expect(capture.mock.calls.every(([input]) => input.cwd === f.root && input.integrity === "archive")).toBe(true);
expect(capture.mock.calls.every(([input]) => input.cwd === f.root && input.singlePass === true)).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();
@@ -111,7 +111,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
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(capture.mock.calls[0][0]).toMatchObject({ singlePass: true });
expect(put).toHaveBeenCalledTimes(1); expect(get).not.toHaveBeenCalled();
get.mockRestore();
const authority = readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)!;
@@ -637,7 +637,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
}
expect(get).not.toHaveBeenCalled();
expect(capture).toHaveBeenCalledTimes(1);
expect(capture.mock.calls[0][0]).toMatchObject({ cwd: parent.cwd, integrity: "archive" });
expect(capture.mock.calls[0][0]).toMatchObject({ cwd: parent.cwd, singlePass: true });
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") {
@@ -811,7 +811,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
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(capture.mock.calls[0][0]).toMatchObject({ singlePass: true });
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);
@@ -554,7 +554,7 @@ async function captureMigration(input: MigrationInput, reserved: ReservedMigrati
if (sourceRoot) await assertArchivedRepositoryIntents(sourceRoot, reserved.archivedRepositoryIntents);
const exclude = nested.map(child => child.relativePath);
const captured = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath, exclude, requireNoGit: onlineSource?.kind === "ordinary",
gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal, integrity: "archive" });
gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal, singlePass: true });
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,
verification: "upload",
@@ -71,7 +71,7 @@ describe("portable remote workspace archives", () => {
const promise = new Promise<void>(done => { resolve = done; });
return { promise, resolve };
}
it("copies each source file once and restores an archive with one body checksum", async () => {
it("copies each source file once and preserves the existing archive format", 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"), "");
@@ -80,13 +80,15 @@ describe("portable remote workspace archives", () => {
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" }) };
const snapshot = { archivePath, ...await captureRemoteWorkspaceDirectory({ cwd, archivePath, singlePass: true }) };
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(manifest.entries.find((entry: { path: string }) => entry.path === "saved.txt").sha256)
.toBe(createHash("sha256").update("saved work\n").digest("hex"));
const compatible = await captureRemoteWorkspaceDirectory({ cwd, archivePath: path.join(root, "compatible.pcws") });
expect(snapshot.sha256).toBe(compatible.sha256);
expect(snapshot.manifestDigest).toBe(compatible.manifestDigest);
expect(snapshot.sha256).toBe(createHash("sha256").update(bytes).digest("hex"));
await fs.rm(cwd, { recursive: true });
const restored = await restore(snapshot);
@@ -96,10 +98,10 @@ describe("portable remote workspace archives", () => {
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");
expectedManifestDigest: snapshot.manifestDigest })).rejects.toThrow("file_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 () => {
it("resumes staged and unstaged Git work from one source 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);
@@ -108,7 +110,7 @@ describe("portable remote workspace archives", () => {
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)] }) };
singlePass: true, 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");
@@ -17,7 +17,7 @@ export async function workspaceArchive(input: {
signal?: AbortSignal;
maxEntries?: number;
maxManifestBytes?: number;
integrity?: "files" | "archive";
singlePass?: boolean;
}): 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");
@@ -27,7 +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";
const singlePass = input.singlePass === true;
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();
@@ -250,18 +250,18 @@ 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<void>, hashContents = true) => {
const readStableFile = async (source: string, consume?: (bytes: Buffer) => Promise<void>, verifySource = true) => {
check();
if (!hashContents) {
if (!verifySource) {
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;
const digest = crypto.createHash("sha256"); 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);
check(); digest.update(bytes); size += bytes.length; await consume?.(bytes);
}
return { sha256: "", size, mode: stat.mode & 0o777 };
return { sha256: digest.digest("hex"), size, mode: stat.mode & 0o777 };
} finally { await handle.close(); }
}
const original = await fs.lstat(source, { bigint: true });
@@ -466,8 +466,8 @@ 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;
if (singlePass) {
Object.assign(entry, { sha256: "0".repeat(64), size: stat.size, mode: stat.mode & 0o777 });
continue;
}
const read = readStableFile(source).then(value => { Object.assign(entry, value); }, error => {
@@ -517,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);
if (!archiveIntegrity) gitSignatures.set(relative, await signature(repo, settings));
if (!singlePass) 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.
if (!archiveIntegrity) await verifyGitObjects(repo);
if (!singlePass) 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]);
@@ -575,7 +575,7 @@ export async function workspaceArchive(input: {
await git(restoredRepo, ["cat-file", "-e", `${object}^{${type}}`]);
}
await fs.rm(path.join(destination, "objects", "info", "alternates"), { force: true });
if (!archiveIntegrity) {
if (!singlePass) {
await verifyGitObjects(restoredRepo);
if (await signature(repo) !== gitSignatures.get(relative)) fail("changed");
}
@@ -583,31 +583,29 @@ export async function workspaceArchive(input: {
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,
...(archiveIntegrity ? { bodySha256: "0".repeat(64) } : {}) };
const manifest = { schema, exclusionPolicy: "paperclip.remote-workspace-exclusions.v1", excluded: [...excludedPaths].sort(), entries };
let body = Buffer.from(JSON.stringify(manifest));
if (body.length > 64 * 1024 * 1024) fail("manifest_too_large");
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), 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");
await output.writeFile(bytes);
}, !singlePass);
if (observed.size !== entry.size || !singlePass && observed.sha256 !== entry.sha256 || observed.mode !== entry.mode) fail("changed");
if (singlePass) entry.sha256 = observed.sha256;
}
if (bodyDigest) {
manifest.bodySha256 = bodyDigest.digest("hex");
if (singlePass) {
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 (!archiveIntegrity && JSON.stringify(before) !== JSON.stringify(await walk(root, true))) fail("changed");
for (const relative of archiveIntegrity ? [] : repositories) {
if (!singlePass && JSON.stringify(before) !== JSON.stringify(await walk(root, true))) fail("changed");
for (const relative of singlePass ? [] : 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 });
@@ -634,10 +632,8 @@ 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[]; bodySha256?: string };
const manifest = JSON.parse(bytes.toString("utf8")) as { schema: string; entries: Entry[] };
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<string, string>(); let bodySize = 0;
for (const entry of manifest.entries) {
@@ -646,8 +642,7 @@ 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
|| (bodyDigest ? entry.sha256 !== undefined : !/^[a-f0-9]{64}$/.test(entry.sha256 ?? ""))) fail("invalid_file");
if (!Number.isSafeInteger(entry.size) || entry.size! < 0 || !/^[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");
@@ -667,21 +662,20 @@ 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 = bodyDigest ? undefined : crypto.createHash("sha256");
const output = await fs.open(target, "wx", 0o600), digest = 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); (bodyDigest ?? digest!).update(part);
const part = buffer.subarray(0, read.bytesRead); await output.writeFile(part); digest.update(part);
offset += read.bytesRead; remaining -= read.bytesRead;
}
if (digest && digest.digest("hex") !== entry.sha256) fail("file_hash_mismatch");
if (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!);
}
@@ -771,7 +765,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; integrity?: "files" | "archive" }) {
export function captureRemoteWorkspaceDirectory(input: { cwd: string; archivePath: string; manifestPath?: string; exclude?: string[]; requireNoGit?: boolean; gitSourceRoots?: string[]; signal?: AbortSignal; singlePass?: boolean }) {
return runArchiveWorker({ ...input, kind: "capture" });
}