packages/forkhorse-runner/src/snapshot.ts

1import { WorkerEntrypoint } from "cloudflare:workers";
2import type { ForkRepository, QuickFileStat } from "./types";
3
4export const QUICK_READ_BYTES = 2 * 1024 * 1024;
5export const QUICK_LIST_PATHS = 2000;
6export const QUICK_STAT_BYTES = 8 * 1024;
7const MAX_ENTRIES = 10_000;
8
9/** Literal repository paths only. Reject aliases instead of normalizing into another file. */
10function pathParts(path: unknown, prefix = false): string[] {
11  if (typeof path !== "string" || path.length > 4096 || path.startsWith("/") || path.includes("\\") ||
12      [...path].some(c => c.charCodeAt(0) < 32 || c.charCodeAt(0) === 127)) throw new Error("Invalid path");
13  if (prefix && path.endsWith("/")) path = path.slice(0, -1);
14  if (prefix && path === "") return [];
15  const parts = (path as string).split("/");
16  if (parts.length > 64 || parts.some(p => !p || p === "." || p === ".." || p.toLowerCase() === ".git"))
17    throw new Error("Invalid path");
18  return parts;
19}
20
21/** Host-owned state: no token operations and no moving refs. Close fences in-flight RPC too. */
22export class Snapshot {
23  private active = true;
24  private bytes = 0;
25  private listed = 0;
26  private entries = 0;
27  private calls = 0;
28  private root?: Promise<string>;
29  private trees = new Map<string, Promise<ArtifactsTreeEntry[]>>();
30  constructor(private repo: Pick<ForkRepository, "readCommit" | "readTree" | "readBlob">,
31    private commit: string, private expiresAt = Infinity) {}
32  close(): void { this.active = false; }
33  private guard(): void {
34    if (!this.active || Date.now() >= this.expiresAt) throw new Error("Snapshot closed");
35  }
36  private async tree(hash: string): Promise<ArtifactsTreeEntry[]> {
37    this.guard();
38    if (!/^[a-f0-9]{40}$/.test(hash)) throw new Error("Invalid tree hash");
39    // Immutable, validated metadata is shared by list/read/stat within this capability.
40    // Cache the promise before awaiting so concurrent calls cannot duplicate traversal.
41    let pending = this.trees.get(hash);
42    if (!pending) {
43      pending = this.loadTree(hash);
44      this.trees.set(hash, pending);
45    }
46    const tree = await pending;
47    this.guard();
48    return tree;
49  }
50  private async loadTree(hash: string): Promise<ArtifactsTreeEntry[]> {
51    const tree = await this.repo.readTree(hash);
52    this.guard();
53    if (!tree || (this.entries += tree.length) > MAX_ENTRIES) throw new Error("Snapshot tree limit");
54    const names = new Set<string>();
55    for (const e of tree) {
56      if (pathParts(e.name).length !== 1 || names.has(e.name) || !/^[a-f0-9]{40}$/.test(e.hash)) throw new Error("Invalid tree");
57      names.add(e.name);
58    }
59    return tree;
60  }
61  private async rootHash(): Promise<string> {
62    this.guard();
63    // Reserve synchronously: concurrent read calls cannot start separate traversals of a ref.
64    if (!/^[a-f0-9]{40}$/.test(this.commit)) throw new Error("Invalid commit");
65    this.root ??= this.repo.readCommit(this.commit).then(c => {
66      this.guard();
67      if (!c || c.hash !== this.commit) throw new Error("Snapshot unavailable");
68      return c.treeHash;
69    });
70    return this.root;
71  }
72  private async entry(parts: string[]): Promise<ArtifactsTreeEntry | null> {
73    let hash = await this.rootHash();
74    for (let i = 0; i < parts.length; i++) {
75      const entry = (await this.tree(hash)).find(e => e.name === parts[i]);
76      if (!entry) return null;
77      if (i === parts.length - 1) return entry;
78      if (entry.type !== "tree") throw new Error("Not a directory");
79      hash = entry.hash;
80    }
81    return { name: "", hash, type: "tree", mode: "40000" };
82  }
83  async read(path: unknown): Promise<string | null> {
84    try {
85      this.guard();
86      if (++this.calls > MAX_ENTRIES) throw new Error("Snapshot call limit");
87      const entry = await this.entry(pathParts(path));
88      if (!entry) return null;
89      if (!((entry.type === "blob" && entry.mode === "100644") || (entry.type === "exec" && entry.mode === "100755"))) throw new Error("Not a regular file");
90      this.guard();
91      const blob = await this.repo.readBlob(entry.hash);
92      this.guard();
93      if (!blob || (this.bytes += blob.size) > QUICK_READ_BYTES) throw new Error("Snapshot read limit");
94      const bytes = await blob.arrayBuffer();
95      this.guard();
96      if (bytes.byteLength !== blob.size) throw new Error("Invalid blob");
97      return new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }).decode(bytes);
98    } catch {
99      // The module may catch this error and return it; never forward provider text or stacks.
100      throw new Error("Snapshot read unavailable or limit exceeded.");
101    }
102  }
103  async stat(path: unknown): Promise<QuickFileStat | null> {
104    try {
105      this.guard();
106      if (++this.calls > MAX_ENTRIES) throw new Error("Snapshot call limit");
107      const entry = await this.entry(pathParts(path));
108      if (!entry) return null;
109      if (!((entry.type === "blob" && entry.mode === "100644") || (entry.type === "exec" && entry.mode === "100755"))) throw new Error("Not a regular file");
110      this.guard();
111      const blob = await this.repo.readBlob(entry.hash);
112      this.guard();
113      if (!blob) throw new Error("Missing blob");
114      const sampleSize = Math.min(blob.size, QUICK_STAT_BYTES);
115      // Reserve before awaiting, sharing the read budget even across concurrent calls.
116      if ((this.bytes += sampleSize) > QUICK_READ_BYTES) throw new Error("Snapshot read limit");
117      const sample = await blob.slice(0, sampleSize).arrayBuffer();
118      this.guard();
119      if (sample.byteLength !== sampleSize) throw new Error("Invalid blob");
120      return { size: blob.size, binary: new Uint8Array(sample).includes(0) };
121    } catch { throw new Error("Snapshot stat unavailable or limit exceeded."); }
122  }
123  async list(prefix: unknown): Promise<string[]> {
124    try {
125      this.guard();
126      if (++this.calls > MAX_ENTRIES) throw new Error("Snapshot call limit");
127      const parts = pathParts(prefix, true);
128      const baseParts = typeof prefix === "string" && prefix.endsWith("/") ? parts : parts.slice(0, -1);
129      const entry = await this.entry(baseParts);
130      if (!entry) return [];
131      const paths: string[] = [];
132      const visit = async (e: ArtifactsTreeEntry, path: string, depth: number): Promise<void> => {
133        this.guard();
134        if (depth > 64) throw new Error("Snapshot depth limit");
135        if (e.type === "tree") {
136          for (const child of await this.tree(e.hash)) await visit(child, path ? `${path}/${child.name}` : child.name, depth + 1);
137        } else if ((e.type === "blob" && e.mode === "100644") || (e.type === "exec" && e.mode === "100755")) {
138          if (!path.startsWith(prefix as string)) return;
139          if (++this.listed > QUICK_LIST_PATHS) throw new Error("Snapshot list limit");
140          paths.push(path);
141        }
142      };
143      await visit(entry, baseParts.join("/"), baseParts.length);
144      this.guard();
145      return paths.toSorted();
146    } catch { throw new Error("Snapshot list unavailable or limit exceeded."); }
147  }
148}
149
150/** Token-free loopback: the nonce selects host-owned state on the execution's Durable Object.
151 * Arbitrary RPC objects cannot be serialized into loopback props. The nonce grants only read/list/stat. */
152export class QuickFiles extends WorkerEntrypoint<{
153  RECEIPT_RUNNER: DurableObjectNamespace<import("./index").ReceiptRunner>;
154}, { runnerId: string; snapshotId: string }> {
155  read(path: string): Promise<string | null> {
156    const { runnerId, snapshotId } = this.ctx.props;
157    return this.env.RECEIPT_RUNNER.get(this.env.RECEIPT_RUNNER.idFromString(runnerId)).readSnapshot(snapshotId, path);
158  }
159  stat(path: string): Promise<QuickFileStat | null> {
160    const { runnerId, snapshotId } = this.ctx.props;
161    return this.env.RECEIPT_RUNNER.get(this.env.RECEIPT_RUNNER.idFromString(runnerId)).statSnapshot(snapshotId, path);
162  }
163  list(prefix: string): Promise<string[]> {
164    const { runnerId, snapshotId } = this.ctx.props;
165    return this.env.RECEIPT_RUNNER.get(this.env.RECEIPT_RUNNER.idFromString(runnerId)).listSnapshot(snapshotId, prefix);
166  }
167}
168