mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
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:
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.
|
an enclosing transaction because a savepoint cannot confirm the outer commit.
|
||||||
|
|
||||||
Reservation runs serially on the original pool. Each member makes one archive,
|
Reservation runs serially on the original pool. Each member makes one archive,
|
||||||
checksums the complete archive body while copying its files, encrypts it, and
|
computes its checksums while copying files into it, encrypts it, and
|
||||||
uploads it. Migration trusts a successful upload. It does not hash each source
|
uploads it. Migration trusts a successful upload. It does not pre-hash source
|
||||||
file, download the upload, extract a test copy, or capture the source again.
|
files, download the upload, extract a test copy, or capture the source again.
|
||||||
Encryption computes the ciphertext checksum as it writes, without a separate
|
Encryption computes the ciphertext checksum as it writes, without a separate
|
||||||
read of the encrypted file. These member operations run concurrently.
|
read of the encrypted file. These member operations run concurrently.
|
||||||
Under the complete fence, preflight compares every selected workspace fingerprint
|
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
|
## Archive compatibility
|
||||||
|
|
||||||
Migration uses the existing PCWS1 container with a `bodySha256` manifest field
|
Migration keeps the PCWS1 archive format. The file checksums required by existing
|
||||||
instead of per-file hashes. The manifest digest binds that checksum. Readers
|
readers are computed during the single source copy, without a preliminary file
|
||||||
check the complete body during extraction. Existing per-file archives remain
|
read. Their fixed-length fields are filled in the archive header after copying.
|
||||||
readable. Deploy the new reader before running this migration: older releases
|
The complete archive is checksummed for storage. Existing readers, saved
|
||||||
cannot read archive-checksummed snapshots. Ordinary run captures keep their
|
snapshots, and interrupted upload journals remain compatible. Ordinary run
|
||||||
existing per-file mode. Object loading still authenticates and decrypts the
|
captures keep their existing mode. Object loading still authenticates and
|
||||||
complete encrypted archive when a real restore is needed.
|
decrypts the complete encrypted archive when a real restore is needed.
|
||||||
|
|
||||||
This deliberately trades migration readback and repeat source comparisons for
|
This deliberately trades migration readback and repeat source comparisons for
|
||||||
speed. It relies on the existing source/writer fences and the storage upload
|
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);
|
expect(maximum).toBe(2);
|
||||||
await expect(run(f)).resolves.toMatchObject({ applied: true, replayed: true });
|
await expect(run(f)).resolves.toMatchObject({ applied: true, replayed: true });
|
||||||
expect(capture).toHaveBeenCalledTimes(2);
|
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();
|
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");
|
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);
|
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);
|
.every(row => readRemoteWorkspaceAuthority(row.metadata))).toBe(true);
|
||||||
expect(await db.select().from(activityLog)).toHaveLength(beforeActivity.length + 2);
|
expect(await db.select().from(activityLog)).toHaveLength(beforeActivity.length + 2);
|
||||||
expect(captured).toHaveBeenCalledTimes(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(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(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);
|
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(uploads).toBe(2);
|
||||||
expect(get).not.toHaveBeenCalled(); expect(head).not.toHaveBeenCalled(); expect(restore).not.toHaveBeenCalled();
|
expect(get).not.toHaveBeenCalled(); expect(head).not.toHaveBeenCalled(); expect(restore).not.toHaveBeenCalled();
|
||||||
expect(capture).toHaveBeenCalledTimes(2);
|
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(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);
|
expect((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, child.id)))[0]).toEqual(childWorkspace);
|
||||||
for (const member of selected) {
|
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!));
|
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(operations).toHaveLength(10); expect(operations.every(row => row.status === "succeeded")).toBe(true);
|
||||||
expect(capture).toHaveBeenCalledTimes(mode === "fresh" ? 10 : 11);
|
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(get).not.toHaveBeenCalled(); expect(restore).not.toHaveBeenCalled();
|
||||||
expect(head.mock.calls.filter(([, key]) => childSnapshots.some(snapshot => snapshot.objectKey === key))).toEqual([]);
|
expect(head.mock.calls.filter(([, key]) => childSnapshots.some(snapshot => snapshot.objectKey === key))).toEqual([]);
|
||||||
if (mode === "fresh") expect(head).not.toHaveBeenCalled();
|
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 });
|
||||||
expect(await migrate()).toMatchObject({ applied: true, replayed: true });
|
expect(await migrate()).toMatchObject({ applied: true, replayed: true });
|
||||||
expect(capture).toHaveBeenCalledTimes(1);
|
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();
|
expect(put).toHaveBeenCalledTimes(1); expect(get).not.toHaveBeenCalled();
|
||||||
get.mockRestore();
|
get.mockRestore();
|
||||||
const authority = readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)!;
|
const authority = readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)!;
|
||||||
@@ -637,7 +637,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
|
|||||||
}
|
}
|
||||||
expect(get).not.toHaveBeenCalled();
|
expect(get).not.toHaveBeenCalled();
|
||||||
expect(capture).toHaveBeenCalledTimes(1);
|
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(readRemoteWorkspaceAuthority((await workspace(child.workspaceId)).metadata)).toEqual(authority);
|
||||||
expect(await fs.readFile(path.join(parent.cwd, "saved.txt"), "utf8")).toBe("original saved work\n");
|
expect(await fs.readFile(path.join(parent.cwd, "saved.txt"), "utf8")).toBe("original saved work\n");
|
||||||
if (mode === "abort") {
|
if (mode === "abort") {
|
||||||
@@ -811,7 +811,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
|
|||||||
const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory");
|
const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory");
|
||||||
await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true });
|
await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true });
|
||||||
expect(restore).not.toHaveBeenCalled(); expect(capture).toHaveBeenCalledTimes(1);
|
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(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(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 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);
|
if (sourceRoot) await assertArchivedRepositoryIntents(sourceRoot, reserved.archivedRepositoryIntents);
|
||||||
const exclude = nested.map(child => child.relativePath);
|
const exclude = nested.map(child => child.relativePath);
|
||||||
const captured = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath, exclude, requireNoGit: onlineSource?.kind === "ordinary",
|
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,
|
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,
|
||||||
verification: "upload",
|
verification: "upload",
|
||||||
|
|||||||
@@ -71,7 +71,7 @@ describe("portable remote workspace archives", () => {
|
|||||||
const promise = new Promise<void>(done => { resolve = done; });
|
const promise = new Promise<void>(done => { resolve = done; });
|
||||||
return { promise, resolve };
|
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");
|
const cwd = await fs.realpath(await source()), file = path.join(cwd, "saved.txt");
|
||||||
await fs.writeFile(file, "saved work\n");
|
await fs.writeFile(file, "saved work\n");
|
||||||
await fs.writeFile(path.join(cwd, "empty.txt"), "");
|
await fs.writeFile(path.join(cwd, "empty.txt"), "");
|
||||||
@@ -80,13 +80,15 @@ describe("portable remote workspace archives", () => {
|
|||||||
const opens: string[] = [];
|
const opens: string[] = [];
|
||||||
archiveIo.afterOpen = async target => { if (target.startsWith(cwd + path.sep)) opens.push(target); };
|
archiveIo.afterOpen = async target => { if (target.startsWith(cwd + path.sep)) opens.push(target); };
|
||||||
const archivePath = path.join(root, "once.pcws");
|
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());
|
expect(opens.sort()).toEqual([path.join(cwd, "empty.txt"), file].sort());
|
||||||
const bytes = await fs.readFile(archivePath), manifestSize = bytes.readUInt32BE(6);
|
const bytes = await fs.readFile(archivePath), manifestSize = bytes.readUInt32BE(6);
|
||||||
const manifest = JSON.parse(bytes.subarray(10, 10 + manifestSize).toString());
|
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.find((entry: { path: string }) => entry.path === "saved.txt").sha256)
|
||||||
expect(manifest.entries.filter((entry: { type: string }) => entry.type === "file")
|
.toBe(createHash("sha256").update("saved work\n").digest("hex"));
|
||||||
.every((entry: { sha256?: string }) => entry.sha256 === undefined)).toBe(true);
|
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"));
|
expect(snapshot.sha256).toBe(createHash("sha256").update(bytes).digest("hex"));
|
||||||
await fs.rm(cwd, { recursive: true });
|
await fs.rm(cwd, { recursive: true });
|
||||||
const restored = await restore(snapshot);
|
const restored = await restore(snapshot);
|
||||||
@@ -96,10 +98,10 @@ describe("portable remote workspace archives", () => {
|
|||||||
bytes[bytes.length - 1] ^= 1;
|
bytes[bytes.length - 1] ^= 1;
|
||||||
const corrupt = path.join(root, "corrupt.pcws"); await fs.writeFile(corrupt, bytes);
|
const corrupt = path.join(root, "corrupt.pcws"); await fs.writeFile(corrupt, bytes);
|
||||||
await expect(restoreRemoteWorkspaceArchive({ archivePath: corrupt, targetDir: path.join(root, "corrupt"),
|
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" });
|
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");
|
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");
|
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);
|
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");
|
await fs.writeFile(path.join(cwd, "new.txt"), "untracked\n");
|
||||||
const archivePath = path.join(root, "git-once.pcws");
|
const archivePath = path.join(root, "git-once.pcws");
|
||||||
const snapshot = { archivePath, ...await captureRemoteWorkspaceDirectory({ cwd, archivePath,
|
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 });
|
await fs.rm(base, { recursive: true }); await fs.rm(cwd, { recursive: true });
|
||||||
const restored = await restore(snapshot);
|
const restored = await restore(snapshot);
|
||||||
expect(await git(restored, "show", ":work.txt")).toBe("staged");
|
expect(await git(restored, "show", ":work.txt")).toBe("staged");
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ export async function workspaceArchive(input: {
|
|||||||
signal?: AbortSignal;
|
signal?: AbortSignal;
|
||||||
maxEntries?: number;
|
maxEntries?: number;
|
||||||
maxManifestBytes?: number;
|
maxManifestBytes?: number;
|
||||||
integrity?: "files" | "archive";
|
singlePass?: boolean;
|
||||||
}): Promise<{ schema: "paperclip.remote-workspace-archive.v1"; manifestDigest: string; byteSize: number; sha256: string; entries: number }> {
|
}): Promise<{ schema: "paperclip.remote-workspace-archive.v1"; manifestDigest: string; byteSize: number; sha256: string; entries: number }> {
|
||||||
const fs = await import("node:fs/promises");
|
const fs = await import("node:fs/promises");
|
||||||
const { createReadStream } = await import("node:fs");
|
const { createReadStream } = await import("node:fs");
|
||||||
@@ -27,7 +27,7 @@ export async function workspaceArchive(input: {
|
|||||||
const child = await import("node:child_process");
|
const child = await import("node:child_process");
|
||||||
const schema = "paperclip.remote-workspace-archive.v1" as const;
|
const schema = "paperclip.remote-workspace-archive.v1" as const;
|
||||||
const magic = Buffer.from("PCWS1\n");
|
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 };
|
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 fail = (reason: string): never => { throw new Error(`remote_workspace_archive_${reason}`); };
|
||||||
const check = () => input.signal?.throwIfAborted();
|
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");
|
|| 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 }] };
|
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();
|
check();
|
||||||
if (!hashContents) {
|
if (!verifySource) {
|
||||||
const handle = await fs.open(source, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW | fs.constants.O_NONBLOCK);
|
const handle = await fs.open(source, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW | fs.constants.O_NONBLOCK);
|
||||||
try {
|
try {
|
||||||
const stat = await handle.stat();
|
const stat = await handle.stat();
|
||||||
if (!stat.isFile() || await fs.realpath(source) !== source) fail("changed");
|
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 })) {
|
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(); }
|
} finally { await handle.close(); }
|
||||||
}
|
}
|
||||||
const original = await fs.lstat(source, { bigint: true });
|
const original = await fs.lstat(source, { bigint: true });
|
||||||
@@ -466,8 +466,8 @@ export async function workspaceArchive(input: {
|
|||||||
} else if (stat.isFile()) {
|
} else if (stat.isFile()) {
|
||||||
const entry: Entry = { path: next, type: "file" };
|
const entry: Entry = { path: next, type: "file" };
|
||||||
entries.push(entry);
|
entries.push(entry);
|
||||||
if (archiveIntegrity) {
|
if (singlePass) {
|
||||||
entry.size = stat.size; entry.mode = stat.mode & 0o777;
|
Object.assign(entry, { sha256: "0".repeat(64), size: stat.size, mode: stat.mode & 0o777 });
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
const read = readStableFile(source).then(value => { Object.assign(entry, value); }, error => {
|
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++) {
|
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 relative = repositories[index]!, repo = path.join(root, relative), normalized = path.join(scratch, `repo-${index}.git`);
|
||||||
const settings = await portableGitSettings(repo);
|
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
|
const offline = await gitStateHash(repo, "shallow") !== null
|
||||||
|| settings.some(setting => /^remote\..*\.promisor$/u.test(setting.key) && setting.value === "true");
|
|| settings.some(setting => /^remote\..*\.promisor$/u.test(setting.key) && setting.value === "true");
|
||||||
if (offline) {
|
if (offline) {
|
||||||
// A shallow local clone uses upload-pack. It drops index-only objects
|
// A shallow local clone uses upload-pack. It drops index-only objects
|
||||||
// and asks a filtered source for promised blobs it never downloaded.
|
// and asks a filtered source for promised blobs it never downloaded.
|
||||||
// Preserve its existing stores and promisor boundaries without fetch.
|
// 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();
|
const format = (await git(repo, ["rev-parse", "--show-object-format"])).trim();
|
||||||
if (!["sha1", "sha256"].includes(format)) fail("unsupported_git_setting");
|
if (!["sha1", "sha256"].includes(format)) fail("unsupported_git_setting");
|
||||||
await git(root, ["init", "--bare", "--template=", `--object-format=${format}`, normalized]);
|
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 git(restoredRepo, ["cat-file", "-e", `${object}^{${type}}`]);
|
||||||
}
|
}
|
||||||
await fs.rm(path.join(destination, "objects", "info", "alternates"), { force: true });
|
await fs.rm(path.join(destination, "objects", "info", "alternates"), { force: true });
|
||||||
if (!archiveIntegrity) {
|
if (!singlePass) {
|
||||||
await verifyGitObjects(restoredRepo);
|
await verifyGitObjects(restoredRepo);
|
||||||
if (await signature(repo) !== gitSignatures.get(relative)) fail("changed");
|
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 originals = new Map(before.map(entry => [entry.path, entry]));
|
||||||
const entries = [...before, ...(await walk(tree)).filter(entry => !originals.has(entry.path))]
|
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);
|
.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,
|
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));
|
let body = Buffer.from(JSON.stringify(manifest));
|
||||||
if (body.length > 64 * 1024 * 1024) fail("manifest_too_large");
|
if (body.length > 64 * 1024 * 1024) fail("manifest_too_large");
|
||||||
const 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 output = await fs.open(input.archivePath, "wx", 0o600);
|
||||||
const bodyDigest = archiveIntegrity ? crypto.createHash("sha256") : undefined;
|
|
||||||
archiveCreated = true;
|
archiveCreated = true;
|
||||||
try {
|
try {
|
||||||
await output.writeFile(Buffer.concat([magic, length, body]));
|
await output.writeFile(Buffer.concat([magic, length, body]));
|
||||||
for (const entry of entries) if (entry.type === "file") {
|
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 => {
|
const observed = await readStableFile(path.join(originals.has(entry.path) ? root : tree, entry.path), async bytes => {
|
||||||
bodyDigest?.update(bytes); await output.writeFile(bytes);
|
await output.writeFile(bytes);
|
||||||
}, !archiveIntegrity);
|
}, !singlePass);
|
||||||
if (observed.size !== entry.size || !archiveIntegrity && observed.sha256 !== entry.sha256 || observed.mode !== entry.mode) fail("changed");
|
if (observed.size !== entry.size || !singlePass && observed.sha256 !== entry.sha256 || observed.mode !== entry.mode) fail("changed");
|
||||||
|
if (singlePass) entry.sha256 = observed.sha256;
|
||||||
}
|
}
|
||||||
if (bodyDigest) {
|
if (singlePass) {
|
||||||
manifest.bodySha256 = bodyDigest.digest("hex");
|
|
||||||
body = Buffer.from(JSON.stringify(manifest));
|
body = Buffer.from(JSON.stringify(manifest));
|
||||||
if ((await output.write(body, 0, body.length, magic.length + 4)).bytesWritten !== body.length) fail("changed");
|
if ((await output.write(body, 0, body.length, magic.length + 4)).bytesWritten !== body.length) fail("changed");
|
||||||
}
|
}
|
||||||
} finally { await output.close(); }
|
} finally { await output.close(); }
|
||||||
directories.length = 0;
|
directories.length = 0;
|
||||||
if (!archiveIntegrity && JSON.stringify(before) !== JSON.stringify(await walk(root, true))) fail("changed");
|
if (!singlePass && JSON.stringify(before) !== JSON.stringify(await walk(root, true))) fail("changed");
|
||||||
for (const relative of archiveIntegrity ? [] : repositories) {
|
for (const relative of singlePass ? [] : repositories) {
|
||||||
if (await signature(path.join(root, relative)) !== gitSignatures.get(relative)) fail("changed");
|
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 });
|
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);
|
const bytes = Buffer.alloc(manifestSize);
|
||||||
if ((await archive.read(bytes, 0, manifestSize, prefix.length)).bytesRead !== manifestSize) fail("truncated");
|
if ((await archive.read(bytes, 0, manifestSize, prefix.length)).bytesRead !== manifestSize) fail("truncated");
|
||||||
check();
|
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.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");
|
if (input.maxEntries !== undefined && manifest.entries.length > input.maxEntries) fail("entry_limit");
|
||||||
const kinds = new Map<string, string>(); let bodySize = 0;
|
const kinds = new Map<string, string>(); let bodySize = 0;
|
||||||
for (const entry of manifest.entries) {
|
for (const entry of manifest.entries) {
|
||||||
@@ -646,8 +642,7 @@ export async function workspaceArchive(input: {
|
|||||||
const parent = path.posix.dirname(entry.path);
|
const parent = path.posix.dirname(entry.path);
|
||||||
if (parent !== "." && kinds.get(parent) !== "directory") fail("invalid_parent");
|
if (parent !== "." && kinds.get(parent) !== "directory") fail("invalid_parent");
|
||||||
if (entry.type === "file") {
|
if (entry.type === "file") {
|
||||||
if (!Number.isSafeInteger(entry.size) || entry.size! < 0
|
if (!Number.isSafeInteger(entry.size) || entry.size! < 0 || !/^[a-f0-9]{64}$/.test(entry.sha256 ?? "")) fail("invalid_file");
|
||||||
|| (bodyDigest ? entry.sha256 !== undefined : !/^[a-f0-9]{64}$/.test(entry.sha256 ?? ""))) fail("invalid_file");
|
|
||||||
bodySize += entry.size!;
|
bodySize += entry.size!;
|
||||||
} else if (entry.type === "symlink") { if (!safeLink(entry.path, entry.target!)) fail("unsafe_symlink"); }
|
} else if (entry.type === "symlink") { if (!safeLink(entry.path, entry.target!)) fail("unsafe_symlink"); }
|
||||||
else if (entry.type !== "directory") fail("invalid_type");
|
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); }
|
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 if (entry.type === "symlink") await fs.symlink(entry.target!, target);
|
||||||
else {
|
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 {
|
try {
|
||||||
let remaining = entry.size!;
|
let remaining = entry.size!;
|
||||||
while (remaining) {
|
while (remaining) {
|
||||||
check(); const read = await archive.read(buffer, 0, Math.min(buffer.length, remaining), offset);
|
check(); const read = await archive.read(buffer, 0, Math.min(buffer.length, remaining), offset);
|
||||||
if (!read.bytesRead) fail("truncated");
|
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;
|
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(); }
|
} finally { await output.close(); }
|
||||||
await fs.chmod(target, entry.mode!);
|
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()) {
|
for (const entry of manifest.entries.filter(entry => entry.type === "directory").reverse()) {
|
||||||
check(); await fs.chmod(path.join(staging, entry.path), entry.mode!);
|
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" });
|
return runArchiveWorker({ ...input, kind: "capture" });
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user