From 685f46f866172d4f1c9d72337e388609f3dce76e Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 31 Oct 2026 21:41:36 +0000 Subject: [PATCH 1/4] refactor: return Results from tar pipe helpers in cache-store --- src/datasets/cache-store.test.ts | 212 ++++++++++++++++++++++++++++- src/datasets/cache-store.ts | 223 ++++++++++++++++++++++--------- 2 files changed, 368 insertions(+), 67 deletions(-) diff --git a/src/datasets/cache-store.test.ts b/src/datasets/cache-store.test.ts index a29492c0..6dc2a58b 100644 --- a/src/datasets/cache-store.test.ts +++ b/src/datasets/cache-store.test.ts @@ -9,11 +9,13 @@ import { } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { PassThrough, Readable } from "node:stream"; +import { PassThrough, Readable, Writable } from "node:stream"; +import { assertFailure, assertSuccess } from "../internal/testing"; import type { GcsObjectClient } from "./cache-store"; import { datasetCacheWriteFailures, + extractTarGzFrom, makeDiskCacheStore, makeGcsCacheStore, pipeTarGzTo, @@ -130,6 +132,22 @@ describe("GcsCacheStore", () => { return info; } + function captureWarn() { + process.env.OR_ENV = "development"; + process.env.LOG_LEVEL = "1"; + const warn = spyOn(console, "warn").mockImplementation(() => {}); + infoSpies.push(warn); + return warn; + } + + function withoutTarOnPath(run: () => Promise): Promise { + const savedPath = process.env.PATH; + process.env.PATH = makeTmpDir(); + return run().finally(() => { + process.env.PATH = savedPath; + }); + } + it("logs a hit when a fresh JSON object is read", async () => { const client = makeMemoryGcsClient(); const store = makeGcsCacheStore({ bucket: "b", client }); @@ -374,15 +392,195 @@ describe("GcsCacheStore", () => { const client = makeMemoryGcsClient(); const dir = makeTmpDir(); const snapshotKey = "repos/harbor-missing.tar.gz"; - await expect( - pipeTarGzTo( - join(dir, "missing"), - client.openObjectWriteStream(snapshotKey, "application/gzip") - ) - ).rejects.toThrow("tar create exited"); + const result = await pipeTarGzTo( + join(dir, "missing"), + client.openObjectWriteStream(snapshotKey, "application/gzip") + ); + assertFailure(result); + expect(result.failure.kind).toBe("tar_exited"); + expect(result.failure.operation).toBe("create"); + if (result.failure.kind === "tar_exited") { + expect(result.failure.code).not.toBe(0); + expect(result.failure.stderr).not.toBe(""); + } expect(client.store.has(snapshotKey)).toBe(false); }); + it("pipeTarGzTo succeeds and ends the destination for a real directory", async () => { + const client = makeMemoryGcsClient(); + const dir = makeTmpDir(); + writeFileSync(join(dir, "task.toml"), "name = 'a'\n"); + const snapshotKey = "repos/harbor-ok.tar.gz"; + const result = await pipeTarGzTo( + dir, + client.openObjectWriteStream(snapshotKey, "application/gzip") + ); + assertSuccess(result); + expect(client.store.has(snapshotKey)).toBe(true); + }); + + it("pipeTarGzTo reports a stream failure when the destination rejects writes", async () => { + const dir = makeTmpDir(); + writeFileSync(join(dir, "task.toml"), "name = 'a'\n"); + const dest = new Writable({ + write(_chunk, _encoding, callback) { + callback(new Error("disk full")); + }, + }); + dest.on("error", () => {}); + const result = await pipeTarGzTo(dir, dest); + assertFailure(result); + expect(result.failure).toEqual({ + kind: "stream_failed", + operation: "create", + message: "disk full", + }); + }); + + it("pipeTarGzTo reports a spawn failure when tar is not on PATH", async () => { + const dir = makeTmpDir(); + writeFileSync(join(dir, "task.toml"), "name = 'a'\n"); + const dest = new PassThrough(); + dest.on("error", () => {}); + dest.resume(); + const result = await withoutTarOnPath(() => pipeTarGzTo(dir, dest)); + assertFailure(result); + expect(result.failure.kind).toBe("spawn_failed"); + expect(result.failure.operation).toBe("create"); + expect(result.failure).toHaveProperty("message"); + }); + + it("extractTarGzFrom reports tar_exited for a corrupt archive", async () => { + const dir = makeTmpDir(); + const result = await extractTarGzFrom( + Readable.from([Buffer.from("not a gzip archive")]), + join(dir, "out") + ); + assertFailure(result); + expect(result.failure.kind).toBe("tar_exited"); + expect(result.failure.operation).toBe("extract"); + if (result.failure.kind === "tar_exited") { + expect(result.failure.code).not.toBe(0); + expect(result.failure.stderr).not.toBe(""); + } + }); + + it("extractTarGzFrom reports a stream failure when the source errors", async () => { + const dir = makeTmpDir(); + const src = new Readable({ + read() { + this.destroy(new Error("connection reset")); + }, + }); + const result = await extractTarGzFrom(src, join(dir, "out")); + assertFailure(result); + expect(result.failure).toEqual({ + kind: "stream_failed", + operation: "extract", + message: "connection reset", + }); + }); + + it("extractTarGzFrom reports a spawn failure when tar is not on PATH", async () => { + const dir = makeTmpDir(); + const result = await withoutTarOnPath(() => + extractTarGzFrom( + Readable.from([Buffer.from("ignored")]), + join(dir, "out") + ) + ); + assertFailure(result); + expect(result.failure.kind).toBe("spawn_failed"); + expect(result.failure.operation).toBe("extract"); + }); + + it("tryHydrateCheckout returns false and warns when the snapshot is corrupt", async () => { + const client = makeMemoryGcsClient(); + await client.uploadObject( + "repos/harbor-abc123.tar.gz", + Buffer.from("not a gzip archive"), + "application/gzip" + ); + const store = makeGcsCacheStore({ bucket: "b", client }); + const dir = makeTmpDir(); + const warn = captureWarn(); + + await expect( + store.tryHydrateCheckout(dir, "harbor", "abc123") + ).resolves.toBe(false); + + expect(warn.mock.calls).toHaveLength(1); + const [message, context] = warn.mock.calls[0] ?? []; + expect(message).toBe("GCS checkout hydrate failed"); + expect(context).toMatchObject({ + key: "repos/harbor-abc123.tar.gz", + error: expect.stringContaining("tar extract exited"), + }); + }); + + it("tryHydrateCheckout returns false and warns when opening the object throws", async () => { + const client: GcsObjectClient = { + ...makeMemoryGcsClient(), + async openObjectReadStream() { + throw new Error("bucket unreachable"); + }, + }; + const store = makeGcsCacheStore({ bucket: "b", client }); + const dir = makeTmpDir(); + const warn = captureWarn(); + + await expect( + store.tryHydrateCheckout(dir, "harbor", "abc123") + ).resolves.toBe(false); + + expect(warn.mock.calls).toEqual([ + [ + "GCS checkout hydrate failed", + { key: "repos/harbor-abc123.tar.gz", error: "bucket unreachable" }, + ], + ]); + }); + + it("snapshotCheckout warns and writes nothing when tar fails", async () => { + const client = makeMemoryGcsClient(); + const store = makeGcsCacheStore({ bucket: "b", client }); + const dir = makeTmpDir(); + const warn = captureWarn(); + + await store.snapshotCheckout(join(dir, "missing"), "harbor", "abc123"); + + expect(client.store.size).toBe(0); + expect(warn.mock.calls).toHaveLength(1); + const [message, context] = warn.mock.calls[0] ?? []; + expect(message).toBe("GCS checkout snapshot failed"); + expect(context).toMatchObject({ + key: "repos/harbor-abc123.tar.gz", + error: expect.stringContaining("tar create exited"), + }); + }); + + it("snapshotCheckout warns when opening the write stream throws", async () => { + const client: GcsObjectClient = { + ...makeMemoryGcsClient(), + openObjectWriteStream() { + throw new Error("bucket unreachable"); + }, + }; + const store = makeGcsCacheStore({ bucket: "b", client }); + const dir = makeTmpDir(); + writeFileSync(join(dir, "task.toml"), "name = 'a'\n"); + const warn = captureWarn(); + + await store.snapshotCheckout(dir, "harbor", "abc123"); + + expect(warn.mock.calls).toEqual([ + [ + "GCS checkout snapshot failed", + { key: "repos/harbor-abc123.tar.gz", error: "bucket unreachable" }, + ], + ]); + }); + it("does not export full-buffer tar helpers", async () => { const cacheStore = await import("./cache-store"); expect("createTarGz" in cacheStore).toBe(false); diff --git a/src/datasets/cache-store.ts b/src/datasets/cache-store.ts index 6668b031..076ceaaa 100644 --- a/src/datasets/cache-store.ts +++ b/src/datasets/cache-store.ts @@ -1,10 +1,16 @@ +import type { ChildProcess } from "node:child_process"; import { spawn } from "node:child_process"; import { join } from "node:path"; import type { Readable, Writable } from "node:stream"; import { finished, pipeline } from "node:stream/promises"; +import * as Result from "effect/Result"; + +import { unknownErrorToString } from "../internal/errors"; import { getGcsStorage } from "../internal/gcs"; import { iLog, wLog } from "../internal/log"; +import type { AsyncResult } from "../internal/result"; +import { tryPromiseResult } from "../internal/result"; import { datasetCacheDisabled, datasetCacheRoot, @@ -60,67 +66,147 @@ export interface CacheStore { ) => Promise; } -async function waitForTar( - proc: ReturnType, - operation: string -): Promise { +type TarOperation = "create" | "extract"; + +export type TarGzError = + | { + readonly kind: "tar_exited"; + readonly operation: TarOperation; + readonly code: number | null; + readonly stderr: string; + } + | { + readonly kind: "spawn_failed"; + readonly operation: TarOperation; + readonly message: string; + } + | { + readonly kind: "stream_failed"; + readonly operation: TarOperation; + readonly message: string; + }; + +export function describeTarGzError(error: TarGzError): string { + switch (error.kind) { + case "tar_exited": { + return `tar ${error.operation} exited ${error.code}: ${error.stderr}`; + } + case "spawn_failed": + case "stream_failed": { + return error.message; + } + default: { + return error satisfies never; + } + } +} + +function waitForTar( + proc: ChildProcess, + operation: TarOperation +): AsyncResult { let stderr = ""; proc.stderr?.on("data", (d: Buffer) => (stderr += d.toString())); - await new Promise((resolve, reject) => { - proc.once("error", reject); + return new Promise((resolve) => { + proc.once("error", (error) => + resolve( + Result.fail({ + kind: "spawn_failed", + operation, + message: unknownErrorToString(error), + }) + ) + ); proc.once("close", (code) => - code === 0 - ? resolve() - : reject( - new Error(`tar ${operation} exited ${code}: ${stderr.slice(-1000)}`) - ) + resolve( + code === 0 + ? Result.succeed(undefined) + : Result.fail({ + kind: "tar_exited", + operation, + code, + stderr: stderr.slice(-1000), + }) + ) ); }); } -export async function pipeTarGzTo(dir: string, dest: Writable): Promise { +async function streamStep( + operation: TarOperation, + step: () => Promise +): AsyncResult { + const outcome = await tryPromiseResult(step); + return Result.mapError(outcome, (error) => ({ + kind: "stream_failed", + operation, + message: unknownErrorToString(error), + })); +} + +async function abortTar( + proc: ChildProcess, + exit: AsyncResult, + streamFailure: Result.Failure +): AsyncResult { + proc.kill(); + const exited = await exit; + return Result.isFailure(exited) && exited.failure.kind === "spawn_failed" + ? exited + : streamFailure; +} + +async function streamTarCreate( + stdout: Readable, + dest: Writable, + exit: AsyncResult +): AsyncResult { + const piped = await streamStep("create", () => + pipeline(stdout, dest, { end: false }) + ); + if (Result.isFailure(piped)) { + return piped; + } + const exited = await exit; + if (Result.isFailure(exited)) { + return exited; + } + return streamStep("create", async () => { + dest.end(); + await finished(dest); + }); +} + +export async function pipeTarGzTo( + dir: string, + dest: Writable +): AsyncResult { const proc = spawn("tar", ["-czf", "-", "-C", dir, "."], { stdio: ["ignore", "pipe", "pipe"], }); - if (proc.stdout === null) { - throw new Error("tar create did not provide stdout"); - } const exit = waitForTar(proc, "create"); - try { - await pipeline(proc.stdout, dest, { end: false }); - await exit; - dest.end(); - await finished(dest); - } catch (error) { - const destinationError = - error instanceof Error ? error : new Error(String(error)); - dest.destroy(destinationError); - proc.kill(); - await exit.catch(() => undefined); - throw error; + const streamed = await streamTarCreate(proc.stdout, dest, exit); + if (Result.isFailure(streamed)) { + dest.destroy(new Error(describeTarGzError(streamed.failure))); + return abortTar(proc, exit, streamed); } + return streamed; } export async function extractTarGzFrom( src: Readable, dir: string -): Promise { +): AsyncResult { mkdirOwnerOnly(dir); const proc = spawn("tar", ["-xzf", "-", "-C", dir], { stdio: ["pipe", "ignore", "pipe"], }); - if (proc.stdin === null) { - throw new Error("tar extract did not provide stdin"); - } const exit = waitForTar(proc, "extract"); - try { - await pipeline(src, proc.stdin); - } catch (error) { - proc.kill(); - await exit.catch(() => undefined); - throw error; + const piped = await streamStep("extract", () => pipeline(src, proc.stdin)); + if (Result.isFailure(piped)) { + return abortTar(proc, exit, piped); } - await exit; + return exit; } export function makeDiskCacheStore(): CacheStore { @@ -300,45 +386,62 @@ export function makeGcsCacheStore(opts: GcsCacheStoreOptions): CacheStore { }, async tryHydrateCheckout(localDir, scope, commit) { const full = joinKey(prefix, checkoutSnapshotKey(scope, commit)); - try { - const src = await client.openObjectReadStream(full); - if (src === undefined) { - iLog("checkout cache hydrate", { - key: full, - scope, - commit, - result: "miss", - }); - return false; - } - await extractTarGzFrom(src, localDir); + const opened = await tryPromiseResult(() => + client.openObjectReadStream(full) + ); + if (Result.isFailure(opened)) { + wLog("GCS checkout hydrate failed", { + key: full, + error: unknownErrorToString(opened.failure), + }); + return false; + } + if (opened.success === undefined) { iLog("checkout cache hydrate", { key: full, scope, commit, - result: "hit", + result: "miss", }); - return true; - } catch (error) { + return false; + } + const extracted = await extractTarGzFrom(opened.success, localDir); + if (Result.isFailure(extracted)) { wLog("GCS checkout hydrate failed", { key: full, - error: String(error), + error: describeTarGzError(extracted.failure), }); return false; } + iLog("checkout cache hydrate", { + key: full, + scope, + commit, + result: "hit", + }); + return true; }, async snapshotCheckout(localDir, scope, commit) { const full = joinKey(prefix, checkoutSnapshotKey(scope, commit)); - try { - const dest = client.openObjectWriteStream(full, "application/gzip"); - await pipeTarGzTo(localDir, dest); - iLog("checkout cache snapshot written", { key: full, scope, commit }); - } catch (error) { + const opened = await tryPromiseResult(async () => + client.openObjectWriteStream(full, "application/gzip") + ); + if (Result.isFailure(opened)) { wLog("GCS checkout snapshot failed", { key: full, - error: String(error), + error: unknownErrorToString(opened.failure), }); + return; + } + const written = await pipeTarGzTo(localDir, opened.success); + if (Result.isFailure(written)) { + wLog("GCS checkout snapshot failed", { + key: full, + error: describeTarGzError(written.failure), + }); + return; } + iLog("checkout cache snapshot written", { key: full, scope, commit }); }, }; } From 0292bb98a968d7fbc963f126fde438b6475bef38 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 31 Oct 2026 21:42:53 +0000 Subject: [PATCH 2/4] refactor: share console spy helper in cache-store tests --- src/datasets/cache-store.test.ts | 48 ++++++++++++++------------------ src/datasets/cache-store.ts | 2 +- 2 files changed, 22 insertions(+), 28 deletions(-) diff --git a/src/datasets/cache-store.test.ts b/src/datasets/cache-store.test.ts index 6dc2a58b..67c1566c 100644 --- a/src/datasets/cache-store.test.ts +++ b/src/datasets/cache-store.test.ts @@ -83,7 +83,7 @@ describe("GcsCacheStore", () => { const savedLogLevel = process.env.LOG_LEVEL; const savedOrEnv = process.env.OR_ENV; const tmpDirs: string[] = []; - const infoSpies: { mockRestore: () => void }[] = []; + const consoleSpies: { mockRestore: () => void }[] = []; afterEach(() => { for (const name of [ "BENCH_DATASET_CACHE_BACKEND", @@ -107,8 +107,8 @@ describe("GcsCacheStore", () => { } else { process.env.OR_ENV = savedOrEnv; } - for (const info of infoSpies.splice(0)) { - info.mockRestore(); + for (const spy of consoleSpies.splice(0)) { + spy.mockRestore(); } while (tmpDirs.length > 0) { const dir = tmpDirs.pop(); @@ -124,20 +124,12 @@ describe("GcsCacheStore", () => { return dir; } - function captureInfo() { - process.env.OR_ENV = "development"; - process.env.LOG_LEVEL = "1"; - const info = spyOn(console, "info").mockImplementation(() => {}); - infoSpies.push(info); - return info; - } - - function captureWarn() { + function captureConsole(level: "info" | "warn") { process.env.OR_ENV = "development"; process.env.LOG_LEVEL = "1"; - const warn = spyOn(console, "warn").mockImplementation(() => {}); - infoSpies.push(warn); - return warn; + const spy = spyOn(console, level).mockImplementation(() => {}); + consoleSpies.push(spy); + return spy; } function withoutTarOnPath(run: () => Promise): Promise { @@ -152,7 +144,7 @@ describe("GcsCacheStore", () => { const client = makeMemoryGcsClient(); const store = makeGcsCacheStore({ bucket: "b", client }); await store.writeJson("hit.json", { ok: true }); - const info = captureInfo(); + const info = captureConsole("info"); await expect(store.readJson("hit.json")).resolves.toEqual({ ok: true }); @@ -167,7 +159,7 @@ describe("GcsCacheStore", () => { it("logs a miss when a JSON object is absent", async () => { const client = makeMemoryGcsClient(); const store = makeGcsCacheStore({ bucket: "b", client }); - const info = captureInfo(); + const info = captureConsole("info"); await expect(store.readJson("missing.json")).resolves.toBeUndefined(); @@ -184,7 +176,7 @@ describe("GcsCacheStore", () => { const store = makeGcsCacheStore({ bucket: "b", client }); await store.writeJson("stale.json", { ok: true }); client.setUpdated("stale.json", 1_000); - const info = captureInfo(); + const info = captureConsole("info"); await expect( store.readJson("stale.json", { maxAgeMs: 1_000, now: 3_000 }) @@ -205,7 +197,7 @@ describe("GcsCacheStore", () => { updated: Date.now(), }); const store = makeGcsCacheStore({ bucket: "b", client }); - const info = captureInfo(); + const info = captureConsole("info"); await expect(store.readJson("invalid.json")).resolves.toBeUndefined(); @@ -324,7 +316,7 @@ describe("GcsCacheStore", () => { const src = makeTmpDir(); writeFileSync(join(src, "task.toml"), "name = 'a'\n"); await store.snapshotCheckout(src, "harbor", "deadbeefdeadbeef"); - const info = captureInfo(); + const info = captureConsole("info"); const dest = makeTmpDir(); await expect( @@ -348,7 +340,7 @@ describe("GcsCacheStore", () => { const client = makeMemoryGcsClient(); const store = makeGcsCacheStore({ bucket: "b", client }); const dir = makeTmpDir(); - const info = captureInfo(); + const info = captureConsole("info"); await expect( store.tryHydrateCheckout(dir, "harbor", "abc123") @@ -372,7 +364,7 @@ describe("GcsCacheStore", () => { const store = makeGcsCacheStore({ bucket: "b", client }); const src = makeTmpDir(); writeFileSync(join(src, "task.toml"), "name = 'a'\n"); - const info = captureInfo(); + const info = captureConsole("info"); await store.snapshotCheckout(src, "harbor", "deadbeefdeadbeef"); @@ -447,7 +439,9 @@ describe("GcsCacheStore", () => { assertFailure(result); expect(result.failure.kind).toBe("spawn_failed"); expect(result.failure.operation).toBe("create"); - expect(result.failure).toHaveProperty("message"); + if (result.failure.kind === "spawn_failed") { + expect(result.failure.message).toContain("tar"); + } }); it("extractTarGzFrom reports tar_exited for a corrupt archive", async () => { @@ -503,7 +497,7 @@ describe("GcsCacheStore", () => { ); const store = makeGcsCacheStore({ bucket: "b", client }); const dir = makeTmpDir(); - const warn = captureWarn(); + const warn = captureConsole("warn"); await expect( store.tryHydrateCheckout(dir, "harbor", "abc123") @@ -527,7 +521,7 @@ describe("GcsCacheStore", () => { }; const store = makeGcsCacheStore({ bucket: "b", client }); const dir = makeTmpDir(); - const warn = captureWarn(); + const warn = captureConsole("warn"); await expect( store.tryHydrateCheckout(dir, "harbor", "abc123") @@ -545,7 +539,7 @@ describe("GcsCacheStore", () => { const client = makeMemoryGcsClient(); const store = makeGcsCacheStore({ bucket: "b", client }); const dir = makeTmpDir(); - const warn = captureWarn(); + const warn = captureConsole("warn"); await store.snapshotCheckout(join(dir, "missing"), "harbor", "abc123"); @@ -569,7 +563,7 @@ describe("GcsCacheStore", () => { const store = makeGcsCacheStore({ bucket: "b", client }); const dir = makeTmpDir(); writeFileSync(join(dir, "task.toml"), "name = 'a'\n"); - const warn = captureWarn(); + const warn = captureConsole("warn"); await store.snapshotCheckout(dir, "harbor", "abc123"); diff --git a/src/datasets/cache-store.ts b/src/datasets/cache-store.ts index 076ceaaa..d2df9c11 100644 --- a/src/datasets/cache-store.ts +++ b/src/datasets/cache-store.ts @@ -86,7 +86,7 @@ export type TarGzError = readonly message: string; }; -export function describeTarGzError(error: TarGzError): string { +function describeTarGzError(error: TarGzError): string { switch (error.kind) { case "tar_exited": { return `tar ${error.operation} exited ${error.code}: ${error.stderr}`; From 8258776e5a10047ecd890b1d17464ec334f282b1 Mon Sep 17 00:00:00 2001 From: Robert Sun Date: Sat, 31 Oct 2026 21:54:07 +0000 Subject: [PATCH 3/4] fix: return mkdir_failed from extractTarGzFrom instead of throwing Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- src/datasets/cache-store.test.ts | 40 ++++++++++++++++++++++++++++++++ src/datasets/cache-store.ts | 20 ++++++++++++++-- 2 files changed, 58 insertions(+), 2 deletions(-) diff --git a/src/datasets/cache-store.test.ts b/src/datasets/cache-store.test.ts index 67c1566c..15133826 100644 --- a/src/datasets/cache-store.test.ts +++ b/src/datasets/cache-store.test.ts @@ -459,6 +459,46 @@ describe("GcsCacheStore", () => { } }); + it("extractTarGzFrom reports mkdir_failed when the target cannot be created", async () => { + const dir = makeTmpDir(); + writeFileSync(join(dir, "blocker"), "not a directory\n"); + const result = await extractTarGzFrom( + Readable.from([Buffer.from("ignored")]), + join(dir, "blocker", "out") + ); + assertFailure(result); + expect(result.failure.kind).toBe("mkdir_failed"); + if (result.failure.kind === "mkdir_failed") { + expect(result.failure.dir).toBe(join(dir, "blocker", "out")); + expect(result.failure.message).toContain("ENOTDIR"); + } + }); + + it("tryHydrateCheckout returns false and warns when the target directory cannot be created", async () => { + const client = makeMemoryGcsClient(); + await client.uploadObject( + "repos/harbor-abc123.tar.gz", + Buffer.from("ignored"), + "application/gzip" + ); + const store = makeGcsCacheStore({ bucket: "b", client }); + const dir = makeTmpDir(); + writeFileSync(join(dir, "blocker"), "not a directory\n"); + const warn = captureConsole("warn"); + + await expect( + store.tryHydrateCheckout(join(dir, "blocker", "out"), "harbor", "abc123") + ).resolves.toBe(false); + + expect(warn.mock.calls).toHaveLength(1); + const [message, context] = warn.mock.calls[0] ?? []; + expect(message).toBe("GCS checkout hydrate failed"); + expect(context).toMatchObject({ + key: "repos/harbor-abc123.tar.gz", + error: expect.stringContaining("ENOTDIR"), + }); + }); + it("extractTarGzFrom reports a stream failure when the source errors", async () => { const dir = makeTmpDir(); const src = new Readable({ diff --git a/src/datasets/cache-store.ts b/src/datasets/cache-store.ts index d2df9c11..05ba2f43 100644 --- a/src/datasets/cache-store.ts +++ b/src/datasets/cache-store.ts @@ -84,6 +84,11 @@ export type TarGzError = readonly kind: "stream_failed"; readonly operation: TarOperation; readonly message: string; + } + | { + readonly kind: "mkdir_failed"; + readonly dir: string; + readonly message: string; }; function describeTarGzError(error: TarGzError): string { @@ -92,7 +97,8 @@ function describeTarGzError(error: TarGzError): string { return `tar ${error.operation} exited ${error.code}: ${error.stderr}`; } case "spawn_failed": - case "stream_failed": { + case "stream_failed": + case "mkdir_failed": { return error.message; } default: { @@ -197,7 +203,17 @@ export async function extractTarGzFrom( src: Readable, dir: string ): AsyncResult { - mkdirOwnerOnly(dir); + const created = Result.try({ + try: () => mkdirOwnerOnly(dir), + catch: (error): TarGzError => ({ + kind: "mkdir_failed", + dir, + message: unknownErrorToString(error), + }), + }); + if (Result.isFailure(created)) { + return created; + } const proc = spawn("tar", ["-xzf", "-", "-C", dir], { stdio: ["pipe", "ignore", "pipe"], }); From a6e5095999c1f1efabe55733e3e2b7d24e69140e Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 31 Oct 2026 22:04:59 +0000 Subject: [PATCH 4/4] refactor: open the snapshot write stream synchronously before piping --- src/datasets/cache-store.ts | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/datasets/cache-store.ts b/src/datasets/cache-store.ts index 05ba2f43..5e39c947 100644 --- a/src/datasets/cache-store.ts +++ b/src/datasets/cache-store.ts @@ -439,13 +439,14 @@ export function makeGcsCacheStore(opts: GcsCacheStoreOptions): CacheStore { }, async snapshotCheckout(localDir, scope, commit) { const full = joinKey(prefix, checkoutSnapshotKey(scope, commit)); - const opened = await tryPromiseResult(async () => - client.openObjectWriteStream(full, "application/gzip") - ); + const opened = Result.try({ + try: () => client.openObjectWriteStream(full, "application/gzip"), + catch: unknownErrorToString, + }); if (Result.isFailure(opened)) { wLog("GCS checkout snapshot failed", { key: full, - error: unknownErrorToString(opened.failure), + error: opened.failure, }); return; }