From d9aeb88f79c503750946ec686940bf0825d71d1b Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 8 Aug 2026 22:38:53 +0100 Subject: [PATCH 1/3] fix(flue-review): hydrate from the GitHub tarball instead of JS git isomorphic-git pack indexing in the workflow DO stopped completing once the repo's shallow pack passed ~16MB, silently stalling every review in the hydrating stage until the watchdog killed it. The tarball of the PR head SHA needs no git objects at all: stream it through the runtime's native gzip DecompressionStream and untar into the workspace (ustar + GNU longname + pax paths, symlinks included, archive root stripped). The diff was already API-fetched; nothing downstream used the git repo. Also keeps the hydration stage/R2 instrumentation added while diagnosing: every stage and R2 operation logs start/end so the next stall identifies itself. --- .../.flue/sandboxes/cloudflare-shell.ts | 49 ++++- infra/flue-review/.flue/workflows/review.ts | 183 +++++++++++++++--- 2 files changed, 205 insertions(+), 27 deletions(-) diff --git a/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts b/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts index 6a4e0b692b..65cf9c7ac2 100644 --- a/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts +++ b/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts @@ -220,6 +220,53 @@ function buildCodeToolDescription(): string { ].join("\n"); } +// Temporary R2 diagnostics (2026-08-08 review-stall incident). Wrapped once and +// cached per underlying bucket: Workspace fingerprints construction options per +// storage, so every construction site must receive the identical instance. +const instrumentedBuckets = new WeakMap(); +let r2OpSeq = 0; +function instrumentR2(bucket: R2Bucket): R2Bucket { + const cached = instrumentedBuckets.get(bucket); + if (cached) return cached; + const wrap = (method: string) => + async function (...args: unknown[]) { + const id = ++r2OpSeq; + const key = typeof args[0] === "string" ? args[0] : "?"; + const t0 = Date.now(); + console.log(JSON.stringify({ message: "r2 op start", id, method, key })); + try { + // oxlint-disable-next-line typescript/no-unsafe-type-assertion + const result = await (bucket as unknown as Record)[method](...args); + console.log( + JSON.stringify({ message: "r2 op end", id, method, key, ms: Date.now() - t0 }), + ); + return result; + } catch (error) { + console.error( + JSON.stringify({ + message: "r2 op failed", + id, + method, + key, + ms: Date.now() - t0, + error: error instanceof Error ? error.message : String(error), + }), + ); + throw error; + } + }; + const proxied = new Proxy(bucket, { + get(target, prop, receiver) { + if (typeof prop === "string" && ["get", "put", "delete", "head", "list"].includes(prop)) { + return wrap(prop); + } + return Reflect.get(target, prop, receiver); + }, + }); + instrumentedBuckets.set(bucket, proxied); + return proxied; +} + export function getDefaultWorkspace(r2?: R2Bucket, name?: string): Workspace { const { storage } = getCloudflareContext(); return new Workspace({ @@ -228,6 +275,6 @@ export function getDefaultWorkspace(r2?: R2Bucket, name?: string): Workspace { // `name` keys R2 objects and namespaces this workspace; required when an // R2 bucket is provided (large files spill under r2:///...). ...(name ? { name } : {}), - ...(r2 ? { r2 } : {}), + ...(r2 ? { r2: instrumentR2(r2) } : {}), }); } diff --git a/infra/flue-review/.flue/workflows/review.ts b/infra/flue-review/.flue/workflows/review.ts index 1722ea0f54..c1506a5338 100644 --- a/infra/flue-review/.flue/workflows/review.ts +++ b/infra/flue-review/.flue/workflows/review.ts @@ -16,8 +16,6 @@ // agent initializer and the clone performed in the Action target the exact same // DO SQLite + R2 namespace. -import { WorkspaceFileSystem } from "@cloudflare/shell"; -import { createGit } from "@cloudflare/shell/git"; import { defineAgent, defineWorkflow, @@ -158,35 +156,168 @@ function buildPrContext(payload: ReviewPayload, priorReview?: string): string { return lines.join("\n"); } -// Hydrate the PR into the durable Workspace via JS git (shallow clone of base, -// then fetch + checkout the PR head -- refs/pull/N/head covers fork PRs). Large -// objects (the git packfile) spill to R2 under the workspace name. Idempotent: -// a HYDRATED marker skips re-cloning on workflow re-entry. +// Temporary hydration diagnostics (2026-08-08 review-stall incident): brackets +// every stage so a hang shows as a start line with no matching end line. R2 +// operations are instrumented in the shared getDefaultWorkspace. Remove once +// the stall is diagnosed. +function hydrateStep(payload: ReviewPayload, step: string, startedAt: number): void { + console.log( + JSON.stringify({ + message: "hydrate step", + step, + ms: Date.now() - startedAt, + attemptId: payload.attemptId, + prNumber: payload.prNumber, + }), + ); +} + +// Untar a gzip'd GitHub tarball stream into the workspace under `destDir`, +// stripping the archive's single top-level directory. Handles ustar regular +// files, directories, symlinks, GNU longname ('L') and pax ('x') path +// overrides. Entries are processed incrementally; only one entry's content is +// buffered at a time. +async function untarInto( + workspace: ReturnType, + stream: ReadableStream, + destDir: string, +): Promise<{ files: number; bytes: number }> { + const decoder = new TextDecoder(); + let buffer = new Uint8Array(0); + let files = 0; + let bytes = 0; + let pendingLongName: string | undefined; + let pendingPaxPath: string | undefined; + const dirsMade = new Set(); + + const append = (chunk: Uint8Array) => { + const next = new Uint8Array(buffer.length + chunk.length); + next.set(buffer, 0); + next.set(chunk, buffer.length); + buffer = next; + }; + const readCString = (view: Uint8Array): string => { + const end = view.indexOf(0); + return decoder.decode(end === -1 ? view : view.subarray(0, end)); + }; + const stripRoot = (name: string): string | undefined => { + const slash = name.indexOf("/"); + if (slash === -1) return undefined; + const rest = name.slice(slash + 1); + return rest.length > 0 ? rest : undefined; + }; + const ensureDir = async (path: string) => { + if (dirsMade.has(path)) return; + await workspace.mkdir(path, { recursive: true }); + dirsMade.add(path); + }; + const parentOf = (path: string): string => path.slice(0, path.lastIndexOf("/")); + + const reader = stream.getReader(); + let done = false; + const need = async (n: number): Promise => { + while (buffer.length < n && !done) { + const r = await reader.read(); + if (r.done) done = true; + else append(r.value); + } + return buffer.length >= n; + }; + + while (await need(512)) { + const header = buffer.subarray(0, 512); + buffer = buffer.subarray(512); + // Two consecutive zero blocks terminate the archive. + if (header.every((b) => b === 0)) break; + + const rawName = readCString(header.subarray(0, 100)); + const prefix = readCString(header.subarray(345, 500)); + const size = parseInt(readCString(header.subarray(124, 136)).trim() || "0", 8); + // Mode bytes (100-108) are ignored: the Workspace has no chmod and the + // reviewer never executes files. + const type = String.fromCharCode(header[156] ?? 48); + const linkTarget = readCString(header.subarray(157, 257)); + + const padded = Math.ceil(size / 512) * 512; + if (!(await need(padded)) && size > 0) { + throw new Error(`tar truncated: needed ${padded} bytes for entry ${rawName}`); + } + const content = buffer.subarray(0, size); + buffer = buffer.subarray(Math.min(padded, buffer.length)); + + if (type === "L") { + pendingLongName = readCString(content); + continue; + } + if (type === "x" || type === "g") { + // pax records: " key=value\n" + const text = decoder.decode(content); + for (const line of text.split("\n")) { + const eq = line.indexOf("="); + if (eq > 0 && line.slice(line.indexOf(" ") + 1, eq) === "path") { + pendingPaxPath = line.slice(eq + 1); + } + } + continue; + } + + const fullName = + pendingPaxPath ?? pendingLongName ?? (prefix ? `${prefix}/${rawName}` : rawName); + pendingLongName = undefined; + pendingPaxPath = undefined; + + const relative = stripRoot(fullName); + if (!relative) continue; + const dest = `${destDir}/${relative}`; + + if (type === "5") { + await ensureDir(dest); + } else if (type === "2") { + await ensureDir(parentOf(dest)); + await workspace.symlink(linkTarget, dest); + } else if (type === "0" || type === "\0" || type === "7") { + await ensureDir(parentOf(dest)); + // Copy out of the rolling buffer: content is a subarray view. + await workspace.writeFileBytes(dest, new Uint8Array(content)); + files += 1; + bytes += size; + } + // Hardlinks and other exotic types don't occur in GitHub tarballs; skip. + } + return { files, bytes }; +} + +// Hydrate the PR into the durable Workspace from the GitHub tarball of the PR +// head SHA (reachable in the base repo for fork PRs too). No git objects, no +// pack indexing -- pack inflation in the DO stalled past ~16MB repo size, +// which is what took reviews down on 2026-08-08. gzip decompression is the +// runtime-native DecompressionStream. Idempotent: a HYDRATED marker skips +// re-fetching on workflow re-entry. async function hydrate(env: Env, payload: ReviewPayload): Promise { + const t0 = Date.now(); const workspace = getDefaultWorkspace(env.REVIEW_WORKSPACE, workspaceName()); - if (await workspace.exists(HYDRATED)) return; - - const fs = new WorkspaceFileSystem(workspace); - const cloneUrl = `https://github.com/${payload.owner}/${payload.repo}.git`; - const git = createGit(fs); - await git.clone({ - url: cloneUrl, - dir: REPO_DIR, - branch: payload.baseRef, - singleBranch: true, - depth: 1, - }); - const fetched = await git.fetch({ - ref: `pull/${payload.prNumber}/head`, - depth: 1, - dir: REPO_DIR, + hydrateStep(payload, "workspace created", t0); + if (await workspace.exists(HYDRATED)) { + hydrateStep(payload, "already hydrated", t0); + return; + } + if (!payload.headSha) throw new Error("hydrate requires the PR head SHA"); + + const url = `https://api.github.com/repos/${payload.owner}/${payload.repo}/tarball/${payload.headSha}`; + const response = await fetch(url, { + headers: { "User-Agent": "emdash-flue-review", Accept: "application/vnd.github+json" }, }); - if (!fetched.fetchHead) throw new Error("PR head fetch did not return a commit"); - if (payload.headSha && fetched.fetchHead.toLowerCase() !== payload.headSha.toLowerCase()) { - throw new Error("PR head changed after the review was requested"); + if (!response.ok || !response.body) { + throw new Error(`tarball fetch failed: ${response.status} ${await response.text()}`); } - await git.checkout({ ref: fetched.fetchHead, dir: REPO_DIR, force: true }); + hydrateStep(payload, "tarball response", t0); + + const tarStream = response.body.pipeThrough(new DecompressionStream("gzip")); + const { files, bytes } = await untarInto(workspace, tarStream, REPO_DIR); + hydrateStep(payload, `untarred ${files} files ${bytes} bytes`, t0); + await workspace.writeFile(HYDRATED, new Date().toISOString()); + hydrateStep(payload, "hydrated", t0); } function logReviewEvent( From cd94087c8abe81e7a025848ad18e22207a6a59b5 Mon Sep 17 00:00:00 2001 From: "emdashbot[bot]" Date: Sat, 8 Aug 2026 21:43:06 +0000 Subject: [PATCH 2/3] style: format --- infra/flue-review/.flue/sandboxes/cloudflare-shell.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts b/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts index 65cf9c7ac2..a40c814250 100644 --- a/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts +++ b/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts @@ -237,9 +237,7 @@ function instrumentR2(bucket: R2Bucket): R2Bucket { try { // oxlint-disable-next-line typescript/no-unsafe-type-assertion const result = await (bucket as unknown as Record)[method](...args); - console.log( - JSON.stringify({ message: "r2 op end", id, method, key, ms: Date.now() - t0 }), - ); + console.log(JSON.stringify({ message: "r2 op end", id, method, key, ms: Date.now() - t0 })); return result; } catch (error) { console.error( From 06ffa4fc02b907ffd46edfeff9ebab8e0c9e4826 Mon Sep 17 00:00:00 2001 From: Matt Kane Date: Sat, 8 Aug 2026 22:51:33 +0100 Subject: [PATCH 3/3] fix(flue-review): harden untar path handling and extract it for testing Review findings on the tarball hydration: the parser is the boundary between an externally-supplied archive and the workspace, so entry paths and symlink targets now reject absolute paths and any traversal that would resolve outside the destination. The parser moves to lib/untar.ts behind a narrow target interface with a vitest suite covering files, directories, symlinks, GNU longname, pax paths, truncation, and the escape rejections. Stale JS-git and incident narrative trimmed from comments. --- infra/flue-review/.flue/lib/untar.ts | 162 ++++++++++++++ .../.flue/sandboxes/cloudflare-shell.ts | 7 +- infra/flue-review/.flue/workflows/review.ts | 136 +----------- infra/flue-review/test/untar.test.ts | 202 ++++++++++++++++++ 4 files changed, 379 insertions(+), 128 deletions(-) create mode 100644 infra/flue-review/.flue/lib/untar.ts create mode 100644 infra/flue-review/test/untar.test.ts diff --git a/infra/flue-review/.flue/lib/untar.ts b/infra/flue-review/.flue/lib/untar.ts new file mode 100644 index 0000000000..1d3ff67d16 --- /dev/null +++ b/infra/flue-review/.flue/lib/untar.ts @@ -0,0 +1,162 @@ +// Streaming untar for GitHub source tarballs (post-gunzip). The parser is the +// trust boundary between an externally-supplied archive and the review +// workspace: entry paths and symlink targets are validated so no write can +// land outside the destination directory. + +/** The subset of the Workspace surface the untar needs; narrow for testing. */ +export interface UntarTarget { + mkdir(path: string, options?: { recursive?: boolean }): Promise; + symlink(target: string, linkPath: string): Promise; + writeFileBytes(path: string, content: Uint8Array): Promise; +} + +function stripRoot(name: string): string | undefined { + const slash = name.indexOf("/"); + if (slash === -1) return undefined; + const rest = name.slice(slash + 1); + return rest.length > 0 ? rest : undefined; +} + +function parentOf(path: string): string { + return path.slice(0, path.lastIndexOf("/")); +} + +function normalizeSegments(path: string): string[] | undefined { + const out: string[] = []; + for (const part of path.split("/")) { + if (part === "" || part === ".") continue; + if (part === "..") { + if (out.length === 0) return undefined; + out.pop(); + continue; + } + out.push(part); + } + return out; +} + +/** + * Untar a ustar stream into `destDir`, stripping the archive's single + * top-level directory. Handles regular files, directories, symlinks, GNU + * longname ('L') and pax ('x') path overrides. One entry's content is + * buffered at a time. Rejects any entry path or symlink target that would + * resolve outside `destDir`. + */ +export async function untarInto( + target: UntarTarget, + stream: ReadableStream, + destDir: string, +): Promise<{ files: number; bytes: number }> { + const decoder = new TextDecoder(); + let buffer = new Uint8Array(0); + let files = 0; + let bytes = 0; + let pendingLongName: string | undefined; + let pendingPaxPath: string | undefined; + const dirsMade = new Set(); + + const append = (chunk: Uint8Array) => { + const next = new Uint8Array(buffer.length + chunk.length); + next.set(buffer, 0); + next.set(chunk, buffer.length); + buffer = next; + }; + const readCString = (view: Uint8Array): string => { + const end = view.indexOf(0); + return decoder.decode(end === -1 ? view : view.subarray(0, end)); + }; + const ensureDir = async (path: string) => { + if (dirsMade.has(path)) return; + await target.mkdir(path, { recursive: true }); + dirsMade.add(path); + }; + + const reader = stream.getReader(); + let done = false; + const need = async (n: number): Promise => { + while (buffer.length < n && !done) { + const r = await reader.read(); + if (r.done) done = true; + else append(r.value); + } + return buffer.length >= n; + }; + + while (await need(512)) { + const header = buffer.subarray(0, 512); + buffer = buffer.subarray(512); + // Two consecutive zero blocks terminate the archive. + if (header.every((b) => b === 0)) break; + + const rawName = readCString(header.subarray(0, 100)); + const prefix = readCString(header.subarray(345, 500)); + const size = parseInt(readCString(header.subarray(124, 136)).trim() || "0", 8); + // Mode bytes (100-108) are ignored: the workspace has no chmod and the + // reviewer never executes files. + const type = String.fromCharCode(header[156] ?? 48); + const linkTarget = readCString(header.subarray(157, 257)); + + const padded = Math.ceil(size / 512) * 512; + if (!(await need(padded)) && size > 0) { + throw new Error(`tar truncated: needed ${padded} bytes for entry ${rawName}`); + } + const content = buffer.subarray(0, size); + buffer = buffer.subarray(Math.min(padded, buffer.length)); + + if (type === "L") { + pendingLongName = readCString(content); + continue; + } + if (type === "x" || type === "g") { + // pax records: " key=value\n" + const text = decoder.decode(content); + for (const line of text.split("\n")) { + const eq = line.indexOf("="); + if (eq > 0 && line.slice(line.indexOf(" ") + 1, eq) === "path") { + pendingPaxPath = line.slice(eq + 1); + } + } + continue; + } + + const fullName = + pendingPaxPath ?? pendingLongName ?? (prefix ? `${prefix}/${rawName}` : rawName); + pendingLongName = undefined; + pendingPaxPath = undefined; + + if (fullName.startsWith("/")) { + throw new Error(`tar entry path is absolute: ${fullName}`); + } + const relative = stripRoot(fullName); + if (!relative) continue; + const segments = normalizeSegments(relative); + if (segments === undefined || segments.length === 0) { + throw new Error(`tar entry path escapes the destination: ${fullName}`); + } + const dest = `${destDir}/${segments.join("/")}`; + + if (type === "5") { + await ensureDir(dest); + } else if (type === "2") { + if (linkTarget.startsWith("/")) { + throw new Error(`tar symlink target is absolute: ${fullName} -> ${linkTarget}`); + } + // Resolve the target against the link's directory; it must stay + // inside destDir even before the workspace resolves anything. + const resolved = normalizeSegments(`${segments.slice(0, -1).join("/")}/${linkTarget}`); + if (resolved === undefined) { + throw new Error(`tar symlink target escapes the destination: ${fullName} -> ${linkTarget}`); + } + await ensureDir(parentOf(dest)); + await target.symlink(linkTarget, dest); + } else if (type === "0" || type === "\0" || type === "7") { + await ensureDir(parentOf(dest)); + // Copy out of the rolling buffer: content is a subarray view. + await target.writeFileBytes(dest, new Uint8Array(content)); + files += 1; + bytes += size; + } + // Hardlinks and other exotic types don't occur in GitHub tarballs; skip. + } + return { files, bytes }; +} diff --git a/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts b/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts index a40c814250..475d6761ac 100644 --- a/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts +++ b/infra/flue-review/.flue/sandboxes/cloudflare-shell.ts @@ -220,9 +220,10 @@ function buildCodeToolDescription(): string { ].join("\n"); } -// Temporary R2 diagnostics (2026-08-08 review-stall incident). Wrapped once and -// cached per underlying bucket: Workspace fingerprints construction options per -// storage, so every construction site must receive the identical instance. +// R2 operations log start/end so a stalled call is visible as an unmatched +// start line. Wrapped once and cached per underlying bucket: Workspace +// fingerprints construction options per storage, so every construction site +// must receive the identical instance. const instrumentedBuckets = new WeakMap(); let r2OpSeq = 0; function instrumentR2(bucket: R2Bucket): R2Bucket { diff --git a/infra/flue-review/.flue/workflows/review.ts b/infra/flue-review/.flue/workflows/review.ts index c1506a5338..651f4beec1 100644 --- a/infra/flue-review/.flue/workflows/review.ts +++ b/infra/flue-review/.flue/workflows/review.ts @@ -2,7 +2,8 @@ // // Reviews one pull request and returns structured findings plus a verdict. No // firecracker container: the PR is hydrated into a durable cf-shell Workspace -// (DO SQLite + R2 for large files) via JS git, and the agent inspects it with a +// (DO SQLite + R2 for large files) from the GitHub tarball of the PR head, and +// the agent inspects it with a // Worker-Loader-backed `code` tool. It does NOT post to GitHub: the workflow's // trusted Action code posts with a write-scoped installation token, so no // secret is ever reachable by the model. @@ -43,6 +44,7 @@ import { type ReviewStage, type ReviewTerminal, } from "../lib/review-watchdog.js"; +import { untarInto } from "../lib/untar.js"; import { getDefaultWorkspace, getShellSandbox } from "../sandboxes/cloudflare-shell.js"; import review from "../skills/review/SKILL.md" with { type: "skill" }; @@ -156,10 +158,9 @@ function buildPrContext(payload: ReviewPayload, priorReview?: string): string { return lines.join("\n"); } -// Temporary hydration diagnostics (2026-08-08 review-stall incident): brackets -// every stage so a hang shows as a start line with no matching end line. R2 -// operations are instrumented in the shared getDefaultWorkspace. Remove once -// the stall is diagnosed. +// Hydration stage logs bracket each phase so a hang shows as a start line +// with no matching end line. R2 operations are instrumented in the shared +// getDefaultWorkspace. function hydrateStep(payload: ReviewPayload, step: string, startedAt: number): void { console.log( JSON.stringify({ @@ -172,127 +173,12 @@ function hydrateStep(payload: ReviewPayload, step: string, startedAt: number): v ); } -// Untar a gzip'd GitHub tarball stream into the workspace under `destDir`, -// stripping the archive's single top-level directory. Handles ustar regular -// files, directories, symlinks, GNU longname ('L') and pax ('x') path -// overrides. Entries are processed incrementally; only one entry's content is -// buffered at a time. -async function untarInto( - workspace: ReturnType, - stream: ReadableStream, - destDir: string, -): Promise<{ files: number; bytes: number }> { - const decoder = new TextDecoder(); - let buffer = new Uint8Array(0); - let files = 0; - let bytes = 0; - let pendingLongName: string | undefined; - let pendingPaxPath: string | undefined; - const dirsMade = new Set(); - - const append = (chunk: Uint8Array) => { - const next = new Uint8Array(buffer.length + chunk.length); - next.set(buffer, 0); - next.set(chunk, buffer.length); - buffer = next; - }; - const readCString = (view: Uint8Array): string => { - const end = view.indexOf(0); - return decoder.decode(end === -1 ? view : view.subarray(0, end)); - }; - const stripRoot = (name: string): string | undefined => { - const slash = name.indexOf("/"); - if (slash === -1) return undefined; - const rest = name.slice(slash + 1); - return rest.length > 0 ? rest : undefined; - }; - const ensureDir = async (path: string) => { - if (dirsMade.has(path)) return; - await workspace.mkdir(path, { recursive: true }); - dirsMade.add(path); - }; - const parentOf = (path: string): string => path.slice(0, path.lastIndexOf("/")); - - const reader = stream.getReader(); - let done = false; - const need = async (n: number): Promise => { - while (buffer.length < n && !done) { - const r = await reader.read(); - if (r.done) done = true; - else append(r.value); - } - return buffer.length >= n; - }; - - while (await need(512)) { - const header = buffer.subarray(0, 512); - buffer = buffer.subarray(512); - // Two consecutive zero blocks terminate the archive. - if (header.every((b) => b === 0)) break; - - const rawName = readCString(header.subarray(0, 100)); - const prefix = readCString(header.subarray(345, 500)); - const size = parseInt(readCString(header.subarray(124, 136)).trim() || "0", 8); - // Mode bytes (100-108) are ignored: the Workspace has no chmod and the - // reviewer never executes files. - const type = String.fromCharCode(header[156] ?? 48); - const linkTarget = readCString(header.subarray(157, 257)); - - const padded = Math.ceil(size / 512) * 512; - if (!(await need(padded)) && size > 0) { - throw new Error(`tar truncated: needed ${padded} bytes for entry ${rawName}`); - } - const content = buffer.subarray(0, size); - buffer = buffer.subarray(Math.min(padded, buffer.length)); - - if (type === "L") { - pendingLongName = readCString(content); - continue; - } - if (type === "x" || type === "g") { - // pax records: " key=value\n" - const text = decoder.decode(content); - for (const line of text.split("\n")) { - const eq = line.indexOf("="); - if (eq > 0 && line.slice(line.indexOf(" ") + 1, eq) === "path") { - pendingPaxPath = line.slice(eq + 1); - } - } - continue; - } - - const fullName = - pendingPaxPath ?? pendingLongName ?? (prefix ? `${prefix}/${rawName}` : rawName); - pendingLongName = undefined; - pendingPaxPath = undefined; - - const relative = stripRoot(fullName); - if (!relative) continue; - const dest = `${destDir}/${relative}`; - - if (type === "5") { - await ensureDir(dest); - } else if (type === "2") { - await ensureDir(parentOf(dest)); - await workspace.symlink(linkTarget, dest); - } else if (type === "0" || type === "\0" || type === "7") { - await ensureDir(parentOf(dest)); - // Copy out of the rolling buffer: content is a subarray view. - await workspace.writeFileBytes(dest, new Uint8Array(content)); - files += 1; - bytes += size; - } - // Hardlinks and other exotic types don't occur in GitHub tarballs; skip. - } - return { files, bytes }; -} - // Hydrate the PR into the durable Workspace from the GitHub tarball of the PR -// head SHA (reachable in the base repo for fork PRs too). No git objects, no -// pack indexing -- pack inflation in the DO stalled past ~16MB repo size, -// which is what took reviews down on 2026-08-08. gzip decompression is the -// runtime-native DecompressionStream. Idempotent: a HYDRATED marker skips -// re-fetching on workflow re-entry. +// head SHA (reachable in the base repo for fork PRs too). No git objects and +// no pack indexing: pure-JS pack inflation in the DO stops completing once +// the repo's shallow pack grows past roughly 16MB, so hydration must not +// depend on git. gzip decompression is the runtime-native DecompressionStream. +// Idempotent: a HYDRATED marker skips re-fetching on workflow re-entry. async function hydrate(env: Env, payload: ReviewPayload): Promise { const t0 = Date.now(); const workspace = getDefaultWorkspace(env.REVIEW_WORKSPACE, workspaceName()); diff --git a/infra/flue-review/test/untar.test.ts b/infra/flue-review/test/untar.test.ts new file mode 100644 index 0000000000..47e3459d44 --- /dev/null +++ b/infra/flue-review/test/untar.test.ts @@ -0,0 +1,202 @@ +import { describe, expect, it } from "vitest"; + +import { untarInto, type UntarTarget } from "../.flue/lib/untar.js"; + +const encoder = new TextEncoder(); + +function header(fields: { + name: string; + size?: number; + type?: string; + linkTarget?: string; + prefix?: string; +}): Uint8Array { + const block = new Uint8Array(512); + const put = (offset: number, value: string, length: number) => { + const bytes = encoder.encode(value); + block.set(bytes.subarray(0, length), offset); + }; + put(0, fields.name, 100); + put(100, "0000644", 8); + put(124, (fields.size ?? 0).toString(8).padStart(11, "0"), 12); + block[156] = (fields.type ?? "0").charCodeAt(0); + if (fields.linkTarget) put(157, fields.linkTarget, 100); + put(257, "ustar", 6); + if (fields.prefix) put(345, fields.prefix, 155); + return block; +} + +function contentBlocks(data: Uint8Array): Uint8Array { + const padded = new Uint8Array(Math.ceil(data.length / 512) * 512); + padded.set(data); + return padded; +} + +function tarball(entries: Uint8Array[]): ReadableStream { + const terminator = new Uint8Array(1024); + const all = [...entries, terminator]; + return new ReadableStream({ + start(controller) { + for (const chunk of all) controller.enqueue(chunk); + controller.close(); + }, + }); +} + +function fileEntry(name: string, text: string): Uint8Array[] { + const data = encoder.encode(text); + return [header({ name, size: data.length }), contentBlocks(data)]; +} + +interface Recorded { + target: UntarTarget; + files: Map; + symlinks: Map; + dirs: Set; +} + +function recorder(): Recorded { + const files = new Map(); + const symlinks = new Map(); + const dirs = new Set(); + return { + files, + symlinks, + dirs, + target: { + mkdir: async (path) => { + dirs.add(path); + }, + symlink: async (target, path) => { + symlinks.set(path, target); + }, + writeFileBytes: async (path, content) => { + files.set(path, content); + }, + }, + }; +} + +describe("untarInto", () => { + it("extracts regular files, directories and symlinks with the root stripped", async () => { + const r = recorder(); + const result = await untarInto( + r.target, + tarball([ + header({ name: "repo-abc/", type: "5" }), + header({ name: "repo-abc/src/", type: "5" }), + ...fileEntry("repo-abc/src/index.ts", "export {}\n"), + ...fileEntry("repo-abc/empty.txt", ""), + header({ name: "repo-abc/link.md", type: "2", linkTarget: "src/index.ts" }), + ]), + "/repo", + ); + + expect(result.files).toBe(2); + expect(new TextDecoder().decode(r.files.get("/repo/src/index.ts"))).toBe("export {}\n"); + expect(r.files.get("/repo/empty.txt")).toEqual(new Uint8Array(0)); + expect(r.symlinks.get("/repo/link.md")).toBe("src/index.ts"); + expect(r.dirs.has("/repo/src")).toBe(true); + }); + + it("applies GNU longname and pax path overrides", async () => { + const longName = `repo-abc/deep/${"d/".repeat(60)}long-file.txt`; + const longNameBytes = encoder.encode(longName); + // pax record: " key=value\n" where len counts the whole record. + const paxBody = "path=repo-abc/pax-named.txt\n"; + const paxLen = String(paxBody.length + 3).length + 1 + paxBody.length; + const paxBytes = encoder.encode(`${paxLen} ${paxBody}`); + const r = recorder(); + await untarInto( + r.target, + tarball([ + header({ name: "repo-abc/@LongLink", type: "L", size: longNameBytes.length }), + contentBlocks(longNameBytes), + header({ name: "repo-abc/truncated", size: 2 }), + contentBlocks(encoder.encode("ok")), + header({ name: "repo-abc/PaxHeader", type: "x", size: paxBytes.length }), + contentBlocks(paxBytes), + ...fileEntry("repo-abc/ignored-name.txt", "pax"), + ]), + "/repo", + ); + + expect([...r.files.keys()].some((k) => k.endsWith("/long-file.txt"))).toBe(true); + expect(r.files.has("/repo/pax-named.txt")).toBe(true); + expect(r.files.has("/repo/ignored-name.txt")).toBe(false); + }); + + it("joins the ustar prefix field with the name", async () => { + const r = recorder(); + await untarInto( + r.target, + tarball([ + header({ name: "nested.txt", prefix: "repo-abc/prefixed", size: 2 }), + contentBlocks(encoder.encode("hi")), + ]), + "/repo", + ); + expect(r.files.has("/repo/prefixed/nested.txt")).toBe(true); + }); + + it("rejects entry paths that traverse out of the destination", async () => { + const r = recorder(); + await expect( + untarInto(r.target, tarball([...fileEntry("repo-abc/../../escape.txt", "x")]), "/repo"), + ).rejects.toThrow(/escapes the destination/); + expect(r.files.size).toBe(0); + }); + + it("rejects absolute entry paths", async () => { + const r = recorder(); + await expect( + untarInto(r.target, tarball([...fileEntry("/etc/passwd", "x")]), "/repo"), + ).rejects.toThrow(/absolute/); + }); + + it("rejects symlink targets that escape the destination", async () => { + const r = recorder(); + await expect( + untarInto( + r.target, + tarball([header({ name: "repo-abc/evil", type: "2", linkTarget: "../../etc/passwd" })]), + "/repo", + ), + ).rejects.toThrow(/escapes the destination/); + await expect( + untarInto( + r.target, + tarball([header({ name: "repo-abc/evil", type: "2", linkTarget: "/etc/passwd" })]), + "/repo", + ), + ).rejects.toThrow(/absolute/); + expect(r.symlinks.size).toBe(0); + }); + + it("allows in-tree relative symlink targets that use ..", async () => { + const r = recorder(); + await untarInto( + r.target, + tarball([ + header({ name: "repo-abc/.claude/", type: "5" }), + header({ name: "repo-abc/.claude/skills", type: "2", linkTarget: "../skills" }), + ]), + "/repo", + ); + expect(r.symlinks.get("/repo/.claude/skills")).toBe("../skills"); + }); + + it("throws on a truncated archive instead of writing partial content", async () => { + const r = recorder(); + // No terminator and only 100 of the 600 declared content bytes. + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(header({ name: "repo-abc/cut.txt", size: 600 })); + controller.enqueue(new Uint8Array(100)); + controller.close(); + }, + }); + await expect(untarInto(r.target, stream, "/repo")).rejects.toThrow(/truncated/); + expect(r.files.size).toBe(0); + }); +});