diff --git a/bun.lock b/bun.lock index cac1f849d6..6eed803372 100644 --- a/bun.lock +++ b/bun.lock @@ -152,6 +152,7 @@ "@google-cloud/workflows": "^4.2.0", "@hono/node-server": "^1.13.0", "@hyperframes/producer": "workspace:^", + "google-auth-library": "^10.5.0", "hono": "^4.6.0", "puppeteer-core": "^24.39.1", "tar": "^7.4.3", diff --git a/docs/packages/gcp-cloud-run.mdx b/docs/packages/gcp-cloud-run.mdx index bce0538480..1f3e86efd2 100644 --- a/docs/packages/gcp-cloud-run.mdx +++ b/docs/packages/gcp-cloud-run.mdx @@ -98,6 +98,8 @@ console.log(progress.status, progress.outputFile, progress.costs.displayCost); Pass `projectDir` for one-shot uploads, or call `deploySite()` separately and reuse the returned site handle across many renders. +While the execution is running, `getRenderProgress` reads the Cloud Workflows **step-entries API** to report live chunk-level progress — `overallProgress` follows the same 10% plan + 80% chunks + 10% assemble split as the AWS adapter, and `chunksCompleted` / `totalChunks` expose the raw counts. The read is best-effort: if the caller lacks `workflowexecutions.stepEntries.list` (included in `roles/workflows.viewer`) the snapshot degrades to coarse `overallProgress = 0` instead of failing. Pass `midFlightProgress: false` to skip the extra API call per poll. Exact frame totals and costs still come from the accumulated result on success. + ## Related Guides diff --git a/packages/cli/src/commands/cloudrun.ts b/packages/cli/src/commands/cloudrun.ts index 3268062255..862670ec8e 100644 --- a/packages/cli/src/commands/cloudrun.ts +++ b/packages/cli/src/commands/cloudrun.ts @@ -538,7 +538,15 @@ async function runRender(args: Record): Promise { while (progress.status === "running") { await new Promise((r) => setTimeout(r, intervalMs)); progress = await getRenderProgress({ executionName: handle.executionName }); - if (!args.json) process.stdout.write(`\r status=${progress.status} `); + if (!args.json) { + const chunks = + progress.totalChunks != null + ? ` chunks=${progress.chunksCompleted}/${progress.totalChunks}` + : ""; + process.stdout.write( + `\r status=${progress.status} progress=${(progress.overallProgress * 100).toFixed(0)}%${chunks} `, + ); + } } if (!args.json) process.stdout.write("\n"); if (args.json) { @@ -570,6 +578,8 @@ async function runProgress(args: Record): Promise { return; } console.log(`status=${progress.status} progress=${(progress.overallProgress * 100).toFixed(0)}%`); + if (progress.totalChunks != null) + console.log(`chunks=${progress.chunksCompleted}/${progress.totalChunks}`); if (progress.totalFrames) console.log(`frames=${progress.framesRendered}/${progress.totalFrames}`); if (progress.outputFile) console.log(`output=${progress.outputFile.gcsUri}`); diff --git a/packages/gcp-cloud-run/package.json b/packages/gcp-cloud-run/package.json index 5896612431..bab8812100 100644 --- a/packages/gcp-cloud-run/package.json +++ b/packages/gcp-cloud-run/package.json @@ -45,6 +45,7 @@ "@google-cloud/workflows": "^4.2.0", "@hono/node-server": "^1.13.0", "@hyperframes/producer": "workspace:^", + "google-auth-library": "^10.5.0", "hono": "^4.6.0", "puppeteer-core": "^24.39.1", "tar": "^7.4.3" diff --git a/packages/gcp-cloud-run/src/sdk/getRenderProgress.test.ts b/packages/gcp-cloud-run/src/sdk/getRenderProgress.test.ts index bcc992e606..12bf573882 100644 --- a/packages/gcp-cloud-run/src/sdk/getRenderProgress.test.ts +++ b/packages/gcp-cloud-run/src/sdk/getRenderProgress.test.ts @@ -8,6 +8,8 @@ import { type ExecutionRecord, type ExecutionsGetClientLike, getRenderProgress, + type StepEntriesListerLike, + type StepEntryRecord, } from "./getRenderProgress.js"; function fakeExecutions(record: ExecutionRecord): ExecutionsGetClientLike { @@ -18,6 +20,18 @@ function fakeExecutions(record: ExecutionRecord): ExecutionsGetClientLike { }; } +function fakeStepEntries(entries: StepEntryRecord[]): StepEntriesListerLike { + return { + async listStepEntries(_executionName: string) { + return entries; + }, + }; +} + +function succeeded(step: string, count = 1): StepEntryRecord[] { + return Array.from({ length: count }, () => ({ step, state: "STATE_SUCCEEDED" })); +} + const accumulated = JSON.stringify({ Plan: { TotalFrames: 90, DurationMs: 4000 }, Chunks: [ @@ -29,17 +43,97 @@ const accumulated = JSON.stringify({ }); describe("getRenderProgress", () => { - it("reports running with no frame data while ACTIVE", async () => { + it("reports running with no frame data while ACTIVE and no step entries yet", async () => { const p = await getRenderProgress({ executionName: "x", executions: fakeExecutions({ state: "ACTIVE", startTime: { seconds: 1700000000 } }), + stepEntries: fakeStepEntries([]), }); expect(p.status).toBe("running"); expect(p.overallProgress).toBe(0); expect(p.totalFrames).toBeNull(); + expect(p.chunksCompleted).toBe(0); + expect(p.totalChunks).toBeNull(); expect(p.fatalErrorEncountered).toBe(false); }); + it("reports chunk-based mid-flight progress from step entries", async () => { + const p = await getRenderProgress({ + executionName: "x", + executions: fakeExecutions({ state: "ACTIVE", startTime: { seconds: 1700000000 } }), + stepEntries: fakeStepEntries([ + ...succeeded("plan"), + ...succeeded("appendSlots", 4), + ...succeeded("renderOneChunk", 2), + { step: "renderOneChunk", state: "STATE_IN_PROGRESS" }, + ]), + }); + expect(p.status).toBe("running"); + // 10% plan + 80% * (2/4 chunks) + expect(p.overallProgress).toBeCloseTo(0.5, 10); + expect(p.chunksCompleted).toBe(2); + expect(p.totalChunks).toBe(4); + expect(p.invocationsObserved).toBe(3); // plan + 2 chunks + expect(p.framesRendered).toBe(0); // step entries carry no frame counts + }); + + it("caps mid-flight progress below 1 once assemble succeeds", async () => { + const p = await getRenderProgress({ + executionName: "x", + executions: fakeExecutions({ state: "ACTIVE" }), + stepEntries: fakeStepEntries([ + ...succeeded("plan"), + ...succeeded("appendSlots", 2), + ...succeeded("renderOneChunk", 2), + ...succeeded("assemble"), + ]), + }); + expect(p.overallProgress).toBe(0.99); + expect(p.chunksCompleted).toBe(2); + }); + + it("reports plan-only progress once plan succeeds but before the chunk list exists", async () => { + const p = await getRenderProgress({ + executionName: "x", + executions: fakeExecutions({ state: "ACTIVE" }), + stepEntries: fakeStepEntries(succeeded("plan")), + }); + expect(p.overallProgress).toBeCloseTo(0.1, 10); + expect(p.totalChunks).toBeNull(); + }); + + it("degrades to the coarse snapshot when the step-entries read fails", async () => { + const p = await getRenderProgress({ + executionName: "x", + executions: fakeExecutions({ state: "ACTIVE" }), + stepEntries: { + async listStepEntries() { + throw new Error("PERMISSION_DENIED"); + }, + }, + }); + expect(p.status).toBe("running"); + expect(p.overallProgress).toBe(0); + expect(p.chunksCompleted).toBe(0); + }); + + it("skips the step-entries read when midFlightProgress is false", async () => { + let called = false; + const p = await getRenderProgress({ + executionName: "x", + executions: fakeExecutions({ state: "ACTIVE" }), + midFlightProgress: false, + stepEntries: { + async listStepEntries() { + called = true; + return []; + }, + }, + }); + expect(called).toBe(false); + expect(p.overallProgress).toBe(0); + }); + it("reports succeeded with parsed frames + cost", async () => { const p = await getRenderProgress({ executionName: "x", @@ -57,6 +151,8 @@ describe("getRenderProgress", () => { expect(p.totalFrames).toBe(90); expect(p.framesRendered).toBe(90); expect(p.invocationsObserved).toBe(5); // plan + 3 chunks + assemble + expect(p.chunksCompleted).toBe(3); + expect(p.totalChunks).toBe(3); expect(p.outputFile).toEqual({ gcsUri: "gs://b/renders/r1/output.mp4", bytes: 123456 }); expect(p.costs.accruedSoFarUsd).toBeGreaterThan(0); expect(p.costs.breakdown.estimated).toBe(false); diff --git a/packages/gcp-cloud-run/src/sdk/getRenderProgress.ts b/packages/gcp-cloud-run/src/sdk/getRenderProgress.ts index 2ecada506c..24f850934f 100644 --- a/packages/gcp-cloud-run/src/sdk/getRenderProgress.ts +++ b/packages/gcp-cloud-run/src/sdk/getRenderProgress.ts @@ -11,11 +11,16 @@ * file, and per-step `DurationMs` (which the handler stamps into every * result), then compute cost against the service's configured vCPU/memory. * - * Progress is therefore coarse while the execution is ACTIVE (we report - * `running` with `overallProgress = 0`) and exact once it SUCCEEDS - * (`overallProgress = 1`, real frame + cost numbers). Mid-flight per-chunk - * progress would require the Workflows step-entries API; that's a tracked - * follow-up, not part of the first version. + * While the execution is ACTIVE, mid-flight progress comes from the + * Workflows **step-entries API** (`executions.stepEntries.list`, REST — the + * Node gapic client doesn't expose it yet): we count succeeded + * `renderOneChunk` entries against the chunk-slot count and map them onto + * the same 10 % Plan + 80 % chunks + 10 % Assemble split the AWS adapter + * uses. Frame counts stay unknown mid-flight (step entries carry no + * payloads), so `framesRendered` is 0 until success — `chunksCompleted` / + * `totalChunks` are the live signals. The step-entries read is best-effort: + * any API/permission failure degrades to the coarse `overallProgress = 0` + * snapshot instead of throwing. */ import { @@ -40,11 +45,16 @@ export interface RenderError { /** Snapshot of a single render's progress + cost + errors at one point in time. */ export interface RenderProgress { status: RenderStatus; - /** `[0, 1]`; coarse while running, exact on success. */ + /** `[0, 1]`; chunk-based while running (step entries), exact on success. */ overallProgress: number; + /** Exact on success; 0 while running (step entries carry no frame counts). */ framesRendered: number; /** `null` until the execution succeeds and the accumulated plan result is read. */ totalFrames: number | null; + /** Chunks whose render step has succeeded so far (live while running). */ + chunksCompleted: number; + /** Planned chunk count once the chunk list is built; `null` before that. */ + totalChunks: number | null; /** Cloud Run invocations the workflow scheduled (Plan + chunks + Assemble), when known. */ invocationsObserved: number; costs: RenderCost; @@ -78,6 +88,19 @@ export interface ExecutionsGetClientLike { getExecution(req: { name: string }): Promise<[ExecutionRecord, ...unknown[]]>; } +/** One step entry from `executions.stepEntries.list` (REST). */ +export interface StepEntryRecord { + /** Step name from workflow.yaml (e.g. `renderOneChunk`). */ + step?: string | null; + /** `STATE_SUCCEEDED` / `STATE_IN_PROGRESS` / `STATE_FAILED`. */ + state?: string | null; +} + +/** Injection seam for the step-entries reader (REST; not in the gapic client). */ +export interface StepEntriesListerLike { + listStepEntries(executionName: string): Promise; +} + /** Options for {@link getRenderProgress}. */ export interface GetRenderProgressOptions { /** Server-assigned execution resource name from a {@link renderToCloudRun} call. */ @@ -88,6 +111,13 @@ export interface GetRenderProgressOptions { memoryGib?: number; /** Test injection seam — production callers leave unset. */ executions?: ExecutionsGetClientLike; + /** Test injection seam for the step-entries reader — production callers leave unset. */ + stepEntries?: StepEntriesListerLike; + /** + * Set to false to skip the step-entries API while the execution is ACTIVE + * (one extra authenticated REST call per poll). Default true. + */ + midFlightProgress?: boolean; } const DEFAULT_VCPU = 4; @@ -129,15 +159,22 @@ export async function getRenderProgress(opts: GetRenderProgressOptions): Promise }); } - // Default snapshot: running / unknown — no frame or cost data until the - // accumulated result is available on success. + // Non-success snapshot: frame + cost data only exist in the accumulated + // result on success, but a live execution still gets chunk-level progress + // from the step-entries API. if (status !== "succeeded") { + const midFlight = + status === "running" && opts.midFlightProgress !== false + ? await tryMidFlightSnapshot(opts) + : null; return { status, - overallProgress: 0, + overallProgress: midFlight?.overallProgress ?? 0, framesRendered: 0, totalFrames: null, - invocationsObserved: 0, + chunksCompleted: midFlight?.chunksCompleted ?? 0, + totalChunks: midFlight?.totalChunks ?? null, + invocationsObserved: midFlight?.invocationsObserved ?? 0, costs: computeRenderCost([], 0), outputFile: null, errors, @@ -183,6 +220,8 @@ export async function getRenderProgress(opts: GetRenderProgressOptions): Promise overallProgress: 1, framesRendered, totalFrames, + chunksCompleted: chunks.length, + totalChunks: chunks.length, invocationsObserved: invocations.length, costs, outputFile, @@ -193,6 +232,126 @@ export async function getRenderProgress(opts: GetRenderProgressOptions): Promise }; } +// ── Mid-flight progress via the step-entries API ───────────────────────────── + +// Step names from terraform/workflow.yaml. `appendSlots` runs once per chunk +// in the (fast, sequential) fillLists loop, so its succeeded-entry count IS +// the planned chunk count — available well before any chunk finishes. +const PLAN_STEP = "plan"; +const CHUNK_SLOT_STEP = "appendSlots"; +const CHUNK_STEP = "renderOneChunk"; +const ASSEMBLE_STEP = "assemble"; + +interface MidFlightSnapshot { + overallProgress: number; + chunksCompleted: number; + totalChunks: number | null; + invocationsObserved: number; +} + +function entrySucceeded(entry: StepEntryRecord): boolean { + return entry.state === "STATE_SUCCEEDED" || entry.state === "SUCCEEDED"; +} + +// fallow-ignore-next-line complexity +function summarizeStepEntries(entries: readonly StepEntryRecord[]): MidFlightSnapshot { + let planDone = false; + let slots = 0; + let chunksCompleted = 0; + let assembleDone = false; + let invocations = 0; + for (const entry of entries) { + if (!entrySucceeded(entry)) continue; + switch (entry.step) { + case PLAN_STEP: + planDone = true; + invocations += 1; + break; + case CHUNK_SLOT_STEP: + slots += 1; + break; + case CHUNK_STEP: + chunksCompleted += 1; + invocations += 1; + break; + case ASSEMBLE_STEP: + assembleDone = true; + invocations += 1; + break; + default: + break; + } + } + const totalChunks = slots > 0 ? slots : null; + return { + overallProgress: midFlightProgress(planDone, chunksCompleted, totalChunks, assembleDone), + chunksCompleted, + totalChunks, + invocationsObserved: invocations, + }; +} + +// Same 10 % Plan + 80 % chunks + 10 % Assemble split as the AWS adapter, +// measured in chunks instead of frames. Never returns 1 — the execution +// itself reports success. +// fallow-ignore-next-line complexity +function midFlightProgress( + planDone: boolean, + chunksCompleted: number, + totalChunks: number | null, + assembleDone: boolean, +): number { + if (assembleDone) return 0.99; + if (!planDone) return 0; + if (totalChunks == null || totalChunks <= 0) return 0.1; + return 0.1 + 0.8 * Math.min(1, chunksCompleted / totalChunks); +} + +async function tryMidFlightSnapshot( + opts: GetRenderProgressOptions, +): Promise { + try { + const lister = opts.stepEntries ?? (await defaultStepEntriesLister()); + return summarizeStepEntries(await lister.listStepEntries(opts.executionName)); + } catch { + // Missing permission (workflowexecutions.stepEntries.list), API not + // enabled, or transient failure — degrade to the coarse snapshot. + return null; + } +} + +const STEP_ENTRIES_PAGE_SIZE = 500; +const STEP_ENTRIES_MAX_PAGES = 20; + +// The gapic ExecutionsClient (v4) has no stepEntries surface, so this hits +// the REST endpoint directly with ADC via google-auth-library. +async function defaultStepEntriesLister(): Promise { + const { GoogleAuth } = await import("google-auth-library"); + const auth = new GoogleAuth({ + scopes: "https://www.googleapis.com/auth/cloud-platform", + }); + const client = await auth.getClient(); + return { + // fallow-ignore-next-line complexity + async listStepEntries(executionName: string): Promise { + const entries: StepEntryRecord[] = []; + let pageToken: string | undefined; + for (let page = 0; page < STEP_ENTRIES_MAX_PAGES; page += 1) { + const token = pageToken ? `&pageToken=${encodeURIComponent(pageToken)}` : ""; + const url = `https://workflowexecutions.googleapis.com/v1/${executionName}/stepEntries?pageSize=${STEP_ENTRIES_PAGE_SIZE}${token}`; + const res = await client.request<{ + stepEntries?: StepEntryRecord[]; + nextPageToken?: string; + }>({ url }); + entries.push(...(res.data.stepEntries ?? [])); + pageToken = res.data.nextPageToken; + if (!pageToken) break; + } + return entries; + }, + }; +} + // fallow-ignore-next-line complexity function mapState(state: string | null | undefined): RenderStatus { switch (state) {