diff --git a/.changeset/asset-download-progress-counter.md b/.changeset/asset-download-progress-counter.md new file mode 100644 index 000000000..6af8f3a82 --- /dev/null +++ b/.changeset/asset-download-progress-counter.md @@ -0,0 +1,5 @@ +--- +"@qawolf/cli": patch +--- + +Show a per-file counter while `flows pull` downloads team-storage assets. The progress line now reads "Downloading team-storage assets (2/12)" and advances as each file starts. The total counts only files that download; reused and skipped files are not included. In human mode the spinner label updates in place, and when a download fails the error line keeps the last counter so you can see where it stopped. Agent mode writes one progress line per file to stderr. The json output does not change. diff --git a/.changeset/stall-timeout-signed-url-downloads.md b/.changeset/stall-timeout-signed-url-downloads.md new file mode 100644 index 000000000..6994a4a2f --- /dev/null +++ b/.changeset/stall-timeout-signed-url-downloads.md @@ -0,0 +1,5 @@ +--- +"@qawolf/cli": patch +--- + +Let large signed-URL downloads finish on slow links. Before this change, flow-bundle and team-storage asset downloads had a fixed 30-second deadline for the full download. A large asset, for example a video file, could not finish in time and `flows pull` failed with a timeout. The 30-second window is now a stall timeout. The timer resets each time data arrives, so a slow download that makes progress can run to completion. A download that receives no data for 30 seconds still fails, and the error message now says the download stalled. diff --git a/.changeset/stream-signed-url-downloads.md b/.changeset/stream-signed-url-downloads.md new file mode 100644 index 000000000..e8416c00b --- /dev/null +++ b/.changeset/stream-signed-url-downloads.md @@ -0,0 +1,5 @@ +--- +"@qawolf/cli": patch +--- + +Stream signed-URL downloads to disk instead of buffering them in memory. Before this change, the CLI held the whole file in memory and briefly needed about twice the file size, so a multi-gigabyte asset could exceed the memory limit of a small CI container. The download now writes each chunk to a `.part` file and renames it into place when the download completes, so peak memory stays near one chunk for any file size. A pull of a 1.4 GB asset now peaks at about 290 MB of memory instead of 3.3 GB. A failed download removes the partial file, and slow disk writes do not count toward the 30-second stall timeout. diff --git a/src/core/messages/auth.ts b/src/core/messages/auth.ts index cd9abd475..2bb0ed8b5 100644 --- a/src/core/messages/auth.ts +++ b/src/core/messages/auth.ts @@ -65,7 +65,7 @@ export const authMessages = { networkUnreachable: "Could not reach the flow bundle storage. Check your network connection and try again.", timedOut: (timeoutMs: number) => - `Downloading the flow bundle timed out after ${formatSeconds(timeoutMs)}. Please try again.`, + `Downloading the flow bundle stalled — no data arrived for ${formatSeconds(timeoutMs)}. Please try again.`, malformed: "The flow bundle download was malformed. Please run `qawolf flows pull` again.", }, diff --git a/src/core/messages/flows.ts b/src/core/messages/flows.ts index ae2a705de..505fbfe21 100644 --- a/src/core/messages/flows.ts +++ b/src/core/messages/flows.ts @@ -33,6 +33,8 @@ export const flowsMessages = { aborted: "Aborted; no changes.", extractingBundle: "Extracting bundle", downloadingTeamStorageAssets: "Downloading team-storage assets", + downloadingTeamStorageAssetsProgress: (current: number, total: number) => + `Downloading team-storage assets (${String(current)}/${String(total)})`, teamStorageRequiresTeamKey: "Team storage requires a team API key; organization keys are not supported here.", summary: (result: PullSummaryInput, assetsAbs: string) => { diff --git a/src/domains/flows/pull/handler.test.ts b/src/domains/flows/pull/handler.test.ts index f14b78fca..8a0f3bef0 100644 --- a/src/domains/flows/pull/handler.test.ts +++ b/src/domains/flows/pull/handler.test.ts @@ -33,14 +33,16 @@ afterEach(async () => { // Real withProgress: runs every task in order, returns their results. // makeFakeUI's stub resolves to []; we need the inner tasks to execute so // the handler reaches its JSON-output branch with real result data. -function makeJsonUi(): UI { +function makeJsonUi(onUpdate?: (message: string) => void): UI { const ui = makeFakeUI(); const withProgress = async ( - steps: readonly { task: () => Promise }[], + steps: readonly { + task: (update: (message: string) => void) => Promise; + }[], ): Promise => { const results: unknown[] = []; for (const step of steps) { - results.push(await step.task()); + results.push(await step.task((message) => onUpdate?.(message))); } return results; }; @@ -57,6 +59,7 @@ function makeCtx( ui: UI, bundlePath: string, envVars: Record = {}, + syncTeamStorageAssets?: ReturnType, ): AuthCommandContext { return { ui, @@ -77,10 +80,12 @@ function makeCtx( ok: true, value: envVars, }), - syncTeamStorageAssets: mock().mockResolvedValue({ - ok: true, - value: { downloadedCount: 0, reusedCount: 0, skippedCount: 0 }, - }), + syncTeamStorageAssets: + syncTeamStorageAssets ?? + mock().mockResolvedValue({ + ok: true, + value: { downloadedCount: 0, reusedCount: 0, skippedCount: 0 }, + }), }), }; } @@ -133,6 +138,37 @@ describe("handleFlowsPull json mode output", () => { expect(JSON.parse(JSON.stringify(payload))).toEqual(payload); }); + it("relays per-file asset download progress to the progress step", async () => { + await buildBundle(bundleArchive, { + flows: [{ name: "a.flow.ts", data: "// a\n" }], + }); + const updates: string[] = []; + const ui = makeJsonUi((message) => updates.push(message)); + const syncMock = mock( + async ( + _assetsAbs: string, + opts?: { + onProgress?: (p: { current: number; total: number }) => void; + }, + ) => { + opts?.onProgress?.({ current: 1, total: 3 }); + opts?.onProgress?.({ current: 2, total: 3 }); + return { + ok: true, + value: { downloadedCount: 3, reusedCount: 0, skippedCount: 0 }, + }; + }, + ); + const ctx = makeCtx(ui, bundleArchive, {}, syncMock); + + await handleFlowsPull(ctx, { env: "env-abc", out: destDir }); + + expect(updates).toEqual([ + "Downloading team-storage assets (1/3)", + "Downloading team-storage assets (2/3)", + ]); + }); + it("does not call ui.output when mode is not json", async () => { await buildBundle(bundleArchive, { flows: [{ name: "a.flow.ts", data: "// a\n" }], diff --git a/src/domains/flows/pull/handler.ts b/src/domains/flows/pull/handler.ts index 77019e52a..37ed67f5c 100644 --- a/src/domains/flows/pull/handler.ts +++ b/src/domains/flows/pull/handler.ts @@ -92,9 +92,19 @@ export async function handleFlowsPull( }, { message: flowsMessages.pull.downloadingTeamStorageAssets, - task: async () => { - const result = - await ctx.platformClient.syncTeamStorageAssets(assetsAbs); + task: async (update) => { + const result = await ctx.platformClient.syncTeamStorageAssets( + assetsAbs, + { + onProgress: ({ current, total }) => + update( + flowsMessages.pull.downloadingTeamStorageAssetsProgress( + current, + total, + ), + ), + }, + ); if (!result.ok) throw new Error(result.error); return result.value; }, diff --git a/src/shell/fs.testUtils.dirTree.test.ts b/src/shell/fs.testUtils.dirTree.test.ts new file mode 100644 index 000000000..e21cfd4d9 --- /dev/null +++ b/src/shell/fs.testUtils.dirTree.test.ts @@ -0,0 +1,149 @@ +import { describe, expect, it } from "bun:test"; +import { isNoEntError } from "~/core/errors.js"; +import { makeMemoryFs } from "./fs.testUtils.js"; + +describe("makeMemoryFs directory tree", () => { + // readdir + it("should return top-level entries when listing root", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/a"); + await fs.writeFile("/b.txt", ""); + const entries = await fs.readdir("/"); + expect(entries.sort()).toEqual(["a", "b.txt"]); + }); + + it("should return direct child names when directory exists", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/a"); + await fs.writeFile("/a/f.txt", ""); + await fs.mkdir("/a/sub"); + const entries = await fs.readdir("/a"); + expect(entries.sort()).toEqual(["f.txt", "sub"]); + }); + + it("should throw ENOENT when directory does not exist", async () => { + const fs = makeMemoryFs(); + let caughtError: unknown; + try { + await fs.readdir("/missing"); + } catch (e) { + caughtError = e; + } + expect(isNoEntError(caughtError)).toBe(true); + }); + + it("should throw ENOTDIR when path is a file", async () => { + const fs = makeMemoryFs(); + await fs.writeFile("/f.txt", "data"); + let caughtError: unknown; + try { + await fs.readdir("/f.txt"); + } catch (e) { + caughtError = e; + } + expect((caughtError as NodeJS.ErrnoException).code).toBe("ENOTDIR"); + }); + + // readdirWithTypes + it("should return correct types when listing root", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/dir"); + await fs.writeFile("/f.txt", ""); + const entries = await fs.readdirWithTypes("/"); + const d = entries.find((e) => e.name === "dir"); + const f = entries.find((e) => e.name === "f.txt"); + expect(d?.isDirectory()).toBe(true); + expect(d?.isFile()).toBe(false); + expect(f?.isFile()).toBe(true); + expect(f?.isDirectory()).toBe(false); + }); + + it("should return FsDirent with isFile true for files", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/d"); + await fs.writeFile("/d/a.txt", ""); + const entries = await fs.readdirWithTypes("/d"); + const f = entries.find((e) => e.name === "a.txt"); + expect(f?.isFile()).toBe(true); + expect(f?.isDirectory()).toBe(false); + }); + + it("should return FsDirent with isDirectory true for subdirs", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/p"); + await fs.mkdir("/p/sub"); + const entries = await fs.readdirWithTypes("/p"); + const d = entries.find((e) => e.name === "sub"); + expect(d?.isDirectory()).toBe(true); + expect(d?.isFile()).toBe(false); + }); + + it("should throw ENOTDIR from readdirWithTypes when path is a file", async () => { + const fs = makeMemoryFs(); + await fs.writeFile("/f.txt", "data"); + let caughtError: unknown; + try { + await fs.readdirWithTypes("/f.txt"); + } catch (e) { + caughtError = e; + } + expect((caughtError as NodeJS.ErrnoException).code).toBe("ENOTDIR"); + }); + + it("should snapshot isFile/isDirectory at readdir time, not reflect later mutations", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/d"); + await fs.writeFile("/d/f.txt", "data"); + const entries = await fs.readdirWithTypes("/d"); + const f = entries.find((e) => e.name === "f.txt")!; + expect(f.isFile()).toBe(true); + await fs.unlink("/d/f.txt"); + expect(f.isFile()).toBe(true); + }); + + // rename + it("should move a file to the new path", async () => { + const fs = makeMemoryFs(); + await fs.writeFile("/src.txt", "hello"); + await fs.rename("/src.txt", "/dst.txt"); + expect(await fs.readFile("/dst.txt")).toBe("hello"); + expect(await fs.pathExists("/src.txt")).toBe(false); + }); + + it("should move a directory and its children to the new path", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/src", { recursive: true }); + await fs.writeFile("/src/a.txt", "hello"); + await fs.mkdir("/src/sub"); + await fs.writeFile("/src/sub/b.txt", "world"); + await fs.mkdir("/dst-parent"); + await fs.rename("/src", "/dst-parent/dst"); + expect(await fs.readFile("/dst-parent/dst/a.txt")).toBe("hello"); + expect(await fs.readFile("/dst-parent/dst/sub/b.txt")).toBe("world"); + expect(await fs.pathExists("/src")).toBe(false); + expect(await fs.pathExists("/src/a.txt")).toBe(false); + }); + + it("should throw ENOENT when renaming a missing path", async () => { + const fs = makeMemoryFs(); + let caughtError: unknown; + try { + await fs.rename("/nope.txt", "/dst.txt"); + } catch (e) { + caughtError = e; + } + expect(isNoEntError(caughtError)).toBe(true); + }); + + it("should throw ENOENT when renaming a directory to a non-existent parent", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/src"); + let caughtError: unknown; + try { + await fs.rename("/src", "/missing/dst"); + } catch (e) { + caughtError = e; + } + expect(isNoEntError(caughtError)).toBe(true); + }); +}); diff --git a/src/shell/fs.testUtils.syncAndStreams.test.ts b/src/shell/fs.testUtils.syncAndStreams.test.ts new file mode 100644 index 000000000..cb8c580d4 --- /dev/null +++ b/src/shell/fs.testUtils.syncAndStreams.test.ts @@ -0,0 +1,182 @@ +import { describe, expect, it } from "bun:test"; +import { isNoEntError } from "~/core/errors.js"; +import { makeMemoryFs } from "./fs.testUtils.js"; + +describe("makeMemoryFs sync methods, streams, and write handles", () => { + // existsSync + it("should return true for an existing file", async () => { + const fs = makeMemoryFs(); + await fs.writeFile("/f.txt", ""); + expect(fs.existsSync("/f.txt")).toBe(true); + }); + + it("should return false for a missing path", () => { + const fs = makeMemoryFs(); + expect(fs.existsSync("/nowhere")).toBe(false); + }); + + // readFileSync + it("should return file content as string", async () => { + const fs = makeMemoryFs(); + await fs.writeFile("/f.txt", "content"); + expect(fs.readFileSync("/f.txt")).toBe("content"); + }); + + it("should throw ENOENT when file does not exist", () => { + const fs = makeMemoryFs(); + let caughtError: unknown; + try { + fs.readFileSync("/missing.txt"); + } catch (e) { + caughtError = e; + } + expect(isNoEntError(caughtError)).toBe(true); + }); + + // mkdirSync + it("should create a directory synchronously", () => { + const fs = makeMemoryFs(); + fs.mkdirSync("/d"); + expect(fs.existsSync("/d")).toBe(true); + }); + + it("should throw ENOENT for non-recursive when parent does not exist", () => { + const fs = makeMemoryFs(); + let caughtError: unknown; + try { + fs.mkdirSync("/a/b/c"); + } catch (e) { + caughtError = e; + } + expect(isNoEntError(caughtError)).toBe(true); + }); + + it("should create all ancestors when recursive is true", () => { + const fs = makeMemoryFs(); + fs.mkdirSync("/a/b/c", { recursive: true }); + expect(fs.existsSync("/a")).toBe(true); + expect(fs.existsSync("/a/b")).toBe(true); + expect(fs.existsSync("/a/b/c")).toBe(true); + }); + + // writeFileSync + it("should write a file synchronously, readable via readFileSync", () => { + const fs = makeMemoryFs(); + fs.writeFileSync("/f.txt", "hello"); + expect(fs.readFileSync("/f.txt")).toBe("hello"); + }); + + it("should throw ENOENT when the parent directory does not exist", () => { + const fs = makeMemoryFs(); + let caughtError: unknown; + try { + fs.writeFileSync("/missing/f.txt", "data"); + } catch (e) { + caughtError = e; + } + expect(isNoEntError(caughtError)).toBe(true); + }); + + // createReadStream + it("should stream file content through data events", async () => { + const fs = makeMemoryFs(); + await fs.writeFile("/f.txt", "hello"); + const stream = fs.createReadStream("/f.txt"); + const chunks: Buffer[] = []; + await new Promise((resolve, reject) => { + stream.on("data", (chunk: Buffer) => chunks.push(chunk)); + stream.on("end", resolve); + stream.on("error", reject); + }); + expect(Buffer.concat(chunks).toString()).toBe("hello"); + }); + + it("should throw ENOENT when streaming a file that does not exist", () => { + const fs = makeMemoryFs(); + let caughtError: unknown; + try { + fs.createReadStream("/missing.txt"); + } catch (e) { + caughtError = e; + } + expect(isNoEntError(caughtError)).toBe(true); + }); + + // utimes + it("should resolve without error", async () => { + const fs = makeMemoryFs(); + await fs.writeFile("/f.txt", ""); + expect( + fs.utimes("/f.txt", new Date(), new Date()), + ).resolves.toBeUndefined(); + }); + + // mkdir with mode option + it("should create directory when mode option is provided", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/d", { recursive: true, mode: 0o700 }); + expect(await fs.pathExists("/d")).toBe(true); + }); + + // writeFile with options + it("should write file when mode option is provided", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/d"); + await fs.writeFile("/d/f.txt", "data", { mode: 0o600 }); + expect(await fs.readFile("/d/f.txt")).toBe("data"); + }); + + // Path keying — tests build paths with a literal "/" while the code under + // test joins with node:path, which emits "\" and a drive prefix on win32. + it("should name the same entry from a win32-separated path as from its POSIX form", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/a/b", { recursive: true }); + await fs.writeFile("\\a\\b\\f.txt", "data"); + expect(await fs.readFile("/a/b/f.txt")).toBe("data"); + }); + + it("should name the same entry from a drive-qualified path as from its POSIX form", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/a", { recursive: true }); + await fs.writeFile("C:\\a\\f.txt", "data"); + expect(await fs.readFile("/a/f.txt")).toBe("data"); + }); + + it("should list a win32-separated write through a POSIX readdir", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("\\x\\y", { recursive: true }); + await fs.writeFile("\\x\\y\\f.txt", "data"); + expect(await fs.readdir("/x/y")).toEqual(["f.txt"]); + expect(await fs.pathExists("/x")).toBe(true); + }); + + it("should accumulate write-handle chunks into the file across close", async () => { + const fs = makeMemoryFs(); + const encoder = new TextEncoder(); + const handle = await fs.openWriteHandle("/streamed.bin"); + await handle.write(encoder.encode("abc")); + await handle.write(encoder.encode("def")); + await handle.close(); + expect(await fs.readFile("/streamed.bin")).toBe("abcdef"); + }); + + it("should replace existing content when a write handle opens the path", async () => { + const fs = makeMemoryFs(); + await fs.writeFile("/f.txt", "old content"); + const handle = await fs.openWriteHandle("/f.txt"); + await handle.write(new TextEncoder().encode("new")); + await handle.close(); + expect(await fs.readFile("/f.txt")).toBe("new"); + }); + + it("should throw ENOENT when a write handle opens under a missing parent", async () => { + const fs = makeMemoryFs(); + let caughtError: unknown; + try { + await fs.openWriteHandle("/no/such/dir/f.bin"); + } catch (e) { + caughtError = e; + } + expect(isNoEntError(caughtError)).toBe(true); + }); +}); diff --git a/src/shell/fs.testUtils.test.ts b/src/shell/fs.testUtils.test.ts index 3208a5b82..4d1b4850d 100644 --- a/src/shell/fs.testUtils.test.ts +++ b/src/shell/fs.testUtils.test.ts @@ -1,4 +1,3 @@ -// oxlint-disable eslint/max-lines -- test file covers all makeMemoryFs methods; splitting would fragment related fixtures import { describe, expect, it } from "bun:test"; import { isNoEntError } from "~/core/errors.js"; import { makeMemoryFs } from "./fs.testUtils.js"; @@ -155,295 +154,4 @@ describe("makeMemoryFs", () => { expect(await fs.pathExists("/x/y")).toBe(true); expect(await fs.pathExists("/x")).toBe(true); }); - - // readdir - it("should return top-level entries when listing root", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/a"); - await fs.writeFile("/b.txt", ""); - const entries = await fs.readdir("/"); - expect(entries.sort()).toEqual(["a", "b.txt"]); - }); - - it("should return direct child names when directory exists", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/a"); - await fs.writeFile("/a/f.txt", ""); - await fs.mkdir("/a/sub"); - const entries = await fs.readdir("/a"); - expect(entries.sort()).toEqual(["f.txt", "sub"]); - }); - - it("should throw ENOENT when directory does not exist", async () => { - const fs = makeMemoryFs(); - let caughtError: unknown; - try { - await fs.readdir("/missing"); - } catch (e) { - caughtError = e; - } - expect(isNoEntError(caughtError)).toBe(true); - }); - - it("should throw ENOTDIR when path is a file", async () => { - const fs = makeMemoryFs(); - await fs.writeFile("/f.txt", "data"); - let caughtError: unknown; - try { - await fs.readdir("/f.txt"); - } catch (e) { - caughtError = e; - } - expect((caughtError as NodeJS.ErrnoException).code).toBe("ENOTDIR"); - }); - - // readdirWithTypes - it("should return correct types when listing root", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/dir"); - await fs.writeFile("/f.txt", ""); - const entries = await fs.readdirWithTypes("/"); - const d = entries.find((e) => e.name === "dir"); - const f = entries.find((e) => e.name === "f.txt"); - expect(d?.isDirectory()).toBe(true); - expect(d?.isFile()).toBe(false); - expect(f?.isFile()).toBe(true); - expect(f?.isDirectory()).toBe(false); - }); - - it("should return FsDirent with isFile true for files", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/d"); - await fs.writeFile("/d/a.txt", ""); - const entries = await fs.readdirWithTypes("/d"); - const f = entries.find((e) => e.name === "a.txt"); - expect(f?.isFile()).toBe(true); - expect(f?.isDirectory()).toBe(false); - }); - - it("should return FsDirent with isDirectory true for subdirs", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/p"); - await fs.mkdir("/p/sub"); - const entries = await fs.readdirWithTypes("/p"); - const d = entries.find((e) => e.name === "sub"); - expect(d?.isDirectory()).toBe(true); - expect(d?.isFile()).toBe(false); - }); - - it("should throw ENOTDIR when path is a file", async () => { - const fs = makeMemoryFs(); - await fs.writeFile("/f.txt", "data"); - let caughtError: unknown; - try { - await fs.readdirWithTypes("/f.txt"); - } catch (e) { - caughtError = e; - } - expect((caughtError as NodeJS.ErrnoException).code).toBe("ENOTDIR"); - }); - - it("should snapshot isFile/isDirectory at readdir time, not reflect later mutations", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/d"); - await fs.writeFile("/d/f.txt", "data"); - const entries = await fs.readdirWithTypes("/d"); - const f = entries.find((e) => e.name === "f.txt")!; - expect(f.isFile()).toBe(true); - await fs.unlink("/d/f.txt"); - expect(f.isFile()).toBe(true); - }); - - // rename - it("should move a file to the new path", async () => { - const fs = makeMemoryFs(); - await fs.writeFile("/src.txt", "hello"); - await fs.rename("/src.txt", "/dst.txt"); - expect(await fs.readFile("/dst.txt")).toBe("hello"); - expect(await fs.pathExists("/src.txt")).toBe(false); - }); - - it("should move a directory and its children to the new path", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/src", { recursive: true }); - await fs.writeFile("/src/a.txt", "hello"); - await fs.mkdir("/src/sub"); - await fs.writeFile("/src/sub/b.txt", "world"); - await fs.mkdir("/dst-parent"); - await fs.rename("/src", "/dst-parent/dst"); - expect(await fs.readFile("/dst-parent/dst/a.txt")).toBe("hello"); - expect(await fs.readFile("/dst-parent/dst/sub/b.txt")).toBe("world"); - expect(await fs.pathExists("/src")).toBe(false); - expect(await fs.pathExists("/src/a.txt")).toBe(false); - }); - - it("should throw ENOENT when renaming a missing path", async () => { - const fs = makeMemoryFs(); - let caughtError: unknown; - try { - await fs.rename("/nope.txt", "/dst.txt"); - } catch (e) { - caughtError = e; - } - expect(isNoEntError(caughtError)).toBe(true); - }); - - it("should throw ENOENT when renaming a directory to a non-existent parent", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/src"); - let caughtError: unknown; - try { - await fs.rename("/src", "/missing/dst"); - } catch (e) { - caughtError = e; - } - expect(isNoEntError(caughtError)).toBe(true); - }); - - // existsSync - it("should return true for an existing file", async () => { - const fs = makeMemoryFs(); - await fs.writeFile("/f.txt", ""); - expect(fs.existsSync("/f.txt")).toBe(true); - }); - - it("should return false for a missing path", () => { - const fs = makeMemoryFs(); - expect(fs.existsSync("/nowhere")).toBe(false); - }); - - // readFileSync - it("should return file content as string", async () => { - const fs = makeMemoryFs(); - await fs.writeFile("/f.txt", "content"); - expect(fs.readFileSync("/f.txt")).toBe("content"); - }); - - it("should throw ENOENT when file does not exist", () => { - const fs = makeMemoryFs(); - let caughtError: unknown; - try { - fs.readFileSync("/missing.txt"); - } catch (e) { - caughtError = e; - } - expect(isNoEntError(caughtError)).toBe(true); - }); - - // mkdirSync - it("should create a directory synchronously", () => { - const fs = makeMemoryFs(); - fs.mkdirSync("/d"); - expect(fs.existsSync("/d")).toBe(true); - }); - - it("should throw ENOENT for non-recursive when parent does not exist", () => { - const fs = makeMemoryFs(); - let caughtError: unknown; - try { - fs.mkdirSync("/a/b/c"); - } catch (e) { - caughtError = e; - } - expect(isNoEntError(caughtError)).toBe(true); - }); - - it("should create all ancestors when recursive is true", () => { - const fs = makeMemoryFs(); - fs.mkdirSync("/a/b/c", { recursive: true }); - expect(fs.existsSync("/a")).toBe(true); - expect(fs.existsSync("/a/b")).toBe(true); - expect(fs.existsSync("/a/b/c")).toBe(true); - }); - - // writeFileSync - it("should write a file synchronously, readable via readFileSync", () => { - const fs = makeMemoryFs(); - fs.writeFileSync("/f.txt", "hello"); - expect(fs.readFileSync("/f.txt")).toBe("hello"); - }); - - it("should throw ENOENT when the parent directory does not exist", () => { - const fs = makeMemoryFs(); - let caughtError: unknown; - try { - fs.writeFileSync("/missing/f.txt", "data"); - } catch (e) { - caughtError = e; - } - expect(isNoEntError(caughtError)).toBe(true); - }); - - // createReadStream - it("should stream file content through data events", async () => { - const fs = makeMemoryFs(); - await fs.writeFile("/f.txt", "hello"); - const stream = fs.createReadStream("/f.txt"); - const chunks: Buffer[] = []; - await new Promise((resolve, reject) => { - stream.on("data", (chunk: Buffer) => chunks.push(chunk)); - stream.on("end", resolve); - stream.on("error", reject); - }); - expect(Buffer.concat(chunks).toString()).toBe("hello"); - }); - - it("should throw ENOENT when file does not exist", () => { - const fs = makeMemoryFs(); - let caughtError: unknown; - try { - fs.createReadStream("/missing.txt"); - } catch (e) { - caughtError = e; - } - expect(isNoEntError(caughtError)).toBe(true); - }); - - // utimes - it("should resolve without error", async () => { - const fs = makeMemoryFs(); - await fs.writeFile("/f.txt", ""); - expect( - fs.utimes("/f.txt", new Date(), new Date()), - ).resolves.toBeUndefined(); - }); - - // mkdir with mode option - it("should create directory when mode option is provided", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/d", { recursive: true, mode: 0o700 }); - expect(await fs.pathExists("/d")).toBe(true); - }); - - // writeFile with options - it("should write file when mode option is provided", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/d"); - await fs.writeFile("/d/f.txt", "data", { mode: 0o600 }); - expect(await fs.readFile("/d/f.txt")).toBe("data"); - }); - - // Path keying — tests build paths with a literal "/" while the code under - // test joins with node:path, which emits "\" and a drive prefix on win32. - it("should name the same entry from a win32-separated path as from its POSIX form", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/a/b", { recursive: true }); - await fs.writeFile("\\a\\b\\f.txt", "data"); - expect(await fs.readFile("/a/b/f.txt")).toBe("data"); - }); - - it("should name the same entry from a drive-qualified path as from its POSIX form", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("/a", { recursive: true }); - await fs.writeFile("C:\\a\\f.txt", "data"); - expect(await fs.readFile("/a/f.txt")).toBe("data"); - }); - - it("should list a win32-separated write through a POSIX readdir", async () => { - const fs = makeMemoryFs(); - await fs.mkdir("\\x\\y", { recursive: true }); - await fs.writeFile("\\x\\y\\f.txt", "data"); - expect(await fs.readdir("/x/y")).toEqual(["f.txt"]); - expect(await fs.pathExists("/x")).toBe(true); - }); }); diff --git a/src/shell/fs.testUtils.ts b/src/shell/fs.testUtils.ts index 22e016333..39011a51a 100644 --- a/src/shell/fs.testUtils.ts +++ b/src/shell/fs.testUtils.ts @@ -1,32 +1,15 @@ import type { FsDirent, Fs } from "./fs.js"; -import { posix } from "node:path"; import { Readable } from "node:stream"; -const dirname = posix.dirname; - -// Callers build paths with a literal "/", but the code under test joins with -// node:path, which emits "\" and a drive prefix on win32. Key every entry in -// one POSIX form so both spellings name the same entry. -function toKey(path: string) { - return path.replace(/\\/g, "/").replace(/^[A-Za-z]:/, ""); -} - -function throwNoEntError( - path: string, - kind: "mkdir" | "open" | "rm" | "stat" | "unlink", -): never { - throw Object.assign( - new Error(`ENOENT: no such file or directory, ${kind} '${path}'`), - { code: "ENOENT" }, - ); -} - -function throwNotDirError(path: string, syscall: string): never { - throw Object.assign( - new Error(`ENOTDIR: not a directory, ${syscall} '${path}'`), - { code: "ENOTDIR" }, - ); -} +import { + addParents, + childNames, + joinPath, + requireParent, + throwNoEntError, + throwNotDirError, + toKey, +} from "./memoryFsTree.testUtils.js"; export function makeMemoryFs(): Fs { const files = new Map(); @@ -34,53 +17,12 @@ export function makeMemoryFs(): Fs { const textEncoder = new TextEncoder(); const textDecoder = new TextDecoder(); - function joinPath(dir: string, name: string) { - return dir === "/" ? `/${name}` : `${dir}/${name}`; - } - - function addParents(filePath: string) { - let dir = dirname(filePath); - while (dir !== "." && dir !== "/") { - dirs.add(dir); - dir = dirname(dir); - } - if (dir === "/") dirs.add(dir); - } - - function childNames(path: string) { - const prefix = path === "/" ? "/" : path + "/"; - const names = new Set(); - for (const f of files.keys()) { - if (f.startsWith(prefix)) { - const segment = f.slice(prefix.length).split("/")[0]; - if (segment) names.add(segment); - } - } - for (const d of dirs) { - if (d !== path && d.startsWith(prefix)) { - const rel = d.slice(prefix.length); - if (!rel.includes("/")) names.add(rel); - } - } - return [...names]; - } - - // `rawPath` is only for the error message, so it echoes the caller's spelling. - function requireParent( - path: string, - rawPath: string, - kind: "mkdir" | "open", - ) { - const parent = dirname(path); - if (parent !== "/" && !dirs.has(parent)) throwNoEntError(rawPath, kind); - } - return { async mkdir(rawPath, opts) { const path = toKey(rawPath); - if (!opts?.recursive) requireParent(path, rawPath, "mkdir"); + if (!opts?.recursive) requireParent(dirs, path, rawPath, "mkdir"); dirs.add(path); - if (opts?.recursive) addParents(path); + if (opts?.recursive) addParents(dirs, path); }, async pathExists(rawPath) { const path = toKey(rawPath); @@ -147,24 +89,39 @@ export function makeMemoryFs(): Fs { }, async writeFile(rawPath, data, _options) { const path = toKey(rawPath); - requireParent(path, rawPath, "open"); + requireParent(dirs, path, rawPath, "open"); files.set( path, typeof data === "string" ? textEncoder.encode(data) : data, ); }, + async openWriteHandle(rawPath) { + const path = toKey(rawPath); + requireParent(dirs, path, rawPath, "open"); + files.set(path, new Uint8Array(0)); + return { + async write(chunk) { + const existing = files.get(path) ?? new Uint8Array(0); + const grown = new Uint8Array(existing.length + chunk.length); + grown.set(existing, 0); + grown.set(chunk, existing.length); + files.set(path, grown); + }, + async close() {}, + }; + }, readdir(rawPath) { const path = toKey(rawPath); if (files.has(path)) throwNotDirError(rawPath, "scandir"); if (!dirs.has(path)) throwNoEntError(rawPath, "open"); - return Promise.resolve(childNames(path)); + return Promise.resolve(childNames(files, dirs, path)); }, readdirWithTypes(rawPath) { const path = toKey(rawPath); if (files.has(path)) throwNotDirError(rawPath, "scandir"); if (!dirs.has(path)) throwNoEntError(rawPath, "open"); return Promise.resolve( - childNames(path).map((name) => { + childNames(files, dirs, path).map((name) => { const fullPath = joinPath(path, name); const isFileSnapshot = files.has(fullPath); const isDirSnapshot = dirs.has(fullPath); @@ -180,13 +137,13 @@ export function makeMemoryFs(): Fs { const oldPath = toKey(rawOldPath); const newPath = toKey(rawNewPath); if (files.has(oldPath)) { - requireParent(newPath, rawNewPath, "open"); + requireParent(dirs, newPath, rawNewPath, "open"); files.set(newPath, files.get(oldPath)!); files.delete(oldPath); return Promise.resolve(); } if (dirs.has(oldPath)) { - requireParent(newPath, rawNewPath, "open"); + requireParent(dirs, newPath, rawNewPath, "open"); const oldPrefix = oldPath + "/"; const newPrefix = newPath + "/"; dirs.delete(oldPath); @@ -219,7 +176,7 @@ export function makeMemoryFs(): Fs { const destination = toKey(rawDestination); const data = files.get(toKey(rawSource)); if (data === undefined) throwNoEntError(rawSource, "open"); - requireParent(destination, rawDestination, "open"); + requireParent(dirs, destination, rawDestination, "open"); files.set(destination, data.slice()); }, existsSync(rawPath) { @@ -233,14 +190,14 @@ export function makeMemoryFs(): Fs { }, writeFileSync(rawPath, data) { const path = toKey(rawPath); - requireParent(path, rawPath, "open"); + requireParent(dirs, path, rawPath, "open"); files.set(path, textEncoder.encode(data)); }, mkdirSync(rawPath, opts) { const path = toKey(rawPath); - if (!opts?.recursive) requireParent(path, rawPath, "mkdir"); + if (!opts?.recursive) requireParent(dirs, path, rawPath, "mkdir"); dirs.add(path); - if (opts?.recursive) addParents(path); + if (opts?.recursive) addParents(dirs, path); }, }; } diff --git a/src/shell/fs.ts b/src/shell/fs.ts index f7b4e0e25..d77958123 100644 --- a/src/shell/fs.ts +++ b/src/shell/fs.ts @@ -2,6 +2,9 @@ import type { Readable } from "node:stream"; import * as fs from "node:fs"; import { isNoEntError } from "~/core/errors.js"; +import { openFsWriteHandle, type FsWriteHandle } from "./fsWriteHandle.js"; + +export type { FsWriteHandle } from "./fsWriteHandle.js"; export async function pathExists(p: string): Promise { try { @@ -42,6 +45,7 @@ export type Fs = { data: string | Uint8Array, options?: { mode?: number }, ): Promise; + openWriteHandle(path: string): Promise; readdir(path: string): Promise; readdirWithTypes(path: string): Promise; rename(oldPath: string, newPath: string): Promise; @@ -82,6 +86,7 @@ export function makeDefaultFs(): Fs { await fs.promises.writeFile(path, data, options ?? undefined); } }, + openWriteHandle: openFsWriteHandle, readdir(path) { return fs.promises.readdir(path); }, diff --git a/src/shell/fsWriteHandle.test.ts b/src/shell/fsWriteHandle.test.ts new file mode 100644 index 000000000..558fd1ca6 --- /dev/null +++ b/src/shell/fsWriteHandle.test.ts @@ -0,0 +1,51 @@ +import { afterEach, describe, expect, it, mock, spyOn } from "bun:test"; +import * as fs from "node:fs"; +import type { FileHandle } from "node:fs/promises"; + +import { openFsWriteHandle } from "./fsWriteHandle.js"; + +afterEach(() => { + mock.restore(); +}); + +function asFileHandle(value: unknown): FileHandle { + return value as FileHandle; +} + +describe("openFsWriteHandle", () => { + // POSIX write(2) may write fewer bytes than requested (e.g. near ENOSPC); + // a short write must not silently drop the unwritten suffix. + it("retries the unwritten suffix when the OS reports a short write", async () => { + const written: number[] = []; + const shortWritingHandle = { + write: async (chunk: Uint8Array) => { + const bytesWritten = Math.min(2, chunk.length); + written.push(...chunk.subarray(0, bytesWritten)); + return { bytesWritten }; + }, + close: async () => {}, + }; + spyOn(fs.promises, "open").mockResolvedValue( + asFileHandle(shortWritingHandle), + ); + + const handle = await openFsWriteHandle("/tmp/out.bin"); + await handle.write(new TextEncoder().encode("abcdef")); + + expect(new TextDecoder().decode(new Uint8Array(written))).toBe("abcdef"); + }); + + it("fails instead of looping when a write reports zero bytes", async () => { + const stuckHandle = { + write: async () => ({ bytesWritten: 0 }), + close: async () => {}, + }; + spyOn(fs.promises, "open").mockResolvedValue(asFileHandle(stuckHandle)); + + const handle = await openFsWriteHandle("/tmp/out.bin"); + + expect(handle.write(new TextEncoder().encode("abc"))).rejects.toThrow( + "0 bytes", + ); + }); +}); diff --git a/src/shell/fsWriteHandle.ts b/src/shell/fsWriteHandle.ts new file mode 100644 index 000000000..57334bc8a --- /dev/null +++ b/src/shell/fsWriteHandle.ts @@ -0,0 +1,33 @@ +import * as fs from "node:fs"; + +/** + * An open file being written incrementally. `write` appends one chunk; + * `close` releases the descriptor and must be called on every path, + * including failures. + */ +export type FsWriteHandle = { + write(chunk: Uint8Array): Promise; + close(): Promise; +}; + +export async function openFsWriteHandle(path: string): Promise { + const handle = await fs.promises.open(path, "w"); + return { + async write(chunk) { + // POSIX write(2) may write fewer bytes than requested (e.g. near + // ENOSPC); loop over the unwritten suffix so a short write cannot + // silently truncate the file. + let remaining = chunk; + while (remaining.length > 0) { + const { bytesWritten } = await handle.write(remaining); + if (bytesWritten === 0) { + throw new Error(`write returned 0 bytes for '${path}'`); + } + remaining = remaining.subarray(bytesWritten); + } + }, + close() { + return handle.close(); + }, + }; +} diff --git a/src/shell/memoryFsTree.testUtils.ts b/src/shell/memoryFsTree.testUtils.ts new file mode 100644 index 000000000..c8992cefe --- /dev/null +++ b/src/shell/memoryFsTree.testUtils.ts @@ -0,0 +1,73 @@ +import { posix } from "node:path"; + +const dirname = posix.dirname; + +// Callers build paths with a literal "/", but the code under test joins with +// node:path, which emits "\" and a drive prefix on win32. Key every entry in +// one POSIX form so both spellings name the same entry. +export function toKey(path: string): string { + return path.replace(/\\/g, "/").replace(/^[A-Za-z]:/, ""); +} + +export function throwNoEntError( + path: string, + kind: "mkdir" | "open" | "rm" | "stat" | "unlink", +): never { + throw Object.assign( + new Error(`ENOENT: no such file or directory, ${kind} '${path}'`), + { code: "ENOENT" }, + ); +} + +export function throwNotDirError(path: string, syscall: string): never { + throw Object.assign( + new Error(`ENOTDIR: not a directory, ${syscall} '${path}'`), + { code: "ENOTDIR" }, + ); +} + +export function joinPath(dir: string, name: string): string { + return dir === "/" ? `/${name}` : `${dir}/${name}`; +} + +export function addParents(dirs: Set, filePath: string): void { + let dir = dirname(filePath); + while (dir !== "." && dir !== "/") { + dirs.add(dir); + dir = dirname(dir); + } + if (dir === "/") dirs.add(dir); +} + +export function childNames( + files: ReadonlyMap, + dirs: ReadonlySet, + path: string, +): string[] { + const prefix = path === "/" ? "/" : path + "/"; + const names = new Set(); + for (const f of files.keys()) { + if (f.startsWith(prefix)) { + const segment = f.slice(prefix.length).split("/")[0]; + if (segment) names.add(segment); + } + } + for (const d of dirs) { + if (d !== path && d.startsWith(prefix)) { + const rel = d.slice(prefix.length); + if (!rel.includes("/")) names.add(rel); + } + } + return [...names]; +} + +// `rawPath` is only for the error message, so it echoes the caller's spelling. +export function requireParent( + dirs: ReadonlySet, + path: string, + rawPath: string, + kind: "mkdir" | "open", +): void { + const parent = dirname(path); + if (parent !== "/" && !dirs.has(parent)) throwNoEntError(rawPath, kind); +} diff --git a/src/shell/platform/createPlatformClient.ts b/src/shell/platform/createPlatformClient.ts index 487b3f6f3..fa7019b67 100644 --- a/src/shell/platform/createPlatformClient.ts +++ b/src/shell/platform/createPlatformClient.ts @@ -18,6 +18,7 @@ import { downloadTeamStorageAssets, type SyncTeamStorageAssetsResult, } from "./teamStorageAssets.js"; +import type { TeamStorageAssetProgress } from "./writeAssetSnapshot.js"; import { environmentWithVariablesResponseSchema, flowsBundleResponseSchema, @@ -36,6 +37,9 @@ export type PlatformClient = { listTeamStorageFiles: () => Promise>; syncTeamStorageAssets: ( assetsAbs: string, + opts?: { + onProgress?: (progress: TeamStorageAssetProgress) => void; + }, ) => Promise>; downloadBundle: ( envId: string, @@ -124,12 +128,12 @@ export function createPlatformClient( ); }, - async syncTeamStorageAssets(assetsAbs) { + async syncTeamStorageAssets(assetsAbs, opts) { const files = await this.listTeamStorageFiles(); if (!files.ok) return files; return downloadTeamStorageAssets( { assetsAbs, files: files.value }, - { fetch: deps.fetch, fs }, + { fetch: deps.fetch, fs, onProgress: opts?.onProgress }, ); }, diff --git a/src/shell/platform/describeErrors.test.ts b/src/shell/platform/describeErrors.test.ts index b115fb10a..8a4b37f85 100644 --- a/src/shell/platform/describeErrors.test.ts +++ b/src/shell/platform/describeErrors.test.ts @@ -1,6 +1,10 @@ import { describe, expect, it } from "bun:test"; -import { describeRequestError } from "./describeErrors.js"; +import { + describeBundleDownloadError, + describeRequestError, + describeTeamStorageDownloadError, +} from "./describeErrors.js"; const baseUrl = "https://app.qawolf.com"; const reason = "Your API key has no access to this environment."; @@ -67,3 +71,31 @@ describe("describeRequestError", () => { expect("errorBody" in described).toBe(false); }); }); + +// Signed-URL downloads use a stall timeout that resets while bytes arrive, so +// the message must describe a stall, not a whole-download deadline. +describe("describeBundleDownloadError", () => { + it("describes a timeout as a stall", () => { + const message = describeBundleDownloadError({ + kind: "timeout", + timeoutMs: 30_000, + }); + + expect(message).toBe( + "Downloading the flow bundle stalled — no data arrived for 30s. Please try again.", + ); + }); +}); + +describe("describeTeamStorageDownloadError", () => { + it("describes a timeout as a stall", () => { + const message = describeTeamStorageDownloadError("interview-video.y4m", { + kind: "timeout", + timeoutMs: 30_000, + }); + + expect(message).toBe( + "Downloading the team-storage asset interview-video.y4m stalled — no data arrived for 30s. Please try again.", + ); + }); +}); diff --git a/src/shell/platform/describeErrors.ts b/src/shell/platform/describeErrors.ts index af32b67fa..a63c010f4 100644 --- a/src/shell/platform/describeErrors.ts +++ b/src/shell/platform/describeErrors.ts @@ -81,7 +81,7 @@ export function describeTeamStorageDownloadError( return `Could not reach team-storage while downloading ${path}. Check your network connection and try again.`; } if (err.kind === "timeout") { - return `Downloading the team-storage asset ${path} timed out after ${formatSeconds(err.timeoutMs)}. Please try again.`; + return `Downloading the team-storage asset ${path} stalled — no data arrived for ${formatSeconds(err.timeoutMs)}. Please try again.`; } return `The team-storage download for ${path} was malformed. Please run \`qawolf flows pull\` again.`; } diff --git a/src/shell/platform/fetchSignedUrl.disk.test.ts b/src/shell/platform/fetchSignedUrl.disk.test.ts new file mode 100644 index 000000000..0993f7e5d --- /dev/null +++ b/src/shell/platform/fetchSignedUrl.disk.test.ts @@ -0,0 +1,144 @@ +import { afterEach, describe, expect, it, mock } from "bun:test"; + +import { makeMemoryFs } from "~/shell/fs.testUtils.js"; +import { fetchSignedUrl } from "./fetchSignedUrl.js"; +import { makeDrippingBodyFetch } from "./slowFetch.testUtils.js"; + +afterEach(() => { + mock.restore(); +}); + +const url = + "https://storage.googleapis.com/bucket/file.tar.gz?X-Goog-Signature=abc"; + +function createFetchMock(response: Response) { + return mock().mockResolvedValue(response); +} + +function asFetch(value: unknown): typeof fetch { + return value as typeof fetch; +} + +describe("fetchSignedUrl disk interaction", () => { + // Streaming to disk keeps peak memory at one chunk, not 2× the file size — + // large assets must not be buffered whole before writing. + it("streams each chunk to disk as it arrives instead of buffering the body", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/assets", { recursive: true }); + const chunkWrites: number[] = []; + const openWriteHandle = fs.openWriteHandle.bind(fs); + fs.openWriteHandle = async (path) => { + const handle = await openWriteHandle(path); + return { + write: async (chunk) => { + chunkWrites.push(chunk.length); + await handle.write(chunk); + }, + close: handle.close, + }; + }; + + const result = await fetchSignedUrl( + { dest: "/assets/file.bin", url }, + { + fetch: makeDrippingBodyFetch(["ab", "cd", "ef"], 10), + fs, + stallTimeoutMs: 200, + }, + ); + + expect(result).toEqual({ ok: true, data: undefined }); + expect(chunkWrites).toEqual([2, 2, 2]); + expect(await fs.readFile("/assets/file.bin")).toBe("abcdef"); + expect(await fs.pathExists("/assets/file.bin.part")).toBe(false); + }); + + // The stall clock measures the network, not the disk: a write that outlasts + // the window must not abort a download whose bytes keep arriving. + it("does not count slow disk writes toward the stall window", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/assets", { recursive: true }); + const openWriteHandle = fs.openWriteHandle.bind(fs); + fs.openWriteHandle = async (path) => { + const handle = await openWriteHandle(path); + return { + write: async (chunk) => { + await new Promise((resolve) => setTimeout(resolve, 80)); + await handle.write(chunk); + }, + close: handle.close, + }; + }; + + const result = await fetchSignedUrl( + { dest: "/assets/file.bin", url }, + { + fetch: makeDrippingBodyFetch(["ab", "cd"], 10), + fs, + stallTimeoutMs: 50, + }, + ); + + expect(result).toEqual({ ok: true, data: undefined }); + expect(await fs.readFile("/assets/file.bin")).toBe("abcd"); + }); + + // An early error return must not leave the transfer running: nothing else + // ever cancels it, so the connection would stay open and Bun would keep + // buffering the body in the background. + it("aborts the request when the destination file cannot be opened", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/assets", { recursive: true }); + const cause = new Error("permission denied"); + fs.openWriteHandle = () => Promise.reject(cause); + const fetchSpy = createFetchMock(new Response("bytes")); + + const result = await fetchSignedUrl( + { dest: "/assets/file.txt", url }, + { fetch: asFetch(fetchSpy), fs }, + ); + + expect(result).toEqual({ ok: false, error: { cause, kind: "network" } }); + expect(fetchSpy.mock.calls[0]?.[1]?.signal?.aborted).toBe(true); + }); + + it("aborts the request when a chunk write fails", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/assets", { recursive: true }); + const cause = new Error("disk full"); + fs.openWriteHandle = async () => ({ + write: () => Promise.reject(cause), + close: async () => {}, + }); + const fetchSpy = createFetchMock(new Response("bytes")); + + const result = await fetchSignedUrl( + { dest: "/assets/file.txt", url }, + { fetch: asFetch(fetchSpy), fs }, + ); + + expect(result).toEqual({ ok: false, error: { cause, kind: "network" } }); + expect(fetchSpy.mock.calls[0]?.[1]?.signal?.aborted).toBe(true); + }); + + // Failed local writes must not reclassify as network stalls, and a partial + // file must not survive the failure. + it("returns a network error and removes the partial file when a write fails", async () => { + const fs = makeMemoryFs(); + await fs.mkdir("/assets", { recursive: true }); + const cause = new Error("disk full"); + fs.openWriteHandle = async () => ({ + write: () => Promise.reject(cause), + close: async () => {}, + }); + + const result = await fetchSignedUrl( + { dest: "/assets/file.txt", url }, + { fetch: asFetch(createFetchMock(new Response("bytes"))), fs }, + ); + + expect(result).toEqual({ ok: false, error: { cause, kind: "network" } }); + expect(await fs.pathExists("/assets/file.txt")).toBe(false); + expect(await fs.pathExists("/assets/file.txt.part")).toBe(false); + }); +}); diff --git a/src/shell/platform/fetchSignedUrl.test.ts b/src/shell/platform/fetchSignedUrl.test.ts index 4e4adf0df..e2905dfb7 100644 --- a/src/shell/platform/fetchSignedUrl.test.ts +++ b/src/shell/platform/fetchSignedUrl.test.ts @@ -5,7 +5,11 @@ import { join } from "node:path"; import { makeMemoryFs } from "~/shell/fs.testUtils.js"; import { fetchSignedUrl } from "./fetchSignedUrl.js"; -import { makeTimingOutBodyFetch } from "./slowFetch.testUtils.js"; +import { + makeDrippingBodyFetch, + makeStallingBodyFetch, + makeTimingOutBodyFetch, +} from "./slowFetch.testUtils.js"; const cleanups: (() => void)[] = []; @@ -125,17 +129,36 @@ describe("fetchSignedUrl", () => { }); }); - // The split that made the timeout visible must not reclassify a failed write. - it("still returns a network error when the body arrives but the write fails", async () => { - const fs = makeMemoryFs(); - const cause = new Error("disk full"); - fs.writeFile = mock(() => Promise.reject(cause)); + // The window is a stall timeout, not a whole-download deadline: it fires only + // when no bytes arrive for its duration. + it("returns a timeout when no data arrives within the stall window", async () => { + const dest = createTempDest(); const result = await fetchSignedUrl( - { dest: "/assets/file.txt", url }, - { fetch: asFetch(createFetchMock(new Response("bytes"))), fs }, + { dest, url }, + { fetch: makeStallingBodyFetch(["first bytes"]), stallTimeoutMs: 50 }, + ); + + expect(result).toEqual({ + ok: false, + error: { kind: "timeout", timeoutMs: 50 }, + }); + }); + + // A large asset on a slow link takes longer than any fixed deadline; as long + // as bytes keep arriving the download must be allowed to finish. + it("succeeds when the download outlasts the stall window but keeps making progress", async () => { + const dest = createTempDest(); + + const result = await fetchSignedUrl( + { dest, url }, + { + fetch: makeDrippingBodyFetch(["a", "b", "c", "d", "e", "f"], 40), + stallTimeoutMs: 100, + }, ); - expect(result).toEqual({ ok: false, error: { cause, kind: "network" } }); + expect(result).toEqual({ ok: true, data: undefined }); + expect(readFileSync(dest, "utf8")).toBe("abcdef"); }); }); diff --git a/src/shell/platform/fetchSignedUrl.ts b/src/shell/platform/fetchSignedUrl.ts index b17932e13..82cf51280 100644 --- a/src/shell/platform/fetchSignedUrl.ts +++ b/src/shell/platform/fetchSignedUrl.ts @@ -1,66 +1,127 @@ import { isTimeoutError } from "~/core/errors.js"; -import { makeDefaultFs, type Fs } from "~/shell/fs.js"; +import { makeDefaultFs, type Fs, type FsWriteHandle } from "~/shell/fs.js"; import type { WireResult } from "./createTrpcClient.js"; import { toError } from "./toError.js"; type Deps = { fetch: typeof globalThis.fetch; fs?: Fs | undefined; + stallTimeoutMs?: number | undefined; }; -const timeoutMs = 30_000; +const defaultStallTimeoutMs = 30_000; + +// bun-types has no global ReadableStreamReadResult and types body reads as +// `any`; this mirrors the default reader's result shape. +type BodyReadResult = + | { done: false; value: Uint8Array } + | { done: true; value: undefined }; export async function fetchSignedUrl( args: { url: string; dest: string }, deps: Deps = { fetch: globalThis.fetch }, ): Promise> { - let response: Response; + const timeoutMs = deps.stallTimeoutMs ?? defaultStallTimeoutMs; + const fs = deps.fs ?? makeDefaultFs(); + // Chunks stream to a sibling .part file so peak memory stays at one chunk + // regardless of asset size; the finished file is renamed into place. + const partPath = `${args.dest}.part`; + + // The window is a stall timeout, not a whole-download deadline: it resets + // every time bytes arrive, so a slow-but-progressing download of any size can + // finish while a genuine stall still fails. Aborting the request signal makes + // fetch reject a pending body read with the abort reason. + const controller = new AbortController(); + let stallTimer: ReturnType | undefined; + const armStallTimer = () => { + clearTimeout(stallTimer); + stallTimer = setTimeout( + () => + controller.abort( + new DOMException("The operation timed out.", "TimeoutError"), + ), + timeoutMs, + ); + }; + try { - response = await deps.fetch(args.url, { - signal: AbortSignal.timeout(timeoutMs), - }); - } catch (error: unknown) { - if (isTimeoutError(error)) { - return { ok: false, error: { kind: "timeout", timeoutMs } }; + armStallTimer(); + let response: Response; + try { + response = await deps.fetch(args.url, { signal: controller.signal }); + } catch (error: unknown) { + if (isTimeoutError(error)) { + return { ok: false, error: { kind: "timeout", timeoutMs } }; + } + return { ok: false, error: { cause: toError(error), kind: "network" } }; } - return { ok: false, error: { cause: toError(error), kind: "network" } }; - } - if (!response.ok) { - const body = await response.text().catch(() => ""); - return { - ok: false, - error: { body, kind: "http", status: response.status }, - }; - } + if (!response.ok) { + const body = await response.text().catch(() => ""); + return { + ok: false, + error: { body, kind: "http", status: response.status }, + }; + } + + if (!response.body) { + return { + ok: false, + error: { cause: new Error("response had no body"), kind: "network" }, + }; + } - if (!response.body) { - return { - ok: false, - error: { cause: new Error("response had no body"), kind: "network" }, + let handle: FsWriteHandle; + try { + handle = await fs.openWriteHandle(partPath); + } catch (error: unknown) { + // Nothing else cancels the transfer on an early return — without the + // abort the connection stays open and the body buffers in the background. + controller.abort(toError(error)); + return { ok: false, error: { cause: toError(error), kind: "network" } }; + } + const discardPart = async () => { + await handle.close().catch(() => {}); + await fs.unlink(partPath).catch(() => {}); }; - } - // Read the body before writing it: the deadline covers the download, so a - // stall part-way through the body is a timeout, while a failed write is local. - let downloaded: ArrayBuffer; - try { - downloaded = await response.arrayBuffer(); - } catch (error: unknown) { - if (isTimeoutError(error)) { - return { ok: false, error: { kind: "timeout", timeoutMs } }; + // Reads and writes report through separate catches so a failed disk write + // stays a local error and never masquerades as a network stall. + const reader = response.body.getReader(); + for (;;) { + let result: BodyReadResult; + try { + result = await reader.read(); + } catch (error: unknown) { + await discardPart(); + if (isTimeoutError(error)) { + return { ok: false, error: { kind: "timeout", timeoutMs } }; + } + return { ok: false, error: { cause: toError(error), kind: "network" } }; + } + if (result.done) break; + // The stall clock measures the network, not the disk: pause it while a + // chunk is being written so a slow disk cannot abort a live download. + clearTimeout(stallTimer); + try { + await handle.write(result.value); + } catch (error: unknown) { + controller.abort(toError(error)); + await discardPart(); + return { ok: false, error: { cause: toError(error), kind: "network" } }; + } + armStallTimer(); } - return { ok: false, error: { cause: toError(error), kind: "network" } }; - } - try { - await (deps.fs ?? makeDefaultFs()).writeFile( - args.dest, - new Uint8Array(downloaded), - ); - } catch (error: unknown) { - return { ok: false, error: { cause: toError(error), kind: "network" } }; + try { + await handle.close(); + await fs.rename(partPath, args.dest); + } catch (error: unknown) { + await fs.unlink(partPath).catch(() => {}); + return { ok: false, error: { cause: toError(error), kind: "network" } }; + } + return { ok: true, data: undefined }; + } finally { + clearTimeout(stallTimer); } - - return { ok: true, data: undefined }; } diff --git a/src/shell/platform/slowFetch.testUtils.ts b/src/shell/platform/slowFetch.testUtils.ts index a88e11af0..1b0762ff3 100644 --- a/src/shell/platform/slowFetch.testUtils.ts +++ b/src/shell/platform/slowFetch.testUtils.ts @@ -47,6 +47,65 @@ export function makeTimingOutBodyFetch(): typeof fetch { ) as unknown as typeof fetch; } +/** + * A fetch whose body delivers `chunks` right away and then stalls forever. Like + * a real fetch, aborting the request signal errors the pending body read with + * the abort reason. + */ +export function makeStallingBodyFetch(chunks: string[]): typeof fetch { + return mock(async (_url, init) => { + const signal = init?.signal; + return new Response( + new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(new TextEncoder().encode(chunk)); + } + signal?.addEventListener("abort", () => + controller.error(signal.reason), + ); + }, + }), + ); + }) as unknown as typeof fetch; +} + +/** + * A fetch whose body delivers one of `chunks` every `intervalMs` and then + * closes — a slow download that never stops making progress. Like a real fetch, + * aborting the request signal errors the pending body read with the abort + * reason. + */ +export function makeDrippingBodyFetch( + chunks: string[], + intervalMs: number, +): typeof fetch { + return mock(async (_url, init) => { + const signal = init?.signal; + return new Response( + new ReadableStream({ + start(controller) { + let delivered = 0; + const timer = setInterval(() => { + const chunk = chunks[delivered]; + delivered += 1; + if (chunk === undefined) { + clearInterval(timer); + controller.close(); + return; + } + controller.enqueue(new TextEncoder().encode(chunk)); + }, intervalMs); + signal?.addEventListener("abort", () => { + clearInterval(timer); + controller.error(signal.reason); + }); + }, + }), + ); + }) as unknown as typeof fetch; +} + /** A fetch that answers after `delayMs`, unless its deadline arrives first. */ export function makeDelayedFetch( makeResponse: () => Response, diff --git a/src/shell/platform/teamStorageAssets.reuse.test.ts b/src/shell/platform/teamStorageAssets.reuse.test.ts index 6ad7fc115..eb8600d95 100644 --- a/src/shell/platform/teamStorageAssets.reuse.test.ts +++ b/src/shell/platform/teamStorageAssets.reuse.test.ts @@ -71,16 +71,21 @@ describe("PlatformClient.syncTeamStorageAssets etag reuse", () => { : file, ); const fetch = makeFetch(changedFiles); + const progress: { current: number; total: number }[] = []; const result = await createPlatformClient("qawolf_key", { baseUrl: "https://test.qawolf.com", fetch: fetch.fetch, - }).syncTeamStorageAssets(assetsDir); + }).syncTeamStorageAssets(assetsDir, { + onProgress: (p) => progress.push(p), + }); expect(result).toEqual({ ok: true, value: { downloadedCount: 1, reusedCount: 1, skippedCount: 0 }, }); + // Progress counts only real downloads; the reused file is not in the total. + expect(progress).toEqual([{ current: 1, total: 1 }]); expect(fetch.assetUrls).toEqual(["https://storage.example.com/root-v2"]); expect(await readFile(join(assetsDir, "root.txt"), "utf8")).toBe("root-v2"); expect(await readFile(join(assetsDir, "nested", "data.csv"), "utf8")).toBe( diff --git a/src/shell/platform/teamStorageAssets.test.ts b/src/shell/platform/teamStorageAssets.test.ts index 576a5897e..7404423a6 100644 --- a/src/shell/platform/teamStorageAssets.test.ts +++ b/src/shell/platform/teamStorageAssets.test.ts @@ -43,6 +43,35 @@ describe("PlatformClient.syncTeamStorageAssets", () => { expect(await fs.readFile("/assets/nested/data.csv")).toBe("nested"); }); + it("reports per-file progress while downloading", async () => { + const fakeFetch = makeFetch([ + { + path: "root.txt", + signedUrl: "https://storage.example.com/root", + size: 4, + }, + { + path: "nested/data.csv", + signedUrl: "https://storage.example.com/nested", + size: 6, + }, + ]); + const progress: { current: number; total: number }[] = []; + + await createPlatformClient("qawolf_key", { + baseUrl: "https://test.qawolf.com", + fetch: fakeFetch.fetch, + fs, + }).syncTeamStorageAssets(assetsDir, { + onProgress: (p) => progress.push(p), + }); + + expect(progress).toEqual([ + { current: 1, total: 2 }, + { current: 2, total: 2 }, + ]); + }); + it("writes downloaded assets through the platform fs dependency", async () => { const fakeFetch = makeFetch([ { diff --git a/src/shell/platform/teamStorageAssets.ts b/src/shell/platform/teamStorageAssets.ts index f9a6bb3ff..2388b5092 100644 --- a/src/shell/platform/teamStorageAssets.ts +++ b/src/shell/platform/teamStorageAssets.ts @@ -1,9 +1,5 @@ -import { dirname, join } from "node:path"; - import { errorMessage } from "~/core/errors.js"; import { makeDefaultFs, type Fs } from "~/shell/fs.js"; -import { describeTeamStorageDownloadError } from "./describeErrors.js"; -import { fetchSignedUrl } from "./fetchSignedUrl.js"; import type { PlatformResult } from "./requestWithRetry.js"; import { readAssetManifest, @@ -21,6 +17,10 @@ import { type ReusableAssetFile, } from "./teamStorageAssetReuse.js"; import type { TeamStorageFile } from "./types.js"; +import { + writeAssetSnapshot, + type TeamStorageAssetProgress, +} from "./writeAssetSnapshot.js"; export type SyncTeamStorageAssetsResult = { downloadedCount: number; @@ -36,6 +36,7 @@ type DownloadTeamStorageAssetsArgs = { type DownloadTeamStorageAssetsDeps = { fetch: typeof globalThis.fetch; fs?: Fs | undefined; + onProgress?: ((progress: TeamStorageAssetProgress) => void) | undefined; }; export async function downloadTeamStorageAssets( @@ -87,7 +88,7 @@ async function writeTeamStorageAssets( try { const downloadedCount = await writeAssetSnapshot({ assetsAbs: args.assetsAbs, - deps: { fetch: deps.fetch, fs }, + deps: { fetch: deps.fetch, fs, onProgress: deps.onProgress }, reusable, safeFiles, tmpAssets, @@ -107,40 +108,3 @@ async function writeTeamStorageAssets( throw error; } } - -type WriteAssetSnapshotArgs = { - assetsAbs: string; - deps: { fetch: typeof globalThis.fetch; fs: Fs }; - reusable: ReadonlySet; - safeFiles: readonly ReusableAssetFile[]; - tmpAssets: string; -}; - -async function writeAssetSnapshot( - args: WriteAssetSnapshotArgs, -): Promise { - let downloadedCount = 0; - await args.deps.fs.mkdir(args.tmpAssets, { recursive: true }); - - for (const { file, relativePath } of args.safeFiles) { - const dest = join(args.tmpAssets, relativePath); - await args.deps.fs.mkdir(dirname(dest), { recursive: true }); - if (args.reusable.has(relativePath)) { - await args.deps.fs.copyFile(join(args.assetsAbs, relativePath), dest); - continue; - } - - const result = await fetchSignedUrl( - { url: file.signedUrl, dest }, - { fetch: args.deps.fetch, fs: args.deps.fs }, - ); - if (!result.ok) { - throw new Error( - describeTeamStorageDownloadError(file.path, result.error), - ); - } - downloadedCount++; - } - - return downloadedCount; -} diff --git a/src/shell/platform/writeAssetSnapshot.ts b/src/shell/platform/writeAssetSnapshot.ts new file mode 100644 index 000000000..bf5bc1c07 --- /dev/null +++ b/src/shell/platform/writeAssetSnapshot.ts @@ -0,0 +1,61 @@ +import { dirname, join } from "node:path"; + +import type { Fs } from "~/shell/fs.js"; +import { describeTeamStorageDownloadError } from "./describeErrors.js"; +import { fetchSignedUrl } from "./fetchSignedUrl.js"; +import type { ReusableAssetFile } from "./teamStorageAssetReuse.js"; + +/** + * Per-file download progress: `current` is the file being downloaded, `total` + * counts only files that actually download (reused and skipped files are + * excluded). + */ +export type TeamStorageAssetProgress = { + current: number; + total: number; +}; + +type WriteAssetSnapshotArgs = { + assetsAbs: string; + deps: { + fetch: typeof globalThis.fetch; + fs: Fs; + onProgress?: ((progress: TeamStorageAssetProgress) => void) | undefined; + }; + reusable: ReadonlySet; + safeFiles: readonly ReusableAssetFile[]; + tmpAssets: string; +}; + +export async function writeAssetSnapshot( + args: WriteAssetSnapshotArgs, +): Promise { + let downloadedCount = 0; + const total = args.safeFiles.filter( + ({ relativePath }) => !args.reusable.has(relativePath), + ).length; + await args.deps.fs.mkdir(args.tmpAssets, { recursive: true }); + + for (const { file, relativePath } of args.safeFiles) { + const dest = join(args.tmpAssets, relativePath); + await args.deps.fs.mkdir(dirname(dest), { recursive: true }); + if (args.reusable.has(relativePath)) { + await args.deps.fs.copyFile(join(args.assetsAbs, relativePath), dest); + continue; + } + + args.deps.onProgress?.({ current: downloadedCount + 1, total }); + const result = await fetchSignedUrl( + { url: file.signedUrl, dest }, + { fetch: args.deps.fetch, fs: args.deps.fs }, + ); + if (!result.ok) { + throw new Error( + describeTeamStorageDownloadError(file.path, result.error), + ); + } + downloadedCount++; + } + + return downloadedCount; +} diff --git a/src/shell/ui/renderers/modes/agent.test.ts b/src/shell/ui/renderers/modes/agent.test.ts index ce31e7721..112fc57b8 100644 --- a/src/shell/ui/renderers/modes/agent.test.ts +++ b/src/shell/ui/renderers/modes/agent.test.ts @@ -151,5 +151,24 @@ describe("agent renderers", () => { expect(spy).toHaveBeenCalledWith("All done!\n"); expect(clack.spinner).not.toHaveBeenCalled(); }); + + it("writes a stderr line when a task reports progress", async () => { + const spy = stderrSpy(); + const { withProgress } = createAgentRenderers(); + + await withProgress( + [ + { + message: "Downloading assets", + task: async (update) => { + update("Downloading assets (1/2)"); + }, + }, + ], + "done", + ); + + expect(spy).toHaveBeenCalledWith("[1/1] Downloading assets (1/2)\n"); + }); }); }); diff --git a/src/shell/ui/renderers/modes/agent.ts b/src/shell/ui/renderers/modes/agent.ts index 4779d7e88..c3154fb14 100644 --- a/src/shell/ui/renderers/modes/agent.ts +++ b/src/shell/ui/renderers/modes/agent.ts @@ -37,8 +37,12 @@ export function createAgentRenderers(): RendererSet { const results: unknown[] = []; const total = steps.length; for (const [i, step] of steps.entries()) { - writeStderrLine(`[${String(i + 1)}/${String(total)}] ${step.message}`); - results.push(await step.task()); + const label = (message: string) => + `[${String(i + 1)}/${String(total)}] ${message}`; + writeStderrLine(label(step.message)); + results.push( + await step.task((message) => writeStderrLine(label(message))), + ); } const { typed, doneMessage } = finalizeResults(results, done); diff --git a/src/shell/ui/renderers/modes/human.ts b/src/shell/ui/renderers/modes/human.ts index bf2c22a63..fdbb8be94 100644 --- a/src/shell/ui/renderers/modes/human.ts +++ b/src/shell/ui/renderers/modes/human.ts @@ -39,9 +39,16 @@ export function createHumanRenderers( let currentLabel = ""; try { for (const [i, step] of steps.entries()) { - currentLabel = `[${String(i + 1)}/${String(total)}] ${step.message}`; + const label = (message: string) => + `[${String(i + 1)}/${String(total)}] ${message}`; + currentLabel = label(step.message); clack.log.step(currentLabel); - results.push(await step.task()); + results.push( + await step.task((message) => { + currentLabel = label(message); + clack.log.step(currentLabel); + }), + ); } const { typed, doneMessage } = finalizeResults(results, done); clack.log.success(doneMessage); @@ -52,18 +59,26 @@ export function createHumanRenderers( } } - // existing spinner path — unchanged const s = clack.spinner(); let currentLabel = ""; try { for (const [i, step] of steps.entries()) { - currentLabel = `[${String(i + 1)}/${String(total)}] ${step.message}`; + const label = (message: string) => + `[${String(i + 1)}/${String(total)}] ${message}`; + currentLabel = label(step.message); if (i === 0) { s.start(currentLabel); } else { s.message(currentLabel); } - results.push(await step.task()); + // Track the latest progress label so a mid-task failure reports + // where the work stopped, not just which step it was in. + results.push( + await step.task((message) => { + currentLabel = label(message); + s.message(currentLabel); + }), + ); } const { typed, doneMessage } = finalizeResults(results, done); s.stop(doneMessage); diff --git a/src/shell/ui/renderers/modes/human.withProgress.update.test.ts b/src/shell/ui/renderers/modes/human.withProgress.update.test.ts new file mode 100644 index 000000000..efe9de7f0 --- /dev/null +++ b/src/shell/ui/renderers/modes/human.withProgress.update.test.ts @@ -0,0 +1,90 @@ +import { afterEach, describe, expect, it, mock } from "bun:test"; + +import { makeClack } from "~/shell/ui/clack/styledClack.mock.js"; +import { createHumanRenderers } from "./human.js"; + +describe("human renderers — withProgress task progress updates", () => { + afterEach(() => { + mock.restore(); + }); + + it("updates the spinner label when a task reports progress", async () => { + const clack = makeClack(); + const { withProgress } = createHumanRenderers(clack); + + await withProgress( + [ + { + message: "Downloading assets", + task: async (update) => { + update("Downloading assets (1/3)"); + update("Downloading assets (2/3)"); + }, + }, + ], + "done", + ); + + const s = clack.createdSpinners[0]!; + expect(s.message).toHaveBeenNthCalledWith( + 1, + "[1/1] Downloading assets (1/3)", + ); + expect(s.message).toHaveBeenNthCalledWith( + 2, + "[1/1] Downloading assets (2/3)", + ); + }); + + it("reports the latest progress label when a task fails", async () => { + const clack = makeClack(); + const { withProgress } = createHumanRenderers(clack); + + let caughtError: unknown; + try { + await withProgress( + [ + { + message: "Downloading assets", + task: async (update) => { + update("Downloading assets (2/3)"); + throw new Error("download failed"); + }, + }, + ], + "done", + ); + } catch (e) { + caughtError = e; + } + + expect(caughtError).toBeInstanceOf(Error); + const s = clack.createdSpinners[0]!; + expect(s.error).toHaveBeenCalledWith("[1/1] Downloading assets (2/3)"); + }); + + it("logs task progress updates as steps when verboseTarget is provided", async () => { + const clack = makeClack(); + const verboseTarget: { write: ((msg: string) => void) | undefined } = { + write: undefined, + }; + const { withProgress } = createHumanRenderers(clack, verboseTarget); + + await withProgress( + [ + { + message: "Downloading assets", + task: async (update) => { + update("Downloading assets (1/2)"); + }, + }, + ], + "Done", + ); + + expect(clack.log.step).toHaveBeenNthCalledWith( + 2, + "[1/1] Downloading assets (1/2)", + ); + }); +}); diff --git a/src/shell/ui/renderers/modes/json.test.ts b/src/shell/ui/renderers/modes/json.test.ts index 5d22e65f6..46d0ba2bf 100644 --- a/src/shell/ui/renderers/modes/json.test.ts +++ b/src/shell/ui/renderers/modes/json.test.ts @@ -200,5 +200,28 @@ describe("json renderers", () => { ); expect(clack.spinner).not.toHaveBeenCalled(); }); + + // Step events already carry structure; per-file updates would only add + // noise to NDJSON consumers. The callback must still be callable so tasks + // can report progress without checking the mode. + it("passes tasks a callable progress update that emits nothing", async () => { + const spy = stderrSpy(); + const { withProgress } = createJsonRenderers(); + + await withProgress( + [ + { + message: "step", + task: async (update) => { + update("progress detail"); + }, + }, + ], + "done", + ); + + const messages = spy.mock.calls.map((call) => String(call[0])); + expect(messages.some((m) => m.includes("progress detail"))).toBe(false); + }); }); }); diff --git a/src/shell/ui/renderers/modes/json.ts b/src/shell/ui/renderers/modes/json.ts index b0aa48ff1..ac0c236a8 100644 --- a/src/shell/ui/renderers/modes/json.ts +++ b/src/shell/ui/renderers/modes/json.ts @@ -37,7 +37,9 @@ export function createJsonRenderers(): RendererSet { step: i + 1, total, }); - results.push(await step.task()); + // Step events already carry structure; per-file progress updates would + // only add noise for NDJSON consumers. + results.push(await step.task(() => {})); } const { typed, doneMessage } = finalizeResults(results, done); diff --git a/src/shell/ui/renderers/modes/progress.ts b/src/shell/ui/renderers/modes/progress.ts index 67751b16a..cc4916cc2 100644 --- a/src/shell/ui/renderers/modes/progress.ts +++ b/src/shell/ui/renderers/modes/progress.ts @@ -1,6 +1,11 @@ export type ProgressStep = { message: string; - task: () => Promise; + /** + * `update` replaces the step's displayed message while the task runs, e.g. + * with a per-file download counter. Tasks without in-flight progress can + * ignore it. + */ + task: (update: (message: string) => void) => Promise; }; export type InferStepResults = {