diff --git a/server/src/services/remote-workspace-snapshot-archive.test.ts b/server/src/services/remote-workspace-snapshot-archive.test.ts index 996eac85864..145f221d76a 100644 --- a/server/src/services/remote-workspace-snapshot-archive.test.ts +++ b/server/src/services/remote-workspace-snapshot-archive.test.ts @@ -11,13 +11,21 @@ import { captureRemoteWorkspaceDirectory, readRemoteWorkspaceArchiveDirectories, const archiveIo = vi.hoisted(() => ({ afterChmod: undefined as ((target: string, mode: number) => void) | undefined, beforeOpen: undefined as ((target: string) => Promise) | undefined, + afterOpen: undefined as ((target: string) => Promise) | undefined, + afterClose: undefined as ((target: string) => void) | undefined, captures: [] as string[], })); vi.mock("node:fs/promises", async importOriginal => { const actual = await importOriginal(); return { ...actual, async open(...args: Parameters) { await archiveIo.beforeOpen?.(String(args[0])); - return actual.open(...args); + const handle = await actual.open(...args), close = handle.close.bind(handle); + let closed = false; + handle.close = async () => { + try { await close(); } finally { if (!closed) { closed = true; archiveIo.afterClose?.(String(args[0])); } } + }; + await archiveIo.afterOpen?.(String(args[0])); + return handle; }, async mkdtemp(...args: Parameters) { const directory = await actual.mkdtemp(...args); if (String(args[0]).includes("pc-workspace-capture-")) archiveIo.captures.push(String(directory)); @@ -35,7 +43,8 @@ describe("portable remote workspace archives", () => { let root: string; beforeEach(async () => { root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-archive-test-")); }); afterEach(async () => { - archiveIo.afterChmod = undefined; archiveIo.beforeOpen = undefined; archiveIo.captures = []; + archiveIo.afterChmod = undefined; archiveIo.beforeOpen = undefined; archiveIo.afterOpen = undefined; + archiveIo.afterClose = undefined; archiveIo.captures = []; await fs.rm(root, { recursive: true, force: true }); }); async function source() { const cwd = path.join(root, "source"); await fs.mkdir(cwd); return cwd; } @@ -54,6 +63,87 @@ describe("portable remote workspace archives", () => { } return JSON.stringify(entries); } + function deferred() { + let resolve!: () => void; + const promise = new Promise(done => { resolve = done; }); + return { promise, resolve }; + } + async function parallelSource() { + const cwd = await source(); + for (const name of ["a", "b", "c", "d", "e", "f"]) { + await fs.mkdir(path.join(cwd, name)); + await fs.writeFile(path.join(cwd, name, "file"), name.repeat(1024)); + } + return fs.realpath(cwd); + } + it("bounds stable reads across directories and preserves bytes after out-of-order completion", async () => { + const cwd = await parallelSource(), blocked: Array> = []; + let gating = true, active = 0, maximum = 0; + const closed: string[] = []; + archiveIo.afterOpen = async target => { + if (!target.startsWith(`${cwd}${path.sep}`)) return; + maximum = Math.max(maximum, ++active); + if (gating) { const gate = deferred(); blocked.push(gate); await gate.promise; } + }; + archiveIo.afterClose = target => { + if (target.startsWith(`${cwd}${path.sep}`)) { active--; closed.push(target); } + }; + const capturing = capture(cwd); + try { + await expect.poll(() => blocked.length).toBe(4); + expect(active).toBe(4); + blocked[3]!.resolve(); + await expect.poll(() => blocked.length).toBe(5); + expect(closed[0]).toBe(path.join(cwd, "d", "file")); + expect(maximum).toBe(4); + } finally { + gating = false; + for (const gate of blocked) gate.resolve(); + await capturing; + } + const snapshot = await capturing; + expect(active).toBe(0); expect(maximum).toBe(4); + archiveIo.afterOpen = undefined; archiveIo.afterClose = undefined; + const helper = path.join(root, "helper.cjs"), archivePath = path.join(root, "serialized.pcws"); + await fs.writeFile(helper, remoteWorkspaceArchiveScript()); + const serialized = JSON.parse((await execute(process.execPath, [helper, "capture", cwd, archivePath])).stdout); + expect(serialized).toMatchObject({ manifestDigest: snapshot.manifestDigest, sha256: snapshot.sha256 }); + expect(await fs.readFile(archivePath)).toEqual(await fs.readFile(snapshot.archivePath)); + }); + it.each(["cancel", "replace"])("drains every parallel stable read after %s", async kind => { + const cwd = await parallelSource(), blocked: Array> = []; + const controller = new AbortController(), reason = new Error("capture cancelled"); + let gating = true, active = 0, settled = false; + archiveIo.afterOpen = async target => { + if (!target.startsWith(`${cwd}${path.sep}`)) return; + active++; + if (gating) { const gate = deferred(); blocked.push(gate); await gate.promise; } + }; + archiveIo.afterClose = target => { if (target.startsWith(`${cwd}${path.sep}`)) active--; }; + const capturing = captureRemoteWorkspaceDirectory({ cwd, archivePath: path.join(root, "snapshot.pcws"), signal: controller.signal }) + .then(() => { settled = true; return null; }, error => { settled = true; return error as Error; }); + try { + await expect.poll(() => blocked.length).toBe(4); + if (kind === "cancel") controller.abort(reason); + else { + await fs.unlink(path.join(cwd, "a", "file")); + await fs.writeFile(path.join(cwd, "a", "file"), "replacement"); + } + for (const gate of blocked.slice(0, 3)) gate.resolve(); + await expect.poll(() => active).toBe(1); + expect(settled).toBe(false); + } finally { + gating = false; + for (const gate of blocked) gate.resolve(); + await capturing; + } + const error = await capturing; + if (kind === "cancel") expect(error === reason || error?.cause === reason).toBe(true); + else expect(error?.message).toBe("remote_workspace_archive_changed"); + expect(active).toBe(0); + await expect(fs.access(path.join(root, "snapshot.pcws"))).rejects.toMatchObject({ code: "ENOENT" }); + for (const directory of archiveIo.captures) await expect(fs.access(directory)).rejects.toMatchObject({ code: "ENOENT" }); + }); it("reads captured directory coverage and rejects a different manifest digest", async () => { const cwd = await source(); await fs.mkdir(path.join(cwd, "checkout")); diff --git a/server/src/services/remote-workspace-snapshot-archive.ts b/server/src/services/remote-workspace-snapshot-archive.ts index cbe63dc7cf5..eeb2f0cbc0b 100644 --- a/server/src/services/remote-workspace-snapshot-archive.ts +++ b/server/src/services/remote-workspace-snapshot-archive.ts @@ -252,6 +252,7 @@ async function workspaceArchive(input: { const before = await handle.stat({ bigint: true }); if (!before.isFile() || original.dev !== before.dev || original.ino !== before.ino || await fs.realpath(source) !== source) fail("changed"); const digest = crypto.createHash("sha256"); let size = 0; + check(); for await (const bytes of handle.createReadStream({ autoClose: false, signal: input.signal })) { check(); digest.update(bytes); size += bytes.length; await consume?.(bytes); @@ -309,6 +310,11 @@ async function workspaceArchive(input: { || next === ".paperclip-runtime" || next.startsWith(".paperclip-runtime/"); const walk = async (root: string, skipGit = false) => { const entries: Entry[] = []; + // One bound for the whole walk, including nested directories. Reserve each + // entry's position before hashing so completion order cannot change identity. + const pending = new Set>(); + let failed = false, failure: unknown; + const checkWalk = () => { if (failed) throw failure; check(); }; const visit = async (relative: string, parentIsRepository = false) => { const directory = relative ? path.join(root, relative) : root; const directoryStat = await fs.lstat(directory); @@ -322,7 +328,7 @@ async function workspaceArchive(input: { await assertRepositoryRoot(directory); } for (const name of names) { - check(); + checkWalk(); const next = relative ? `${relative}/${name}` : name; if (!safeRelative(next)) fail("unsupported_filename"); if (excluded(next)) { @@ -356,12 +362,27 @@ async function workspaceArchive(input: { if (!safeLink(next, link)) fail("unsafe_symlink"); entries.push({ path: next, type: "symlink", target: link }); } else if (stat.isFile()) { - entries.push({ path: next, type: "file", ...await readStableFile(source) }); + const entry: Entry = { path: next, type: "file" }; + entries.push(entry); + const read = readStableFile(source).then(value => { Object.assign(entry, value); }, error => { + if (!failed) { failed = true; failure = error; } + }); + pending.add(read); + void read.then(() => pending.delete(read)); + if (pending.size >= 4) { await Promise.race(pending); checkWalk(); } } else fail("unsupported_entry"); } }; - await visit(""); - return entries; + try { + await visit(""); + await Promise.all(pending); + checkWalk(); + return entries; + } finally { + // A rejected/cancelled walk must close every active stable read before its + // caller removes scratch or returns control to a mutating source owner. + await Promise.all(pending); + } }; const manifestHash = (manifest: { schema: string; entries: Entry[] }) => crypto.createHash("sha256").update(JSON.stringify(manifest)).digest("hex"); if (input.kind === "capture") {