mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(workspaces): read saved workspace files in place instead of extracting the whole archive (#20442)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - When a cloud task finishes, its workspace is saved as an encrypted
snapshot and its sandbox is released. The board's Files view then reads
that saved copy
> - Saved Files access downloaded, decrypted and extracted the whole
archive for every request, and refused archives over 512 MiB with
`too_large`
> - A finished task in a Paperclip-sized repository saves a snapshot of
about 730 MiB, so its files could not be listed or opened at all
> - The archive format already records each file's size and SHA-256, and
the snapshot records the manifest digest, so single files can be read
and verified without the rest of the archive
> - This pull request reads the saved archive in place, one verified
file at a time
> - The benefit is that users can open any file of a large finished
workspace quickly, with bounded memory and no whole-archive transfer
## Linked Issues or Issue Description
**What happened?**
In restart-chaos waves on Daytona with a real Paperclip clone, the
committed workspace snapshots were 763,944,535 bytes. After the sandbox
was released, every saved Files read returned 422 `too_large` (506 reads
in total), because `createSavedWorkspaceLoader` refuses archives over
`maxArchiveBytes` (512 MiB). The board Files view uses the same loader,
so it fails the same way.
**Expected behavior**
A user can list and open any single file of a saved workspace of
Paperclip-repository size without the server extracting or buffering the
whole archive. Memory and time stay bounded per request, and integrity
is still verified.
**Steps to reproduce**
1. Run a cloud task whose workspace is a large repository, so the final
snapshot exceeds 512 MiB.
2. Let the task finish and its sandbox be released.
3. Open the task's Files view, or call `GET
/api/issues/:id/file-resources/content?workspace=execution&path=…`: it
returns 422 `too_large`.
**Paperclip version or commit**
`master` at f73ff4ace9.
## What Changed
- `openWorkspaceSnapshotRanges`
(`remote-workspace-snapshot-storage.ts`):
- range-reads the snapshot object's header;
- decrypts any plaintext byte range with AES-256-CTR, starting from the
AES-GCM counter block for that range;
- checks that the object size matches the wrapped key's plaintext size.
- New `saved-workspace-archive.ts`:
- range-reads and decrypts the archive header and manifest;
- verifies the manifest's SHA-256 against the snapshot's recorded
`manifestDigest`;
- applies the restore path's manifest checks (safe paths, parent
directories, file sizes and hashes, total size) and computes each file's
offset;
- keeps parsed manifests of recent snapshots within a 128 MiB budget.
- `remote-workspace-files.ts`: with the new `openSavedWorkspaceArchive`
option, saved stat, list, read and download use the verified manifest,
not the extracted-directory helper.
- Path normalization, denied paths, symlink and missing-path errors,
traversal, paging, truncation and text detection mirror the helper
exactly.
- A read or download decrypts only that file into a private temporary
file and verifies its SHA-256 before serving it.
- The changed-files view needs repository status, so it still uses the
bounded extraction. A saved copy beyond its limits now reports
`changed_unavailable`, a specific code, while its other views stay
available.
- `app.ts` and the support inspection route pass the new opener.
- `doc/SPEC-implementation.md` describes in-place saved reads and the
remaining changed-files limits.
## Verification
- Red before the fix, green after: `reads a saved workspace above the
extraction cap in place, one verified file at a time`. With extraction
only (master's Files service), the capped snapshot returns 422
`too_large`. With the archive reader, preview, list and a 256 KiB
download succeed through the ordinary routes. Every storage read is a
byte-range read, and no extraction directory is created.
- Parity: `answers saved stat and list exactly like the extracted
helper` compares archive-backed results with the extracted-helper path
on the same snapshot.
- Stat covers files, directories, nested and binary files, missing
paths, paths through a file, symlinks, denied names, and invalid paths.
- List covers all mode, a subdirectory, paging, queries, a scan limit, a
non-directory, and a denied directory.
- In recent mode, the entries match. Order is by path, because a saved
copy has no modification times; the extracted copy's mtimes were just
its extraction times.
- Integrity: `refuses a saved file whose stored bytes no longer match
its verified manifest`. Tampered stored bytes return 422 `file_changed`,
and no bytes are served.
- `keeps changed files on bounded extraction and names its limit…`: over
the cap, the changed view returns `changed_unavailable` and the other
views still work.
- The existing location-clearing preview test now checks for range reads
and an empty request temp directory, not for an extraction directory.
- Measured locally, with local-disk storage and a 749,006,840-byte
snapshot (88 × 8 MiB pack file, a 10 MiB file, 2,000 small files):
| path | time |
| --- | --- |
| Old: download, decrypt and extract all, once per request | 2,338 ms,
plus a full 749 MB transfer from S3 |
| New, cold: open (header and manifest range reads) | 7 ms; open and
read a small file: 13 ms |
| New, warm (cached manifest) | open 2 ms; open and read: 6 ms |
| New: list 100 entries and download a 10 MiB file | 23 ms |
- `remote-workspace-file-resources`, `remote-workspaces.integration`,
`remote-workspace-save-concurrency`, and
`remote-workspace-snapshot-retention` pass (212 tests).
- `PAPERCLIP_TYPECHECK_LOW_MEMORY=1 pnpm typecheck` in `server` passed.
- Not yet verified: the Daytona R1 wave's file checks after release. The
chaos harness owner will run them on a build with this change.
## Risks
- Range decryption bypasses GCM's whole-object tag. Integrity instead
comes from the manifest's SHA-256 matching the digest recorded in the
database, and from each file's SHA-256 in that manifest. A file is
verified before it is served. Text detection on stat samples the first 4
KiB without verifying the whole file. It only sets a display hint, and
content is always verified before it is shown.
- Saved files carry no modification times, so the recent view of a saved
copy is ordered by path, where it was previously in effectively random
extraction order.
- The changed-files view of a saved copy over 512 MiB still reports
`changed_unavailable`, because it needs a full extraction to run `git
status`.
- No schema or API change.
## Model Used
- Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use
(shell, file edits, local benchmarks).
## Checklist
- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
5ab98b0063
commit
120017e39d
7 files changed
+826
-32
No files matched your search
+30
-12
@@ -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.
|
||||
|
||||
@@ -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<void>) {
|
||||
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<unknown>) => {
|
||||
try {
|
||||
const value = JSON.parse(JSON.stringify(await work()));
|
||||
const strip = (entry: Record<string, unknown>) => { 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<Parameters<typeof createRemoteWorkspaceFiles>[1]>["savedFileDownloadLimits"]) =>
|
||||
createRemoteWorkspaceFiles(db, { openSavedWorkspaceArchive: opener, savedFileDownloadLimits: limits });
|
||||
const attempt = async (limits: Parameters<typeof files>[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<void>(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<import("node:stream").Readable> = [];
|
||||
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<void>(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<void>(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(); }
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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 };
|
||||
|
||||
@@ -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<SavedWorkspaceFiles>;
|
||||
openSavedWorkspaceArchive?: OpenSavedWorkspaceArchive;
|
||||
savedFileDownloadLimits?: Partial<SavedFileDownloadLimits>;
|
||||
}
|
||||
|
||||
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<RemoteWorkspaceFileEntry> {
|
||||
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<bigint> };
|
||||
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<Buffer> {
|
||||
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<void>, limits: SavedFileDownloadLimits, signal?: AbortSignal): Promise<RemoteWorkspaceDownload> {
|
||||
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 <T>(action: "stat" | "list", args: Record<string, unknown>): Promise<T> => {
|
||||
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<RemoteWorkspaceDownload> => {
|
||||
const expected = await execute<RemoteWorkspaceFileEntry>("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[] = [];
|
||||
|
||||
@@ -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<typeof loadWo
|
||||
} finally { await rm(staged, { force: true }); }
|
||||
} finally { await rm(scratch, { recursive: true, force: true }); }
|
||||
}
|
||||
|
||||
const GCM_HEADER_BYTES = MAGIC.length + 12;
|
||||
const GCM_TAG_BYTES = 16;
|
||||
const CTR_BLOCK_BYTES = 16;
|
||||
const GCM_MAX_PLAINTEXT_BYTES = (2 ** 32 - 2) * CTR_BLOCK_BYTES;
|
||||
|
||||
export interface WorkspaceSnapshotRanges {
|
||||
plaintextBytes: number;
|
||||
plaintextSha256: string;
|
||||
read(start: number, length: number, signal?: AbortSignal): Promise<Readable>;
|
||||
close(): void;
|
||||
}
|
||||
|
||||
export async function openWorkspaceSnapshotRanges(input: {
|
||||
storage: StorageService; reference: WorkspaceSnapshotRef; signal?: AbortSignal; idleTimeoutMs?: number;
|
||||
}): Promise<WorkspaceSnapshotRanges> {
|
||||
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<Readable> {
|
||||
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<ReturnType<StorageService["getObject"]>>;
|
||||
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<Buffer[]> {
|
||||
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;
|
||||
}
|
||||
@@ -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<T>(work: Promise<T>, signal?: AbortSignal): Promise<T> {
|
||||
if (!signal) return work;
|
||||
if (signal.aborted) return Promise.reject(signal.reason);
|
||||
return new Promise<T>((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<T>(work: () => Promise<T>, signal?: AbortSignal): Promise<T> {
|
||||
signal?.throwIfAborted();
|
||||
if (activeIndexBuilds >= MAX_CONCURRENT_INDEX_BUILDS) {
|
||||
await new Promise<void>((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<Readable>;
|
||||
close(): void;
|
||||
}
|
||||
|
||||
export type OpenSavedWorkspaceArchive = (input: {
|
||||
companyId: string; workspaceId: string; snapshotId: string; signal?: AbortSignal;
|
||||
}) => Promise<SavedWorkspaceArchive>;
|
||||
|
||||
type ArchiveIndex = { entries: Map<string, SavedArchiveEntry>; children: Map<string, SavedArchiveEntry[]>; 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<string, SavedArchiveEntry>(), children = new Map<string, SavedArchiveEntry[]>([["", []]]);
|
||||
let offset = ARCHIVE_MAGIC.length + 4 + manifestBytes.length;
|
||||
for (const raw of manifest.entries as Array<Record<string, unknown>>) {
|
||||
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<Buffer> {
|
||||
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<string, ArchiveIndex>(), building = new Map<string, Promise<ArchiveIndex>>();
|
||||
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; }
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user