diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index f0eb256c26e..a91f069d681 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -2833,18 +2833,36 @@ cherry-pick, revert, merge, and bisect. The source files remain protected. Finis or abort the Git operation before saving. V1 does not claim a complete saved copy while these continuation records cannot be restored safely. -Saved-copy browsing restores and verifies the complete archive for each request. -Availability batches share one verified access per execution workspace within -the request and allow at most eight remote workspace accesses. Each path still -checks current authority. V1 has no extraction cache across requests. Large saved workspaces can therefore need -substantial transfer time and temporary disk space for each browse request. -Saved Files access admits at most two open extractions per process. It limits an -encrypted archive to 512 MiB, its manifest to 4 MiB, and extraction to 10,000 -entries. It reserves at most 2 GiB of temporary space across those accesses and -checks free disk space with a 256 MiB margin. Capacity failures stop before an -object download. Cleanup releases capacity after it removes the owned files. -These limits apply to saved-copy file access. Agent workspace saves and restores -retain their existing archive limits. +Saved-copy browsing reads the committed archive in place. It range-reads and +decrypts the archive header and manifest, and verifies the manifest against the +snapshot's recorded manifest digest. Listing and file details come from that +verified manifest. Reading or downloading one file range-reads and decrypts only +that file's bytes into a private temporary file, verifies them against the +file's manifest SHA-256, and only then serves them. No browse request downloads or +extracts the whole archive, so archive size does not limit saved Files access. +A preview decrypts and verifies the file in memory, within its preview size limit. +A download reserves the file's size against a 4 GiB per-process budget and checks +free temporary disk with a 256 MiB margin before writing. Cleanup releases the +reservation. A file larger than the budget reports `too_large`. A storage read +that sends no bytes for five minutes is closed. The process keeps parsed +manifests of recent snapshots, charged by manifest size plus their entries, up +to 256 MiB. Concurrent opens of one snapshot share one manifest read, and at most +two manifests are read and parsed at once per process. Manifests are limited to 64 MiB, the archive format's limit, and +500,000 entries; a larger saved copy reports `too_large`. +Saved files carry no modification times, so the recent view of a saved copy is +ordered by path. Availability batches share one verified access per execution +workspace within the request and allow at most eight remote workspace accesses. +Each path still checks current authority. +The changed-files view needs the repository status, so it still restores the +complete archive for that request. That extraction admits at most two open +extractions per process. It limits an encrypted archive to 512 MiB, its manifest +to 4 MiB, and extraction to 10,000 entries. It reserves at most 2 GiB of +temporary space across those accesses and checks free disk space with a 256 MiB +margin. A saved copy beyond those limits reports changed files as unavailable; +its other views stay available. Capacity failures stop before an object download. +Cleanup releases capacity after it removes the owned files. These limits apply to +saved-copy file access. Agent workspace saves and restores retain their existing +archive limits. Each managed repository preparation also saves a complete workspace archive. These saves preserve each preparation boundary; V1 does not combine them or use incremental uploads. diff --git a/server/src/__tests__/remote-workspace-file-resources.test.ts b/server/src/__tests__/remote-workspace-file-resources.test.ts index 402d0c7f812..cf57b066568 100644 --- a/server/src/__tests__/remote-workspace-file-resources.test.ts +++ b/server/src/__tests__/remote-workspace-file-resources.test.ts @@ -25,6 +25,7 @@ import { beginWorkspaceSnapshot, uploadWorkspaceSnapshot } from "../services/rem import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; import { createStorageService } from "../storage/service.js"; import { createSavedWorkspaceLoader } from "../services/remote-workspaces.js"; +import { createSavedWorkspaceArchiveOpener } from "../services/saved-workspace-archive.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; const execFileAsync = promisify(execFile); @@ -328,8 +329,9 @@ describeDatabase("remote workspace ordinary file routes", () => { { workspace: "execution", path: "docs/result.md" })).rejects.toMatchObject({ status: 422, details: { code: "file_changed" } }); }); - async function savedFileFixture(keepLocation = false) { + async function savedFileFixture(keepLocation = false, populate?: (root: string) => Promise) { const f = await fixture(), storageRoot = await fs.mkdtemp(path.join(os.tmpdir(), "saved-file-budget-")); + await populate?.(f.root); directories.push(storageRoot); const storage = createStorageService(createLocalDiskStorageProvider(path.join(storageRoot, "objects"))); const archivePath = path.join(storageRoot, "saved.pcws"), archive = await captureRemoteWorkspaceDirectory({ cwd: f.root, archivePath }); @@ -345,6 +347,277 @@ describeDatabase("remote workspace ordinary file routes", () => { return { ...f, storage, saved, input: { companyId: f.ids.company, workspaceId: f.ids.workspace, snapshotId: snapshot.id } }; } + const richWorkspace = async (root: string) => { + await fs.mkdir(path.join(root, "docs", "deep", "deeper"), { recursive: true }); + await fs.mkdir(path.join(root, "node_modules", "pkg"), { recursive: true }); + await fs.mkdir(path.join(root, "src")); + await fs.writeFile(path.join(root, "docs", "deep", "notes.txt"), "nested notes\n"); + await fs.writeFile(path.join(root, "docs", "deep", "deeper", "leaf.md"), "leaf\n"); + await fs.writeFile(path.join(root, "docs", "binary.bin"), Buffer.from([0, 1, 2, 3, 255, 0, 7])); + await fs.writeFile(path.join(root, "docs", "big.bin"), Buffer.alloc(256 * 1024, 7)); + await fs.writeFile(path.join(root, "node_modules", "pkg", "index.js"), "module.exports = 1;\n"); + await fs.writeFile(path.join(root, "server.key"), "secret"); + for (let index = 0; index < 12; index++) await fs.writeFile(path.join(root, "src", `file-${index}.ts`), `export const v${index} = ${index};\n`); + await fs.symlink("result.md", path.join(root, "docs", "link.md")); + }; + + it("reads a saved workspace above the extraction cap in place, one verified file at a time", async () => { + const f = await savedFileFixture(false, richWorkspace); + const capped = createSavedWorkspaceLoader(db, f.storage, { limits: { maxArchiveBytes: 1024 } }); + const base = `/api/issues/${f.ids.issue}/file-resources`; + const extractOnly = filesApp(f.ids.company, createRemoteWorkspaceFiles(db, { loadSavedWorkspace: capped })); + const refused = await request(extractOnly).get(`${base}/content`).query({ workspace: "execution", path: "docs/result.md" }).expect(422); + expect(refused.body.details.code).toBe("too_large"); + + const getObject = f.storage.getObject.bind(f.storage), ranges: Array<{ start: number; end: number } | undefined> = []; + vi.spyOn(f.storage, "getObject").mockImplementation(async (...args) => { ranges.push(args[2]?.range); return getObject(...args); }); + const tmpBefore = (await fs.readdir(os.tmpdir())).filter(name => name.startsWith("pc-workspace-files-")).length; + const app = filesApp(f.ids.company, createRemoteWorkspaceFiles(db, { loadSavedWorkspace: capped, + openSavedWorkspaceArchive: createSavedWorkspaceArchiveOpener(db, f.storage) })); + const preview = await request(app).get(`${base}/content`).query({ workspace: "execution", path: "docs/result.md" }).expect(200); + expect(preview.body).toMatchObject({ content: { data: "Remote useful result\n" }, resource: { source: { kind: "saved", snapshotId: f.saved.id } } }); + const listed = await request(app).get(`${base}/list`).query({ workspace: "execution", path: "docs" }).expect(200); + expect(listed.body.items.map((item: { relativePath: string }) => item.relativePath)).toEqual(expect.arrayContaining(["docs/big.bin", "docs/result.md"])); + const download = await request(app).get(`${base}/content`).query({ workspace: "execution", path: "docs/big.bin", download: true }) + .buffer(true).parse((response, done) => { const chunks: Buffer[] = []; response.on("data", (chunk: Buffer) => chunks.push(chunk)); response.on("end", () => done(null, Buffer.concat(chunks))); }) + .expect(200); + expect(Buffer.compare(download.body as Buffer, Buffer.alloc(256 * 1024, 7))).toBe(0); + expect(ranges.length).toBeGreaterThan(0); + expect(ranges.every(range => range !== undefined)).toBe(true); + const fetched = ranges.reduce((total, range) => total + range!.end - range!.start + 1, 0); + expect(fetched).toBeLessThan(f.saved.byteSize + 64 * 1024); + expect((await fs.readdir(os.tmpdir())).filter(name => name.startsWith("pc-workspace-files-")).length).toBe(tmpBefore); + }); + + it("answers saved stat and list exactly like the extracted helper", async () => { + const f = await savedFileFixture(false, richWorkspace); + const extracted = createRemoteWorkspaceFiles(db, { loadSavedWorkspace: createSavedWorkspaceLoader(db, f.storage) }); + const archived = createRemoteWorkspaceFiles(db, { openSavedWorkspaceArchive: createSavedWorkspaceArchiveOpener(db, f.storage) }); + const scope = { companyId: f.ids.company, workspaceId: f.ids.workspace }; + const outcome = async (work: () => Promise) => { + try { + const value = JSON.parse(JSON.stringify(await work())); + const strip = (entry: Record) => { delete entry.modifiedAt; return entry; }; + if (Array.isArray(value?.entries)) value.entries.forEach(strip); else if (value && typeof value === "object") strip(value); + return { value }; + } catch (error) { return { status: (error as { status?: number }).status, code: (error as { details?: { code?: string } }).details?.code }; } + }; + const [left, right] = [await extracted.open(scope), await archived.open(scope)]; + try { + for (const relative of ["docs/result.md", "docs", "docs/binary.bin", "docs/big.bin", "docs/deep/deeper/leaf.md", "missing.md", + "docs/result.md/inner", "docs/link.md", ".env", "server.key", "node_modules/pkg/index.js", "../outside", "docs//x"]) { + expect(await outcome(() => right!.stat(relative)), relative).toEqual(await outcome(() => left!.stat(relative))); + } + const lists = [ + { mode: "all", limit: 100, offset: 0, maxScannedEntries: 5000, maxDepth: 20 }, + { mode: "all", relativePath: "docs", limit: 100, offset: 0, maxScannedEntries: 5000, maxDepth: 20 }, + { mode: "all", relativePath: "src", limit: 5, offset: 3, maxScannedEntries: 5000, maxDepth: 20 }, + { mode: "all", query: "leaf", limit: 100, offset: 0, maxScannedEntries: 5000, maxDepth: 20 }, + { mode: "all", query: "file", limit: 100, offset: 0, maxScannedEntries: 7, maxDepth: 20 }, + { mode: "recent", limit: 100, offset: 0, maxScannedEntries: 5000, maxDepth: 1 }, + { mode: "recent", limit: 4, offset: 2, maxScannedEntries: 5000, maxDepth: 20 }, + { mode: "all", relativePath: "docs/result.md", limit: 100, offset: 0, maxScannedEntries: 5000, maxDepth: 20 }, + { mode: "all", relativePath: "node_modules", limit: 100, offset: 0, maxScannedEntries: 5000, maxDepth: 20 }, + ] as const; + for (const list of lists) { + const [expected, actual] = [await outcome(() => left!.list({ ...list })), await outcome(() => right!.list({ ...list }))]; + if (list.mode === "recent" && list.limit === 100) { + const byPath = (value: { value?: { entries?: Array<{ relativePath: string }> } }) => + [...(value.value?.entries ?? [])].sort((a, b) => a.relativePath.localeCompare(b.relativePath)); + expect(byPath(actual), JSON.stringify(list)).toEqual(byPath(expected)); + expect(byPath(actual).map(entry => entry.relativePath)) + .toEqual((actual.value as { entries: Array<{ relativePath: string }> }).entries.map(entry => entry.relativePath)); + } else if (list.mode !== "recent") { + expect(actual, JSON.stringify(list)).toEqual(expected); + } + } + } finally { await left?.close(); await right?.close(); } + }); + + it("refuses a saved file whose stored bytes no longer match its verified manifest", async () => { + const f = await savedFileFixture(false, richWorkspace); + const getObject = f.storage.getObject.bind(f.storage); + vi.spyOn(f.storage, "getObject").mockImplementation(async (...args) => { + const object = await getObject(...args); + if (!args[2]?.range || args[2].range.end - args[2].range.start < 1024) return object; + const { Transform } = await import("node:stream"); + return { ...object, stream: object.stream.pipe(new Transform({ transform(bytes: Buffer, _encoding, callback) { + const tampered = Buffer.from(bytes); tampered[0] ^= 0xff; callback(null, tampered); + } })) }; + }); + const archived = createRemoteWorkspaceFiles(db, { openSavedWorkspaceArchive: createSavedWorkspaceArchiveOpener(db, f.storage) }); + const access = await archived.open({ companyId: f.ids.company, workspaceId: f.ids.workspace }); + try { + await expect(access!.download("docs/big.bin")).rejects.toMatchObject({ status: 422, details: { code: "file_changed" } }); + } finally { await access?.close(); } + }); + + it("keeps changed files on bounded extraction and names its limit for a saved workspace above the cap", async () => { + const f = await savedFileFixture(false, richWorkspace); + const opener = createSavedWorkspaceArchiveOpener(db, f.storage), scope = { companyId: f.ids.company, workspaceId: f.ids.workspace }; + const changed = { mode: "changed", limit: 100, offset: 0, maxScannedEntries: 5000, maxDepth: 20 } as const; + const capped = createRemoteWorkspaceFiles(db, { openSavedWorkspaceArchive: opener, + loadSavedWorkspace: createSavedWorkspaceLoader(db, f.storage, { limits: { maxArchiveBytes: 1024 } }) }); + const cappedAccess = await capped.open(scope); + try { await expect(cappedAccess!.list({ ...changed })).rejects.toMatchObject({ status: 422, details: { code: "changed_unavailable" } }); } + finally { await cappedAccess?.close(); } + const bounded = createRemoteWorkspaceFiles(db, { openSavedWorkspaceArchive: opener, loadSavedWorkspace: createSavedWorkspaceLoader(db, f.storage) }); + const boundedAccess = await bounded.open(scope); + try { + await expect(boundedAccess!.list({ ...changed })).rejects.toMatchObject({ details: { code: "changed_unavailable" } }); + expect(await boundedAccess!.stat("docs/result.md")).toMatchObject({ kind: "file", byteSize: 21 }); + } finally { await boundedAccess?.close(); } + }); + + it("reserves bounded temporary disk for saved downloads and releases it after each one", async () => { + const f = await savedFileFixture(false, richWorkspace); + const scope = { companyId: f.ids.company, workspaceId: f.ids.workspace }, opener = createSavedWorkspaceArchiveOpener(db, f.storage); + const files = (limits: NonNullable[1]>["savedFileDownloadLimits"]) => + createRemoteWorkspaceFiles(db, { openSavedWorkspaceArchive: opener, savedFileDownloadLimits: limits }); + const attempt = async (limits: Parameters[0]) => { + const access = await files(limits).open(scope); + try { + const download = await access!.download("docs/big.bin"); + let size = 0; for await (const chunk of download.stream) size += chunk.length; await download.cleanup(); + return size; + } finally { await access?.close(); } + }; + await expect(attempt({ maxReservedBytes: 1024 })).rejects.toMatchObject({ status: 422, details: { code: "too_large" } }); + await expect(attempt({ availableBytes: async () => 0n })).rejects.toMatchObject({ status: 422, details: { code: "workspace_unavailable" } }); + const fitsOne = { maxReservedBytes: 256 * 1024, freeSpaceMarginBytes: 0 }; + expect(await attempt(fitsOne)).toBe(256 * 1024); + expect(await attempt(fitsOne)).toBe(256 * 1024); + const access = await files(fitsOne).open(scope); + try { + const held = await access!.download("docs/big.bin"); + await expect(access!.download("docs/big.bin")).rejects.toMatchObject({ status: 422, details: { code: "workspace_unavailable" } }); + held.stream.destroy(); await held.cleanup(); + } finally { await access?.close(); } + const preview = await (await files(fitsOne).open(scope))!.read("docs/result.md", { maxBytes: 1024 }); + expect(preview.toString()).toBe("Remote useful result\n"); + let freeSpaceChecked!: () => void; + const checking = new Promise(resolve => { freeSpaceChecked = resolve; }); + const slow = { ...fitsOne, availableBytes: async () => { await checking; return 1n << 40n; } }; + const [first, second] = [await files(slow).open(scope), await files(slow).open(scope)]; + try { + const together = Promise.allSettled([first!.download("docs/big.bin"), second!.download("docs/big.bin")]); + await new Promise(resolve => setTimeout(resolve, 50)); + freeSpaceChecked(); + const outcomes = await together; + expect(outcomes.filter(outcome => outcome.status === "fulfilled")).toHaveLength(1); + expect(outcomes.find(outcome => outcome.status === "rejected")).toMatchObject({ reason: { status: 422, details: { code: "workspace_unavailable" } } }); + for (const outcome of outcomes) if (outcome.status === "fulfilled") { outcome.value.stream.destroy(); await outcome.value.cleanup(); } + } finally { await first?.close(); await second?.close(); } + }); + + it("closes the storage stream when a saved range read is abandoned or stalls", async () => { + const f = await savedFileFixture(false, richWorkspace); + const getObject = f.storage.getObject.bind(f.storage), sources: Array = []; + let stall = false; + vi.spyOn(f.storage, "getObject").mockImplementation(async (...args) => { + const object = await getObject(...args); + if (stall && args[2]?.range && args[2].range.end - args[2].range.start > 1024) { + object.stream.destroy(); + const { PassThrough } = await import("node:stream"); + const silent = new PassThrough(); + sources.push(silent); + return { ...object, stream: silent }; + } + sources.push(object.stream); + return object; + }); + const opener = createSavedWorkspaceArchiveOpener(db, f.storage, { idleTimeoutMs: 100 }); + const archive = await opener({ companyId: f.ids.company, workspaceId: f.ids.workspace, snapshotId: f.saved.id }); + try { + const entry = archive.entry("docs/big.bin")!; + const abandoned = await archive.read(entry, 0, entry.size); + await new Promise(resolve => abandoned.once("readable", resolve)); + abandoned.destroy(); + await vi.waitFor(() => expect(sources.at(-1)!.destroyed).toBe(true)); + stall = true; + const stalled = await archive.read(entry, 0, entry.size); + await expect((async () => { for await (const chunk of stalled) void chunk; })()).rejects.toThrow("remote_workspace_snapshot_download_stalled"); + await vi.waitFor(() => expect(sources.at(-1)!.destroyed).toBe(true)); + } finally { archive.close(); } + }); + + it("charges one cached index for overlapping cold opens and names an oversized manifest", async () => { + const f = await savedFileFixture(false, richWorkspace); + const input = { companyId: f.ids.company, workspaceId: f.ids.workspace, snapshotId: f.saved.id }; + const single = createSavedWorkspaceArchiveOpener(db, f.storage); + const once = await single(input); const charge = once.cachedIndexBytes(); once.close(); + const getObject = f.storage.getObject.bind(f.storage), manifestReads: unknown[] = []; + const spy = vi.spyOn(f.storage, "getObject").mockImplementation(async (...args) => { + if (args[2]?.range?.start === 19) manifestReads.push(args[2].range); + return getObject(...args); + }); + const overlapping = createSavedWorkspaceArchiveOpener(db, f.storage); + const opened = await Promise.all([overlapping(input), overlapping(input), overlapping(input)]); + spy.mockRestore(); + expect(manifestReads).toHaveLength(2); + expect(opened.map(archive => archive.cachedIndexBytes())).toEqual([charge, charge, charge]); + for (const archive of opened) archive.close(); + const capped = createRemoteWorkspaceFiles(db, { openSavedWorkspaceArchive: createSavedWorkspaceArchiveOpener(db, f.storage, { maxIndexEntries: 3 }) }); + await expect(capped.open({ companyId: f.ids.company, workspaceId: f.ids.workspace })) + .rejects.toMatchObject({ status: 422, details: { code: "too_large" } }); + }); + + it("lets a cancelled saved open stop waiting for a busy manifest build", async () => { + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const held = async () => { + const f = await savedFileFixture(false); + const getObject = f.storage.getObject.bind(f.storage); + vi.spyOn(f.storage, "getObject").mockImplementation(async (...args) => { + if (args[2]?.range?.start === 19) await gate; + return getObject(...args); + }); + const opener = createSavedWorkspaceArchiveOpener(db, f.storage); + return { opener, input: { companyId: f.ids.company, workspaceId: f.ids.workspace, snapshotId: f.saved.id } }; + }; + const [first, second, third] = [await held(), await held(), await held()]; + const busy = [first.opener(first.input), second.opener(second.input)]; + await new Promise(resolve => setTimeout(resolve, 50)); + const queued = new AbortController(), sharing = new AbortController(); + const queuedOpen = third.opener({ ...third.input, signal: queued.signal }); + const sharedOpen = first.opener({ ...first.input, signal: sharing.signal }); + await new Promise(resolve => setTimeout(resolve, 50)); + const started = performance.now(); + queued.abort(new Error("request closed")); sharing.abort(new Error("request closed")); + await expect(queuedOpen).rejects.toThrow("request closed"); + await expect(sharedOpen).rejects.toThrow("request closed"); + expect(performance.now() - started).toBeLessThan(1000); + release(); + for (const archive of await Promise.all(busy)) { expect(archive.entry("docs/result.md")).toBeDefined(); archive.close(); } + const reopened = await third.opener(third.input); + expect(reopened.entry("docs/result.md")).toBeDefined(); reopened.close(); + }); + + it("retries a failed shared manifest build once for all of its waiters", async () => { + const f = await savedFileFixture(false); + const getObject = f.storage.getObject.bind(f.storage); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let manifestReads = 0, failed = false; + vi.spyOn(f.storage, "getObject").mockImplementation(async (...args) => { + if (args[2]?.range?.start === 19) { + manifestReads++; + if (!failed) { failed = true; await gate; throw new Error("transient storage failure"); } + } + return getObject(...args); + }); + const opener = createSavedWorkspaceArchiveOpener(db, f.storage); + const input = { companyId: f.ids.company, workspaceId: f.ids.workspace, snapshotId: f.saved.id }; + const first = opener(input); + await new Promise(resolve => setTimeout(resolve, 50)); + const waiters = [opener(input), opener(input), opener(input)]; + await new Promise(resolve => setTimeout(resolve, 50)); + release(); + await expect(first).rejects.toThrow("transient storage failure"); + for (const archive of await Promise.all(waiters)) { expect(archive.entry("docs/result.md")).toBeDefined(); archive.close(); } + expect(manifestReads).toBe(3); + }); + it("keeps a production saved preview available when cleanup clears its location during archive loading", async () => { const f = await savedFileFixture(true); const requestTemp = await fs.mkdtemp(path.join(os.tmpdir(), "saved-file-location-request-")); @@ -356,15 +629,12 @@ describeDatabase("remote workspace ordinary file routes", () => { managedPluginAutoInstall: [], decisionServiceOptions: { wakeOriginAgent: async () => null } }); const originalTmpdir = process.env.TMPDIR; process.env.TMPDIR = requestTemp; - let extractedCopy: string | undefined; + let cleared = false; const getObject = f.storage.getObject.bind(f.storage); const retrieve = vi.spyOn(f.storage, "getObject").mockImplementation(async (...args) => { const object = await getObject(...args); - const copies = (await fs.readdir(requestTemp)).filter(name => name.startsWith("pc-workspace-files-")); - expect(copies).toHaveLength(1); - extractedCopy = path.join(requestTemp, copies[0]!); - // The committed bytes and writer epoch remain unchanged. Only the - // allocation cleanup recorded by production has finished meanwhile. + if (cleared) return object; + cleared = true; await db.update(executionWorkspaces).set({ metadata: { remoteWorkspace: { ...f.authority, snapshotId: f.saved.id, location: null, } } }).where(eq(executionWorkspaces.id, f.ids.workspace)); @@ -379,9 +649,9 @@ describeDatabase("remote workspace ordinary file routes", () => { expect({ status: response.status, code: response.body.details?.code }).toEqual({ status: 200, code: undefined }); expect(response.body).toMatchObject({ resource: { source: { kind: "saved", snapshotId: f.saved.id } }, content: { encoding: "utf8", data: "Remote useful result\n" } }); - expect(retrieve).toHaveBeenCalledTimes(1); - expect(extractedCopy).toBeDefined(); - await expect(fs.stat(extractedCopy!)).rejects.toMatchObject({ code: "ENOENT" }); + expect(retrieve).toHaveBeenCalled(); + expect(retrieve.mock.calls.every(call => call[2]?.range !== undefined)).toBe(true); + expect(await fs.readdir(requestTemp)).toEqual([]); } finally { retrieve.mockRestore(); try { await app.locals.paperclipShutdown(); } diff --git a/server/src/app.ts b/server/src/app.ts index ac2e0e23234..987ebd685a8 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -128,6 +128,7 @@ import { workflowRoutes } from "./routes/workflows.js"; import { fileResourceRoutes } from "./routes/file-resources.js"; import { createRemoteWorkspaceFiles } from "./services/remote-workspace-files.js"; import { createSavedWorkspaceLoader } from "./services/remote-workspaces.js"; +import { createSavedWorkspaceArchiveOpener } from "./services/saved-workspace-archive.js"; import { routineRoutes } from "./routes/routines.js"; import { environmentRoutes } from "./routes/environments.js"; import { executionWorkspaceRoutes } from "./routes/execution-workspaces.js"; @@ -811,6 +812,7 @@ export async function createApp( api.use(fileResourceRoutes(db, { remoteFiles: createRemoteWorkspaceFiles(db, { environmentRuntime: environmentRuntimeService(db, { pluginWorkerManager: workerManager }), loadSavedWorkspace: createSavedWorkspaceLoader(db, opts.storageService), + openSavedWorkspaceArchive: createSavedWorkspaceArchiveOpener(db, opts.storageService), }) })); api.use(routineRoutes(db, { pluginWorkerManager: workerManager })); api.use( diff --git a/server/src/services/customer-success-inspection.ts b/server/src/services/customer-success-inspection.ts index 5254e27a971..63b5b49f1ce 100644 --- a/server/src/services/customer-success-inspection.ts +++ b/server/src/services/customer-success-inspection.ts @@ -23,6 +23,7 @@ import { skillPackageStore } from "./skill-package-store.js"; import { getRunLogStore, isRunLogStoreType } from "./run-log-store.js"; import { createRemoteWorkspaceFiles } from "./remote-workspace-files.js"; import { createSavedWorkspaceLoader } from "./remote-workspaces.js"; +import { createSavedWorkspaceArchiveOpener } from "./saved-workspace-archive.js"; import { environmentRuntimeService } from "./environment-runtime.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; @@ -340,6 +341,7 @@ export async function readCustomerSuccessResource( const remoteFiles = createRemoteWorkspaceFiles(tx, { environmentRuntime: environmentRuntimeService(tx, { pluginWorkerManager: options.pluginWorkerManager, readOnlyInspection: true }), loadSavedWorkspace: createSavedWorkspaceLoader(tx, storage), + openSavedWorkspaceArchive: createSavedWorkspaceArchiveOpener(tx, storage), }); const files = workspaceFileResourceService(tx, undefined, { remoteFiles }); const input = { path: q.path ?? "", ...q.context, limit: q.limit, offset: q.offset }; diff --git a/server/src/services/remote-workspace-files.ts b/server/src/services/remote-workspace-files.ts index d929e8be3db..bd2c6ef591b 100644 --- a/server/src/services/remote-workspace-files.ts +++ b/server/src/services/remote-workspace-files.ts @@ -1,11 +1,12 @@ import { createHash, randomUUID } from "node:crypto"; -import { createReadStream } from "node:fs"; +import { createReadStream, createWriteStream } from "node:fs"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { execFile } from "node:child_process"; import { promisify } from "node:util"; import type { Readable } from "node:stream"; +import { pipeline } from "node:stream/promises"; import { and, eq } from "drizzle-orm"; import { environments, environmentLeases, executionWorkspaces, type Db } from "@paperclipai/db"; import type { Environment, EnvironmentLease, WorkspaceFileListMode, WorkspaceFileSource } from "@paperclipai/shared"; @@ -13,6 +14,7 @@ import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; import { HttpError, notFound, unprocessable } from "../errors.js"; import type { EnvironmentRuntimeService } from "./environment-runtime.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; +import type { OpenSavedWorkspaceArchive, SavedWorkspaceArchive } from "./saved-workspace-archive.js"; const execFileAsync = promisify(execFile); const READ_TIMEOUT_MS = 30_000; @@ -71,6 +73,8 @@ export interface RemoteWorkspaceFileOptions { snapshotId: string; signal?: AbortSignal; }): Promise; + openSavedWorkspaceArchive?: OpenSavedWorkspaceArchive; + savedFileDownloadLimits?: Partial; } export interface RemoteWorkspaceFiles { @@ -258,6 +262,168 @@ async function verifyDownload(file: string, expected: RemoteWorkspaceFileEntry): } finally { await handle.close(); } } +class ArchiveReadFailure extends Error { + constructor(readonly code: string) { super(code); } +} + +const ARCHIVE_DENIED_NAMES = new Set([".git", ".paperclip", ".paperclip-runtime", "node_modules", ".pnpm-store", ".yarn", ".cache", + ".turbo", ".next", ".vite", ".vercel", "dist", "build", "coverage", "runtime-services", ".runtime"]); +const ARCHIVE_LIST_SKIPPED = new Set(["ENOENT", "ENOTDIR", "denied_path", "symlink_denied", "not_regular_file"]); +const archiveFail = (code: string): never => { throw new ArchiveReadFailure(code); }; + +function archiveDeniedPath(relative: string) { + const parts = relative.toLowerCase().split("/"); + const file = parts.at(-1) || ""; + return parts.some(part => ARCHIVE_DENIED_NAMES.has(part) || part === ".aws" || part === ".ssh") + || file === ".env" || file.startsWith(".env.") || /\.(pem|key|p12|pfx)$/.test(file) + || ["id_rsa", "id_ed25519", ".npmrc", ".pypirc", ".netrc", "kubeconfig"].includes(file) + || (parts.at(-2) === ".docker" && file === "config.json") || (parts.at(-2) === ".kube" && file === "config"); +} + +function archiveNormalize(relative: unknown, allowRoot = false): string { + if (allowRoot && !relative) return ""; + if (typeof relative !== "string" || !relative || relative.length > 4096 || /[\\\u0000-\u001f\u007f]/u.test(relative) + || path.isAbsolute(relative) || /^[a-z][a-z0-9+.-]*:/iu.test(relative) || relative.startsWith("~") + || relative.split("/").some(part => part === ".." || part === "." || part === "")) archiveFail("invalid_path"); + if (archiveDeniedPath(relative as string)) archiveFail("denied_path"); + return relative as string; +} + +function archiveConfined(archive: SavedWorkspaceArchive, relative: string) { + let current = ""; + for (const part of relative.split("/")) { + if (current && archive.entry(current)?.type !== "directory") archiveFail("ENOTDIR"); + current = current ? `${current}/${part}` : part; + const entry = archive.entry(current); + if (!entry) archiveFail("ENOENT"); + if (entry!.type === "symlink") archiveFail("symlink_denied"); + } + return archive.entry(relative)!; +} + +function archiveEntryInfo(archive: SavedWorkspaceArchive, relativePath: unknown) { + const relative = archiveNormalize(relativePath); + const entry = archiveConfined(archive, relative); + if (entry.type !== "directory" && entry.type !== "file") archiveFail("not_regular_file"); + return { entry, result: { kind: entry.type === "directory" ? "directory" : "file", relativePath: relative, + byteSize: entry.type === "file" ? entry.size : null, modifiedAt: archive.savedAt } as RemoteWorkspaceFileEntry }; +} + +async function archiveEntry(archive: SavedWorkspaceArchive, relativePath: unknown, action: "stat" | "list", withHash: boolean, + signal?: AbortSignal): Promise { + const { entry, result } = archiveEntryInfo(archive, relativePath); + if (entry.type === "file" && action === "stat") { + const chunks: Buffer[] = []; + for await (const chunk of await archive.read(entry, 0, Math.min(4096, entry.size), signal)) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + const sample = Buffer.concat(chunks); + let controls = 0, nul = false; + for (const byte of sample) { if (!byte) nul = true; if (byte < 32 && ![9, 10, 13].includes(byte)) controls++; } + result.text = !nul && (!sample.length || controls / sample.length < 0.02); + if (withHash) result.sha256 = entry.sha256!; + } + return result; +} + +function archiveList(archive: SavedWorkspaceArchive, requested: RemoteWorkspaceFileListInput) { + const start = archiveNormalize(requested.relativePath || "", true); + if (start && archiveConfined(archive, start).type !== "directory") archiveFail("not_directory"); + const maxScanned = Math.min(5000, Math.max(1, requested.maxScannedEntries)); + const maxDepth = Math.min(20, Math.max(0, requested.maxDepth)); + const limit = Math.min(100, Math.max(1, requested.limit)); + const offset = Math.min(10000, Math.max(0, requested.offset)); + const query = (requested.query || "").toLowerCase(); + let scannedCount = 0, truncated = false; + const entries: RemoteWorkspaceFileEntry[] = []; + const add = (relative: string) => { + try { + const { result } = archiveEntryInfo(archive, relative); + if (!query || relative.toLowerCase().includes(query)) entries.push(result); + } catch (error) { + if (!(error instanceof ArchiveReadFailure) || !ARCHIVE_LIST_SKIPPED.has(error.code)) throw error; + } + }; + const immediate = requested.mode === "all" && !query; + const queue = [{ relative: start, depth: 0 }]; + scan: while (queue.length) { + const current = queue.shift()!; + const sorted = archive.children(current.relative); + const passes = immediate ? [(entry: { type: string }) => entry.type === "directory", (entry: { type: string }) => entry.type !== "directory"] + : [() => true]; + for (const include of passes) for (const child of sorted) { + if (!include(child)) continue; + if (++scannedCount > maxScanned) { truncated = true; break scan; } + const relative = current.relative ? `${current.relative}/${child.name}` : child.name; + if (archiveDeniedPath(relative) || child.type === "symlink" || /[\\\u0000-\u001f\u007f]/u.test(relative)) continue; + if (child.type === "directory") { + if (immediate) add(relative); + else if (current.depth < maxDepth) queue.push({ relative, depth: current.depth + 1 }); + else truncated = true; + } else if (child.type === "file") add(relative); + if (immediate && entries.length > offset + limit) { truncated = true; break scan; } + } + } + if (requested.mode === "recent") entries.sort((a, b) => (b.modifiedAt || "").localeCompare(a.modifiedAt || "") + || a.relativePath.localeCompare(b.relativePath)); + return { entries: entries.slice(offset, offset + limit), scannedCount: Math.min(scannedCount, maxScanned), + truncated: truncated || entries.length > offset + limit }; +} + +export type SavedFileDownloadLimits = { maxReservedBytes: number; freeSpaceMarginBytes: number; availableBytes: () => Promise }; +const SAVED_FILE_DOWNLOAD_LIMITS: SavedFileDownloadLimits = { + maxReservedBytes: 4 * 1024 * 1024 * 1024, freeSpaceMarginBytes: 256 * 1024 * 1024, + availableBytes: async () => { const disk = await fs.statfs(os.tmpdir(), { bigint: true }); return disk.bavail * disk.bsize; }, +}; +let savedFileDownloadReservedBytes = 0; + +async function archiveRead(archive: SavedWorkspaceArchive, relativePath: string, expected: RemoteWorkspaceFileEntry, maxBytes: number, + signal?: AbortSignal): Promise { + const entry = archive.entry(relativePath); + if (!entry || entry.type !== "file" || entry.size !== expected.byteSize) throw fileFailure("file_changed"); + if (entry.size > maxBytes) throw fileFailure("too_large"); + const chunks: Buffer[] = [], hash = createHash("sha256"); + let size = 0; + for await (const chunk of await archive.read(entry, 0, entry.size, signal)) { + const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + size += bytes.length; + if (size > entry.size) throw fileFailure("file_changed"); + hash.update(bytes); chunks.push(bytes); + } + if (size !== entry.size || hash.digest("hex") !== entry.sha256) throw fileFailure("file_changed"); + return Buffer.concat(chunks); +} + +async function archiveDownload(archive: SavedWorkspaceArchive, relativePath: string, expected: RemoteWorkspaceFileEntry, + checkCurrent: () => Promise, limits: SavedFileDownloadLimits, signal?: AbortSignal): Promise { + const entry = archive.entry(relativePath); + if (!entry || entry.type !== "file" || entry.size !== expected.byteSize) throw fileFailure("file_changed"); + if (entry.size > limits.maxReservedBytes) throw fileFailure("too_large"); + if (savedFileDownloadReservedBytes + entry.size > limits.maxReservedBytes) throw fileFailure("workspace_unavailable"); + savedFileDownloadReservedBytes += entry.size; + let released = false; + const release = () => { if (!released) { released = true; savedFileDownloadReservedBytes -= entry.size; } }; + try { + if (await limits.availableBytes() < BigInt(savedFileDownloadReservedBytes + limits.freeSpaceMarginBytes)) { + throw fileFailure("workspace_unavailable"); + } + } catch (error) { release(); throw error; } + let directory: string | undefined; + const cleanup = async () => { + try { if (directory) await fs.rm(directory, { recursive: true, force: true }); } + finally { release(); } + }; + try { + directory = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-saved-file-")); + const file = path.join(directory, "file"); + await pipeline(await archive.read(entry, 0, entry.size, signal), createWriteStream(file, { flags: "wx", mode: 0o600 }), + ...(signal ? [{ signal }] : [])); + await verifyDownload(file, expected); + await checkCurrent(); + return { stream: createReadStream(file), byteSize: entry.size, cleanup }; + } catch (error) { await cleanup(); throw error; } +} + export function createRemoteWorkspaceFiles(db: Db, options: RemoteWorkspaceFileOptions = {}): RemoteWorkspaceFiles { return { async open(input) { @@ -270,6 +436,7 @@ export function createRemoteWorkspaceFiles(db: Db, options: RemoteWorkspaceFileO if (!authority) return null; const initialAuthority = JSON.stringify(authority); let saved: SavedWorkspaceFiles | undefined; + let archive: SavedWorkspaceArchive | undefined; let savedLocationReleased = false; const checkCurrent = async () => { input.signal?.throwIfAborted(); @@ -280,7 +447,7 @@ export function createRemoteWorkspaceFiles(db: Db, options: RemoteWorkspaceFileO // An extracted committed save no longer depends on the allocation. // Allow only its monotonic location removal; every logical field stays // pinned, and a foreign or reintroduced location remains a conflict. - if (saved && authority.location && currentAuthority?.location === null + if ((saved || archive) && authority.location && currentAuthority?.location === null && JSON.stringify({ ...currentAuthority, location: authority.location }) === initialAuthority) { savedLocationReleased = true; return; @@ -294,17 +461,41 @@ export function createRemoteWorkspaceFiles(db: Db, options: RemoteWorkspaceFileO let lease: EnvironmentLease | undefined; let source: WorkspaceFileSource; let root: string; + const savedReady = Boolean(options.loadSavedWorkspace || options.openSavedWorkspaceArchive); const loadSaved = async () => { - if (!authority.snapshotId || authority.dirty || authority.writerRunId || !options.loadSavedWorkspace) { + if (!authority.snapshotId || authority.dirty || authority.writerRunId || !savedReady) { throw fileFailure("workspace_unavailable"); } - saved = await options.loadSavedWorkspace({ companyId: input.companyId, workspaceId: input.workspaceId, snapshotId: authority.snapshotId, - ...(input.signal ? { signal: input.signal } : {}) }); + const scope = { companyId: input.companyId, workspaceId: input.workspaceId, snapshotId: authority.snapshotId, + ...(input.signal ? { signal: input.signal } : {}) }; + if (options.openSavedWorkspaceArchive) { + try { archive = await options.openSavedWorkspaceArchive(scope); } + catch (error) { + if (error instanceof HttpError) throw error; + if (error instanceof Error && error.message === "remote_workspace_archive_entry_limit") throw fileFailure("too_large"); + if (error instanceof Error && /^remote_workspace_(snapshot|archive)_/.test(error.message)) throw fileFailure("workspace_unavailable"); + throw error; + } + return { root: "", source: { kind: "saved", savedAt: archive.savedAt, snapshotId: authority.snapshotId } as WorkspaceFileSource }; + } + saved = await options.loadSavedWorkspace!(scope); return { root: saved.root, source: { kind: "saved", savedAt: saved.savedAt, snapshotId: authority.snapshotId } as WorkspaceFileSource }; }; + const extractedRoot = async () => { + if (saved) return saved.root; + if (!options.loadSavedWorkspace || !authority.snapshotId) throw fileFailure("changed_unavailable"); + try { + saved = await options.loadSavedWorkspace({ companyId: input.companyId, workspaceId: input.workspaceId, snapshotId: authority.snapshotId, + ...(input.signal ? { signal: input.signal } : {}) }); + } catch (error) { + if (error instanceof HttpError && error.status === 422) throw fileFailure("changed_unavailable"); + throw error; + } + return saved.root; + }; // A final save clears dirty state and writer authority. Its accepted // bytes do not depend on an allocation that can already be closing. - if (authority.snapshotId && !authority.dirty && !authority.writerRunId && options.loadSavedWorkspace) { + if (authority.snapshotId && !authority.dirty && !authority.writerRunId && savedReady) { ({ root, source } = await loadSaved()); } else if (authority.location) { const [leaseRow] = await db.select().from(environmentLeases).where(and( @@ -330,16 +521,29 @@ export function createRemoteWorkspaceFiles(db: Db, options: RemoteWorkspaceFileO ({ root, source } = await loadSaved()); } let closed = false; - const close = async () => { if (closed) return; closed = true; await saved?.cleanup(); }; + const close = async () => { if (closed) return; closed = true; archive?.close(); await saved?.cleanup(); }; const execute = async (action: "stat" | "list", args: Record): Promise => { if (closed) throw fileFailure("workspace_unavailable"); await checkCurrent(); - const payload = JSON.stringify({ root, action, ...args }); + const changedList = action === "list" && (args.options as RemoteWorkspaceFileListInput | undefined)?.mode === "changed"; + if (archive && !changedList) { + let value: unknown; + try { + value = action === "stat" + ? await archiveEntry(archive, args.relativePath, "stat", args.withHash === true, input.signal) + : archiveList(archive, args.options as RemoteWorkspaceFileListInput); + } catch (error) { throw fileFailure(error instanceof ArchiveReadFailure ? error.code : "workspace_unavailable"); } + await checkCurrent(); + try { return parseHelperResult(action, value, args.withHash === true) as T; } + catch { throw fileFailure("workspace_unavailable"); } + } + const helperRoot = archive ? await extractedRoot() : root; + const payload = JSON.stringify({ root: helperRoot, action, ...args }); let stdout: string; if (saved) { try { ({ stdout } = await execFileAsync(process.execPath, ["--input-type=commonjs", "-e", REMOTE_WORKSPACE_FILES_SCRIPT, payload], { - cwd: root, timeout: READ_TIMEOUT_MS, maxBuffer: MAX_HELPER_OUTPUT_BYTES, + cwd: helperRoot, timeout: READ_TIMEOUT_MS, maxBuffer: MAX_HELPER_OUTPUT_BYTES, signal: input.signal, env: { ...process.env, NODE_OPTIONS: "", NODE_PATH: "" }, })); } catch (error) { throw fileFailure(String((error as { stderr?: string }).stderr ?? "workspace_unavailable").trim()); } @@ -360,6 +564,8 @@ export function createRemoteWorkspaceFiles(db: Db, options: RemoteWorkspaceFileO const download = async (relativePath: string): Promise => { const expected = await execute("stat", { relativePath, withHash: true }); if (expected.kind !== "file") throw fileFailure("not_regular_file"); + if (archive) return archiveDownload(archive, relativePath, expected, checkCurrent, + { ...SAVED_FILE_DOWNLOAD_LIMITS, ...options.savedFileDownloadLimits }, input.signal); if (saved) { await checkCurrent(); const file = path.join(root, ...relativePath.split("/")); @@ -384,6 +590,11 @@ export function createRemoteWorkspaceFiles(db: Db, options: RemoteWorkspaceFileO const expected = await stat(relativePath); if (expected.kind !== "file") throw fileFailure("not_regular_file"); if (expected.byteSize! > readOptions.maxBytes) throw fileFailure("too_large"); + if (archive) { + const bytes = await archiveRead(archive, relativePath, expected, readOptions.maxBytes, input.signal); + await checkCurrent(); + return bytes; + } const transfer = await download(relativePath); try { const chunks: Buffer[] = []; diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index 8ec543041c1..2e7f38fe228 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -7,7 +7,7 @@ import { link, lstat, mkdtemp, open, realpath, rm } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import { pipeline } from "node:stream/promises"; -import { Transform } from "node:stream"; +import { pipeline as streamPipeline, Readable, Transform } from "node:stream"; import { isDeepStrictEqual } from "node:util"; import { and, eq } from "drizzle-orm"; import { executionWorkspaceSnapshots, type Db } from "@paperclipai/db"; @@ -434,3 +434,98 @@ async function loadWorkspaceSnapshotArchiveTimed(input: Parameters; + close(): void; +} + +export async function openWorkspaceSnapshotRanges(input: { + storage: StorageService; reference: WorkspaceSnapshotRef; signal?: AbortSignal; idleTimeoutMs?: number; +}): Promise { + input.signal?.throwIfAborted(); + const idleTimeoutMs = input.idleTimeoutMs ?? DOWNLOAD_IDLE_TIMEOUT_MS; + const ref = input.reference, wrapped = await snapshotKey(ref); + if (ref.byteSize !== GCM_HEADER_BYTES + wrapped.plaintextBytes + GCM_TAG_BYTES + || wrapped.plaintextBytes > GCM_MAX_PLAINTEXT_BYTES) failure("object_size_mismatch"); + const header = Buffer.concat(await collect(await openObjectRange(input.storage, ref, 0, GCM_HEADER_BYTES, idleTimeoutMs, input.signal), + GCM_HEADER_BYTES)); + if (header.length !== GCM_HEADER_BYTES) failure("range_truncated"); + if (!header.subarray(0, MAGIC.length).equals(MAGIC)) failure("ciphertext_header_invalid"); + const iv = header.subarray(MAGIC.length); + const key = Buffer.from(wrapped.dataKey, "base64"); + if (key.length !== 32) { key.fill(0); failure("key_invalid"); } + let closed = false; + return { + plaintextBytes: wrapped.plaintextBytes, + plaintextSha256: wrapped.plaintextSha256, + async read(start, length, signal) { + if (closed) failure("ranges_closed"); + if (!Number.isSafeInteger(start) || !Number.isSafeInteger(length) || start < 0 || length < 0 + || start + length > wrapped.plaintextBytes) failure("range_invalid"); + if (length === 0) return Readable.from([]); + const block = Math.floor(start / CTR_BLOCK_BYTES), skip = start - block * CTR_BLOCK_BYTES; + const counter = Buffer.alloc(16); iv.copy(counter, 0); counter.writeUInt32BE(2 + block, 12); + const ciphertext = await openObjectRange(input.storage, ref, GCM_HEADER_BYTES + block * CTR_BLOCK_BYTES, + skip + length, idleTimeoutMs, signal); + let skipped = 0, emitted = 0; + const bounded = new Transform({ transform(bytes: Buffer, _encoding, callback) { + let part = bytes; + if (skipped < skip) { const drop = Math.min(skip - skipped, part.length); skipped += drop; part = part.subarray(drop); } + const room = length - emitted; + if (part.length > room) part = part.subarray(0, room); + emitted += part.length; + callback(null, part.length ? part : undefined); + }, flush(callback) { callback(emitted === length ? null : new Error("remote_workspace_snapshot_range_truncated")); } }); + return streamPipeline(ciphertext, createDecipheriv("aes-256-ctr", key, counter), bounded, () => undefined); + }, + close() { closed = true; key.fill(0); }, + }; +} + +async function openObjectRange(storage: StorageService, ref: WorkspaceSnapshotRef, start: number, length: number, + idleTimeoutMs: number, signal?: AbortSignal): Promise { + const stalled = new Error("remote_workspace_snapshot_download_stalled"); + const idle = new AbortController(); + let timer = setTimeout(() => idle.abort(stalled), idleTimeoutMs); + timer.unref(); + const combined = signal ? AbortSignal.any([signal, idle.signal]) : idle.signal; + let object: Awaited>; + try { + object = await sendThroughStorageOutage(() => storage.getObject(ref.companyId, ref.objectKey, { + range: { start, end: start + length - 1 }, signal: combined }), combined); + } catch (error) { clearTimeout(timer); throw idle.signal.aborted ? stalled : error; } + let received = 0; + const guard = new Transform({ transform(bytes: Buffer, _encoding, callback) { + timer.refresh(); + received += bytes.length; + callback(received > length ? new Error("remote_workspace_snapshot_object_size_mismatch") : null, bytes); + } }); + const output = streamPipeline(object.stream, guard, () => clearTimeout(timer)); + clearTimeout(timer); + timer = setTimeout(() => output.destroy(stalled), idleTimeoutMs); + timer.unref(); + output.once("close", () => clearTimeout(timer)); + return output; +} + +async function collect(stream: Readable, limit: number): Promise { + const chunks: Buffer[] = []; + let size = 0; + try { + for await (const chunk of stream) { + const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + size += bytes.length; + if (size > limit) failure("object_size_mismatch"); + chunks.push(bytes); + } + } finally { stream.destroy(); } + return chunks; +} diff --git a/server/src/services/saved-workspace-archive.ts b/server/src/services/saved-workspace-archive.ts new file mode 100644 index 00000000000..c5263d581e1 --- /dev/null +++ b/server/src/services/saved-workspace-archive.ts @@ -0,0 +1,196 @@ +import { createHash } from "node:crypto"; +import type { Readable } from "node:stream"; +import { and, eq } from "drizzle-orm"; +import { executionWorkspaceSnapshots, type Db } from "@paperclipai/db"; +import type { StorageService } from "../storage/types.js"; +import { getStorageService } from "../storage/index.js"; +import { getWorkspaceSnapshot, openWorkspaceSnapshotRanges } from "./remote-workspace-snapshot-storage.js"; + +const ARCHIVE_MAGIC = Buffer.from("PCWS1\n"); +const ARCHIVE_SCHEMA = "paperclip.remote-workspace-archive.v1"; +const MAX_MANIFEST_BYTES = 64 * 1024 * 1024; +const MAX_INDEX_ENTRIES = 500_000; +const INDEX_ENTRY_BYTES = 256; +const DEFAULT_INDEX_CACHE_BYTES = 256 * 1024 * 1024; +const MAX_CONCURRENT_INDEX_BUILDS = 2; +let activeIndexBuilds = 0; +const indexBuildWaiters: Array<() => void> = []; + +function untilAborted(work: Promise, signal?: AbortSignal): Promise { + if (!signal) return work; + if (signal.aborted) return Promise.reject(signal.reason); + return new Promise((resolve, reject) => { + const abort = () => reject(signal.reason); + signal.addEventListener("abort", abort, { once: true }); + work.then(value => { signal.removeEventListener("abort", abort); resolve(value); }, + error => { signal.removeEventListener("abort", abort); reject(error); }); + }); +} + +async function withIndexBuildSlot(work: () => Promise, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); + if (activeIndexBuilds >= MAX_CONCURRENT_INDEX_BUILDS) { + await new Promise((resolve, reject) => { + const granted = () => { signal?.removeEventListener("abort", abort); resolve(); }; + const abort = () => { + const waiting = indexBuildWaiters.indexOf(granted); + if (waiting >= 0) indexBuildWaiters.splice(waiting, 1); + reject(signal!.reason); + }; + signal?.addEventListener("abort", abort, { once: true }); + indexBuildWaiters.push(granted); + }); + } else activeIndexBuilds++; + try { return await work(); } + finally { + const next = indexBuildWaiters.shift(); + if (next) next(); else activeIndexBuilds--; + } +} + +export type SavedArchiveEntry = { + path: string; name: string; type: "directory" | "file" | "symlink"; size: number; sha256: string | null; offset: number; +}; + +export interface SavedWorkspaceArchive { + snapshotId: string; + savedAt: string; + byteSize: number; + entry(relativePath: string): SavedArchiveEntry | undefined; + children(relativePath: string): readonly SavedArchiveEntry[]; + cachedIndexBytes(): number; + read(entry: SavedArchiveEntry, start: number, length: number, signal?: AbortSignal): Promise; + close(): void; +} + +export type OpenSavedWorkspaceArchive = (input: { + companyId: string; workspaceId: string; snapshotId: string; signal?: AbortSignal; +}) => Promise; + +type ArchiveIndex = { entries: Map; children: Map; charge: number }; + +function invalid(reason: string): never { throw new Error(`remote_workspace_archive_${reason}`); } + +const unsafePath = (relative: unknown) => typeof relative !== "string" || !relative || relative.length > 4096 + || relative.startsWith("/") || /[\\\u0000-\u001f\u007f]/u.test(relative) + || relative.split("/").some(part => !part || part === "." || part === ".."); + +function buildIndex(manifestBytes: Buffer, archiveBytes: number, maxEntries: number): ArchiveIndex { + let manifest: { schema?: unknown; entries?: unknown }; + try { manifest = JSON.parse(manifestBytes.toString("utf8")); } catch { invalid("manifest_mismatch"); } + if (manifest.schema !== ARCHIVE_SCHEMA || !Array.isArray(manifest.entries)) invalid("manifest_mismatch"); + if (manifest.entries.length > maxEntries) invalid("entry_limit"); + const entries = new Map(), children = new Map([["", []]]); + let offset = ARCHIVE_MAGIC.length + 4 + manifestBytes.length; + for (const raw of manifest.entries as Array>) { + const relative = raw?.path; + if (unsafePath(relative) || entries.has(relative as string)) invalid("invalid_entry"); + const entryPath = relative as string, parent = entryPath.includes("/") ? entryPath.slice(0, entryPath.lastIndexOf("/")) : ""; + if (parent && entries.get(parent)?.type !== "directory") invalid("invalid_parent"); + let entry: SavedArchiveEntry; + if (raw.type === "file") { + if (!Number.isSafeInteger(raw.size) || (raw.size as number) < 0 || typeof raw.sha256 !== "string" || !/^[a-f0-9]{64}$/.test(raw.sha256)) { + invalid("invalid_file"); + } + entry = { path: entryPath, name: entryPath.slice(parent ? parent.length + 1 : 0), type: "file", size: raw.size as number, + sha256: raw.sha256 as string, offset }; + offset += entry.size; + } else if (raw.type === "directory" || raw.type === "symlink") { + entry = { path: entryPath, name: entryPath.slice(parent ? parent.length + 1 : 0), type: raw.type, size: 0, sha256: null, offset: 0 }; + if (raw.type === "directory") children.set(entryPath, []); + } else invalid("invalid_type"); + entries.set(entryPath, entry); + children.get(parent)!.push(entry); + } + if (!Number.isSafeInteger(offset) || offset !== archiveBytes) invalid("size_mismatch"); + for (const list of children.values()) list.sort((a, b) => a.name.localeCompare(b.name)); + return { entries, children, charge: manifestBytes.length + entries.size * INDEX_ENTRY_BYTES }; +} + +async function readAll(stream: Readable, limit: number): Promise { + const chunks: Buffer[] = []; + let size = 0; + try { + for await (const chunk of stream) { + const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + size += bytes.length; + if (size > limit) invalid("truncated"); + chunks.push(bytes); + } + } finally { stream.destroy(); } + return Buffer.concat(chunks); +} + +export function createSavedWorkspaceArchiveOpener(db: Db, storage: StorageService = getStorageService(), options: { + indexCacheBytes?: number; idleTimeoutMs?: number; maxIndexEntries?: number; +} = {}): OpenSavedWorkspaceArchive { + const maxEntries = Math.min(options.maxIndexEntries ?? MAX_INDEX_ENTRIES, MAX_INDEX_ENTRIES); + const cacheBudget = options.indexCacheBytes ?? DEFAULT_INDEX_CACHE_BYTES; + const cache = new Map(), building = new Map>(); + let cachedBytes = 0; + const remember = (key: string, index: ArchiveIndex) => { + const existing = cache.get(key); + if (existing) { cache.delete(key); cachedBytes -= existing.charge; } + if (index.charge > cacheBudget) return; + cache.set(key, index); cachedBytes += index.charge; + for (const [oldest, value] of cache) { + if (cachedBytes <= cacheBudget) break; + cache.delete(oldest); cachedBytes -= value.charge; + } + }; + return async input => { + input.signal?.throwIfAborted(); + const [row] = await db.select({ updatedAt: executionWorkspaceSnapshots.updatedAt }).from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, input.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, input.workspaceId), + eq(executionWorkspaceSnapshots.id, input.snapshotId), eq(executionWorkspaceSnapshots.status, "committed"))); + const saved = row ? await getWorkspaceSnapshot(db, input) : null; + if (!row || !saved) throw new Error("remote_workspace_snapshot_unavailable"); + const ranges = await openWorkspaceSnapshotRanges({ storage, reference: saved, signal: input.signal, + ...(options.idleTimeoutMs ? { idleTimeoutMs: options.idleTimeoutMs } : {}) }); + try { + const key = `${saved.companyId}/${saved.workspaceId}/${saved.id}/${saved.manifestDigest}`; + let index: ArchiveIndex | undefined; + for (;;) { + index = cache.get(key); + if (index) { cache.delete(key); cache.set(key, index); break; } + const shared = building.get(key); + if (shared) { + index = await untilAborted(shared.catch(() => undefined), input.signal); + if (index) break; + continue; + } + const build = withIndexBuildSlot(async () => { + const prefixLength = ARCHIVE_MAGIC.length + 4; + if (ranges.plaintextBytes < prefixLength) invalid("truncated"); + const prefix = await readAll(await ranges.read(0, prefixLength, input.signal), prefixLength); + if (!prefix.subarray(0, ARCHIVE_MAGIC.length).equals(ARCHIVE_MAGIC)) invalid("invalid_header"); + const manifestSize = prefix.readUInt32BE(ARCHIVE_MAGIC.length); + if (manifestSize > MAX_MANIFEST_BYTES || prefixLength + manifestSize > ranges.plaintextBytes) invalid("manifest_too_large"); + const manifest = await readAll(await ranges.read(prefixLength, manifestSize, input.signal), manifestSize); + if (manifest.length !== manifestSize || createHash("sha256").update(manifest).digest("hex") !== saved.manifestDigest) { + invalid("manifest_mismatch"); + } + return buildIndex(manifest, ranges.plaintextBytes, maxEntries); + }, input.signal); + building.set(key, build); + try { index = await build; remember(key, index); } + finally { if (building.get(key) === build) building.delete(key); } + break; + } + const built = index; + return { + snapshotId: saved.id, + savedAt: row.updatedAt.toISOString(), + byteSize: saved.byteSize, + entry: relativePath => built.entries.get(relativePath), + children: relativePath => built.children.get(relativePath) ?? [], + cachedIndexBytes: () => cachedBytes, + read: (entry, start, length, signal) => { + if (entry.type !== "file" || start < 0 || length < 0 || start + length > entry.size) invalid("range_invalid"); + return ranges.read(entry.offset + start, length, signal); + }, + close: () => ranges.close(), + }; + } catch (error) { ranges.close(); throw error; } + }; +}