mirror of
https://github.com/heygen-com/hyperframes.git
synced 2026-09-11 14:50:02 +00:00
feat(gcp-cloud-run): Google Cloud Run + Workflows distributed render adapter (#1253)
* feat(gcp-cloud-run): add Google Cloud Run + Workflows distributed render adapter Adds @hyperframes/gcp-cloud-run, the GCP counterpart to @hyperframes/aws-lambda (issue #932). The OSS distributed primitives (plan, renderChunk x N, assemble) are unchanged; this package is the storage/compute/orchestration glue. Package: Cloud Run handler (one image, three actions), runs under bun; GCS transport; in-image chrome-headless-shell resolver; client SDK (renderToCloudRun, getRenderProgress, deploySite, computeRenderCost); Dockerfile; Cloud Workflows definition; Terraform module; CLI cloudrun deploy|sites|render|render-batch|progress|destroy with --output-resolution and --strict-variables; 62 unit tests + docs + live smoke script. Shared extraction (removes ~640 lines of adapter duplication): move the cloud-agnostic config validator + content-hash into producer/distributed; both adapters import them. Validated end-to-end on GCP at 37.4 dB PSNR vs baseline. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(cli): resolve @hyperframes/gcp-cloud-run in the CLI build + root build The CLI bundle (esbuild) couldn't resolve `@hyperframes/gcp-cloud-run/sdk`, failing Build/Typecheck/CLI-smoke (and the perf/windows/regression jobs that build first). Mirror the aws-lambda handling: mark the gcp adapter + its /sdk subpath external in tsup.config.ts with a source alias, and add gcp-cloud-run to the root `build` filter so its dist exists for publish + runtime. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(ci): copy gcp-cloud-run manifest in Dockerfile.test for frozen install The regression test image runs `bun install --frozen-lockfile` after copying each workspace package.json individually. The CLI now depends on @hyperframes/gcp-cloud-run (workspace:*), so the frozen install fails to resolve it unless its manifest is present. Add the COPY line. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(cli): add machine-sizing flags to `cloudrun deploy` Closes the parity gap with `lambda deploy` (which exposes --memory etc.). `cloudrun deploy` now threads --cpu, --memory, --max-instances, and --timeout into the Terraform apply; omitted flags keep the module defaults (4 vCPU / 16Gi / 100 instances / 3600s). For finer control, apply the module directly. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(gcp-cloud-run): address PR review (security, waste, limits, alerts) - server.ts: bucket-allowlist guard no longer fails open silently. Unset env logs a one-time WARNING; "*" is an explicit opt-out; otherwise it enforces. - server.ts: stop double-shipping audio.aac. It already rides in the plan tarball every consumer downloads, so drop the redundant standalone upload (plan) + re-download/overwrite (assemble); assemble reads it from the untar, falling back to a supplied AudioGcsUri for compat. - server.ts: chunk extension via path.extname() instead of slice(lastIndexOf). - workflow.yaml: clamp parallel concurrency_limit to math.min(chunkCount, 20) — Cloud Workflows hard-caps concurrent iterations at 20. - Dockerfile: pin bun (bun-v1.3.9) so an interop change can't silently break the image rebuild. - terraform: add min_instances var (default 0); add a workflow-failure alert (finished_execution_count status=FAILED) alongside the request-count one. - costAccounting: document that displayCost excludes GCS storage/egress. Verified against the actual APIs: @google-cloud/workflows@4.4.0 ICreateExecutionRequest has no executionId (so the idempotency-token suggestion isn't available in this client); Workflows concurrency cap is 20; failure metric is workflows.googleapis.com/finished_execution_count (status label). 174 adapter tests pass, fallow/oxlint/oxfmt/terraform clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(gcp-cloud-run): address round-2 review — error code + CFR forwarding - workflow.yaml: rename the zero-chunk failure code PLAN_TOO_LARGE → PLAN_PRODUCED_ZERO_CHUNKS. The old code implied a size-ceiling breach (the opposite cause), misleading anyone triaging the alert. - workflow.yaml: forward Config.cfr to the assemble step (`Cfr: ${("cfr" in config) and config.cfr}`). It was read by the handler but never sent, so exact-CFR was silently off for every Cloud Run render. Uses the same `in`-operator guard already proven in the retryable predicate. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(release): include gcp-cloud-run in set-version PACKAGES list set-version.ts (driven by release:prepare) bumps an explicit package list to the shared version on each release. gcp-cloud-run was wired into the build + publish.yml but missing here, so a release would leave it at a stale version and publish.yml would push the wrong version. Add it so the new package version-bumps + publishes in lockstep with the others. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
James Russo
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
806b226b34
commit
4da567df22
@@ -0,0 +1,35 @@
|
||||
/**
|
||||
* `computeRenderCost` unit tests — Cloud Run vCPU/GiB-second + Workflows
|
||||
* step math.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import { type BilledCloudRunInvocation, computeRenderCost } from "./costAccounting.js";
|
||||
|
||||
describe("computeRenderCost", () => {
|
||||
it("returns zero for no invocations", () => {
|
||||
const cost = computeRenderCost([], 0);
|
||||
expect(cost.accruedSoFarUsd).toBe(0);
|
||||
expect(cost.displayCost).toBe("$0.0000");
|
||||
});
|
||||
|
||||
it("sums vCPU + memory seconds plus per-request and step charges", () => {
|
||||
const invs: BilledCloudRunInvocation[] = [
|
||||
{ durationMs: 10_000, vcpu: 4, memoryGib: 16, estimated: false },
|
||||
{ durationMs: 10_000, vcpu: 4, memoryGib: 16, estimated: false },
|
||||
];
|
||||
const cost = computeRenderCost(invs, 6);
|
||||
// 2 × 10s: vCPU 80 vcpu-s × 0.000024 = 0.00192; mem 320 GiB-s × 0.0000025 = 0.0008;
|
||||
// requests 2 × 4e-7 ≈ 0 → raw 0.0027208, rounded to 4 dp = 0.0027.
|
||||
// workflows 6 × 1e-5 = 0.00006 → rounds up to 0.0001 at 4 dp.
|
||||
expect(cost.breakdown.cloudRunUsd).toBeCloseTo(0.0027, 4);
|
||||
expect(cost.breakdown.workflowsUsd).toBeCloseTo(0.0001, 4);
|
||||
expect(cost.accruedSoFarUsd).toBeGreaterThan(0);
|
||||
expect(cost.breakdown.gcsEstimate).toBe("not-included");
|
||||
});
|
||||
|
||||
it("flags estimated when any invocation was estimated", () => {
|
||||
const cost = computeRenderCost([{ durationMs: 0, vcpu: 4, memoryGib: 16, estimated: true }], 4);
|
||||
expect(cost.breakdown.estimated).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,113 @@
|
||||
/**
|
||||
* Per-render cost accounting for {@link getRenderProgress}.
|
||||
*
|
||||
* Google bills the render service two ways:
|
||||
*
|
||||
* - **Cloud Run** by **vCPU-seconds** and **GiB-seconds** of request
|
||||
* processing time, plus a flat per-request charge. Each handler
|
||||
* invocation returns its own `DurationMs` in the result body, so the
|
||||
* progress reader can recover billed time per step without a separate
|
||||
* Cloud Monitoring query — multiply by the service's configured vCPU /
|
||||
* memory to get the resource-seconds.
|
||||
* - **Cloud Workflows** by **steps executed**. The orchestration is a
|
||||
* fixed shape (Plan + N×RenderChunk + Assemble + a handful of control
|
||||
* steps), so the step count scales with chunk count.
|
||||
*
|
||||
* The math is documented inline so the constants stay close to the pricing
|
||||
* source they came from. Cost is **best-effort**: GCP pricing varies by
|
||||
* region + committed-use discounts; we use on-demand `us-central1` (Tier 1)
|
||||
* rates as of 2026-06 and label the result `displayCost` so callers see the
|
||||
* dollar value but downstream automation can also read the raw number.
|
||||
*/
|
||||
|
||||
/** Cloud Run request-based billing, us-central1 Tier 1: USD per vCPU-second. */
|
||||
const CLOUD_RUN_USD_PER_VCPU_SECOND = 0.000024;
|
||||
/** Cloud Run request-based billing, us-central1 Tier 1: USD per GiB-second. */
|
||||
const CLOUD_RUN_USD_PER_GIB_SECOND = 0.0000025;
|
||||
/** Cloud Run: USD per request ($0.40 per million). */
|
||||
const CLOUD_RUN_USD_PER_REQUEST = 0.0000004;
|
||||
/** Cloud Workflows: USD per internal step ($0.01 per 1,000, after a free tier). */
|
||||
const WORKFLOWS_USD_PER_STEP = 0.00001;
|
||||
|
||||
/** Per-invocation billed slice the cost calc cares about. */
|
||||
export interface BilledCloudRunInvocation {
|
||||
/** Wall-clock the handler reported via `DurationMs` in its result body. */
|
||||
durationMs: number;
|
||||
/** vCPU the Cloud Run service was configured with at invocation time. */
|
||||
vcpu: number;
|
||||
/** Memory in GiB the Cloud Run service was configured with. */
|
||||
memoryGib: number;
|
||||
/** `true` if the duration was inferred (step result missing) rather than read from the handler payload. */
|
||||
estimated: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Result of {@link computeRenderCost}.
|
||||
*
|
||||
* NOTE: `displayCost` / `accruedSoFarUsd` cover Cloud Run compute + Cloud
|
||||
* Workflows steps only. They EXCLUDE GCS storage + network egress for the
|
||||
* plan tarball (which can be ~100 MB), chunk artifacts, and the final output
|
||||
* — see `breakdown.gcsEstimate`. Treat the figure as a compute-cost floor,
|
||||
* not the authoritative total bill.
|
||||
*/
|
||||
export interface RenderCost {
|
||||
/** USD accrued to date (Cloud Run + Workflows only; excludes GCS — see note above). */
|
||||
accruedSoFarUsd: number;
|
||||
/** Human-readable USD string, e.g. `"$0.0214"`. Excludes GCS storage/egress. */
|
||||
displayCost: string;
|
||||
breakdown: {
|
||||
cloudRunUsd: number;
|
||||
workflowsUsd: number;
|
||||
/** GCS transfer + storage cost varies by tier; we don't try to compute it here. */
|
||||
gcsEstimate: "not-included";
|
||||
/** `true` if any invocation fell back to estimated billing. */
|
||||
estimated: boolean;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Sum Cloud Run vCPU-seconds + GiB-seconds + per-request charges and Cloud
|
||||
* Workflows steps into an aggregate USD figure.
|
||||
*
|
||||
* `workflowSteps` is the count of Workflows steps executed so far — Plan
|
||||
* (1) + RenderChunk (chunkCount) + Assemble (1) + the control steps
|
||||
* (BuildChunkList, AssertChunkCount, …). Pass the count the progress reader
|
||||
* derived from the execution; a rough constant overhead is fine since the
|
||||
* step charge is a rounding error next to Cloud Run compute.
|
||||
*/
|
||||
export function computeRenderCost(
|
||||
invocations: BilledCloudRunInvocation[],
|
||||
workflowSteps: number,
|
||||
): RenderCost {
|
||||
let cloudRunUsd = 0;
|
||||
let anyEstimated = false;
|
||||
for (const inv of invocations) {
|
||||
const seconds = inv.durationMs / 1000;
|
||||
cloudRunUsd += seconds * inv.vcpu * CLOUD_RUN_USD_PER_VCPU_SECOND;
|
||||
cloudRunUsd += seconds * inv.memoryGib * CLOUD_RUN_USD_PER_GIB_SECOND;
|
||||
cloudRunUsd += CLOUD_RUN_USD_PER_REQUEST;
|
||||
if (inv.estimated) anyEstimated = true;
|
||||
}
|
||||
const workflowsUsd = workflowSteps * WORKFLOWS_USD_PER_STEP;
|
||||
const accruedSoFarUsd = roundUsd(cloudRunUsd + workflowsUsd);
|
||||
return {
|
||||
accruedSoFarUsd,
|
||||
displayCost: formatUsd(accruedSoFarUsd),
|
||||
breakdown: {
|
||||
cloudRunUsd: roundUsd(cloudRunUsd),
|
||||
workflowsUsd: roundUsd(workflowsUsd),
|
||||
gcsEstimate: "not-included",
|
||||
estimated: anyEstimated,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function roundUsd(usd: number): number {
|
||||
// Four decimal places — enough resolution for per-chunk granularity.
|
||||
// Anything finer is noise vs GCP's own rounding.
|
||||
return Math.round(usd * 10_000) / 10_000;
|
||||
}
|
||||
|
||||
function formatUsd(usd: number): string {
|
||||
return `$${usd.toFixed(4)}`;
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
/**
|
||||
* `deploySite` unit tests — content-addressed siteId, existence
|
||||
* short-circuit, and the upload path over `FakeGcs`.
|
||||
*/
|
||||
|
||||
import { afterEach, describe, expect, it } from "bun:test";
|
||||
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { asStorage, FakeGcs } from "../__fixtures__/fakeGcs.js";
|
||||
import { deploySite } from "./deploySite.js";
|
||||
|
||||
const tmpDirs: string[] = [];
|
||||
function mkProject(content: string): string {
|
||||
const dir = mkdtempSync(join(tmpdir(), "hf-site-"));
|
||||
tmpDirs.push(dir);
|
||||
writeFileSync(join(dir, "index.html"), content);
|
||||
return dir;
|
||||
}
|
||||
afterEach(() => {
|
||||
for (const d of tmpDirs.splice(0)) rmSync(d, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
describe("deploySite", () => {
|
||||
it("uploads and returns a content-addressed handle", async () => {
|
||||
const gcs = new FakeGcs();
|
||||
const dir = mkProject("<html>v1</html>");
|
||||
const handle = await deploySite({ projectDir: dir, bucketName: "b", storage: asStorage(gcs) });
|
||||
expect(handle.uploaded).toBe(true);
|
||||
expect(handle.projectGcsUri).toBe(`gs://b/sites/${handle.siteId}/project.tar.gz`);
|
||||
expect(gcs.objects.has(handle.projectGcsUri)).toBe(true);
|
||||
});
|
||||
|
||||
it("produces a stable siteId for identical content", async () => {
|
||||
const a = await deploySite({
|
||||
projectDir: mkProject("<html>same</html>"),
|
||||
bucketName: "b",
|
||||
storage: asStorage(new FakeGcs()),
|
||||
});
|
||||
const b = await deploySite({
|
||||
projectDir: mkProject("<html>same</html>"),
|
||||
bucketName: "b",
|
||||
storage: asStorage(new FakeGcs()),
|
||||
});
|
||||
expect(a.siteId).toBe(b.siteId);
|
||||
});
|
||||
|
||||
it("produces different siteIds for different content", async () => {
|
||||
const a = await deploySite({
|
||||
projectDir: mkProject("<html>one</html>"),
|
||||
bucketName: "b",
|
||||
storage: asStorage(new FakeGcs()),
|
||||
});
|
||||
const b = await deploySite({
|
||||
projectDir: mkProject("<html>two</html>"),
|
||||
bucketName: "b",
|
||||
storage: asStorage(new FakeGcs()),
|
||||
});
|
||||
expect(a.siteId).not.toBe(b.siteId);
|
||||
});
|
||||
|
||||
it("short-circuits the upload when the object already exists", async () => {
|
||||
const gcs = new FakeGcs();
|
||||
const dir = mkProject("<html>cache</html>");
|
||||
const first = await deploySite({ projectDir: dir, bucketName: "b", storage: asStorage(gcs) });
|
||||
expect(first.uploaded).toBe(true);
|
||||
|
||||
const second = await deploySite({ projectDir: dir, bucketName: "b", storage: asStorage(gcs) });
|
||||
expect(second.uploaded).toBe(false);
|
||||
expect(second.siteId).toBe(first.siteId);
|
||||
// Only one upload op total.
|
||||
expect(gcs.ops.filter((o) => o.kind === "upload").length).toBe(1);
|
||||
});
|
||||
|
||||
it("honours an explicit siteId override", async () => {
|
||||
const gcs = new FakeGcs();
|
||||
const handle = await deploySite({
|
||||
projectDir: mkProject("<html></html>"),
|
||||
bucketName: "b",
|
||||
siteId: "my-git-sha",
|
||||
storage: asStorage(gcs),
|
||||
});
|
||||
expect(handle.siteId).toBe("my-git-sha");
|
||||
expect(handle.projectGcsUri).toBe("gs://b/sites/my-git-sha/project.tar.gz");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,130 @@
|
||||
/**
|
||||
* `deploySite` — upload a project directory to GCS once per content hash
|
||||
* and return a reusable handle.
|
||||
*
|
||||
* `renderToCloudRun` calls this implicitly when no `siteHandle` is passed,
|
||||
* but exposing it as a standalone verb lets adopters bundle a project ahead
|
||||
* of time and reuse the handle across many renders without re-tarring the
|
||||
* project tree on every call.
|
||||
*
|
||||
* The handle is **content-addressed**: `siteId` is derived from a SHA-256
|
||||
* over the project files. Two `deploySite` calls on an unchanged tree
|
||||
* produce the same `siteId` and short-circuit the upload after a single
|
||||
* existence check.
|
||||
*/
|
||||
|
||||
import { mkdtempSync, rmSync, statSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { Storage } from "@google-cloud/storage";
|
||||
import { hashProjectDir } from "@hyperframes/producer/distributed";
|
||||
import { formatGcsUri, tarDirectory, uploadFileToGcs } from "../gcsTransport.js";
|
||||
|
||||
/** Options for {@link deploySite}. */
|
||||
export interface DeploySiteOptions {
|
||||
/** Local project directory containing `index.html` (and any composition assets). */
|
||||
projectDir: string;
|
||||
/** GCS bucket the Terraform module provisioned. */
|
||||
bucketName: string;
|
||||
/**
|
||||
* Override the content-addressed site id. Useful when the caller has a
|
||||
* stable external identifier they want to use (e.g. a git SHA); if unset,
|
||||
* the hash of the project tree picks it.
|
||||
*/
|
||||
siteId?: string;
|
||||
/** Injection seam for tests. Production callers leave unset. */
|
||||
storage?: Storage;
|
||||
}
|
||||
|
||||
/** Stable handle returned by {@link deploySite}. Pass back to {@link renderToCloudRun}. */
|
||||
export interface SiteHandle {
|
||||
/** Content-addressed (or caller-supplied) identifier; stable across re-uploads of the same tree. */
|
||||
siteId: string;
|
||||
/** Bucket the site landed in. Surfaced separately so callers don't have to re-parse `projectGcsUri`. */
|
||||
bucketName: string;
|
||||
/** Full `gs://bucket/sites/<siteId>/project.tar.gz` URI; pass through to `renderToCloudRun`. */
|
||||
projectGcsUri: string;
|
||||
/** Tarball size in bytes; useful for "did we actually skip the upload?" assertions. */
|
||||
bytes: number;
|
||||
/** ISO timestamp of the most recent upload OR the existing object the short-circuit found. */
|
||||
uploadedAt: string;
|
||||
/** `false` if the object already existed and we skipped the upload. */
|
||||
uploaded: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Upload `projectDir` to `gs://bucketName/sites/<siteId>/project.tar.gz`.
|
||||
*
|
||||
* Short-circuits when an object with the same key already exists in the
|
||||
* bucket — `siteId` derives from the project's content hash, so the same
|
||||
* bytes produce the same key, and re-uploading would be redundant.
|
||||
*/
|
||||
// fallow-ignore-next-line complexity
|
||||
export async function deploySite(opts: DeploySiteOptions): Promise<SiteHandle> {
|
||||
if (!statSync(opts.projectDir).isDirectory()) {
|
||||
throw new Error(`[deploySite] projectDir is not a directory: ${opts.projectDir}`);
|
||||
}
|
||||
|
||||
const siteId = opts.siteId ?? hashProjectDir(opts.projectDir);
|
||||
const key = `sites/${siteId}/project.tar.gz`;
|
||||
const projectGcsUri = formatGcsUri({ bucket: opts.bucketName, key });
|
||||
const storage = opts.storage ?? new Storage();
|
||||
const file = storage.bucket(opts.bucketName).file(key);
|
||||
|
||||
// Existence short-circuit. Adopters re-rendering the same project on a
|
||||
// tight inner loop (CI smoke, demo flows) save the tar+gzip+upload pass
|
||||
// on every iteration.
|
||||
const existing = await headObject(file);
|
||||
if (existing) {
|
||||
return {
|
||||
siteId,
|
||||
bucketName: opts.bucketName,
|
||||
projectGcsUri,
|
||||
bytes: existing.bytes,
|
||||
uploadedAt: existing.lastModified,
|
||||
uploaded: false,
|
||||
};
|
||||
}
|
||||
|
||||
const workdir = mkdtempSync(join(tmpdir(), "hf-deploy-site-"));
|
||||
try {
|
||||
const tarball = join(workdir, "project.tar.gz");
|
||||
await tarDirectory(opts.projectDir, tarball);
|
||||
const size = statSync(tarball).size;
|
||||
await uploadFileToGcs(storage, tarball, projectGcsUri, "application/gzip");
|
||||
return {
|
||||
siteId,
|
||||
bucketName: opts.bucketName,
|
||||
projectGcsUri,
|
||||
bytes: size,
|
||||
uploadedAt: new Date().toISOString(),
|
||||
uploaded: true,
|
||||
};
|
||||
} finally {
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Narrow surface of the `@google-cloud/storage` `File` this module uses —
|
||||
* lets the test double implement just `exists()` + `getMetadata()` without
|
||||
* pulling the full client type.
|
||||
*/
|
||||
interface FileLike {
|
||||
exists(): Promise<[boolean, ...unknown[]]>;
|
||||
getMetadata(): Promise<[{ size?: string | number; updated?: string }, ...unknown[]]>;
|
||||
}
|
||||
|
||||
// fallow-ignore-next-line complexity
|
||||
async function headObject(file: FileLike): Promise<{ bytes: number; lastModified: string } | null> {
|
||||
const [exists] = await file.exists();
|
||||
if (!exists) return null;
|
||||
const [meta] = await file.getMetadata();
|
||||
const sizeRaw = meta.size;
|
||||
const bytes =
|
||||
typeof sizeRaw === "string" ? Number(sizeRaw) : typeof sizeRaw === "number" ? sizeRaw : 0;
|
||||
return {
|
||||
bytes: Number.isFinite(bytes) ? bytes : 0,
|
||||
lastModified: meta.updated ?? new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
/**
|
||||
* `getRenderProgress` unit tests — state mapping + parsing the accumulated
|
||||
* workflow result into frame totals, output file, and cost.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import {
|
||||
type ExecutionRecord,
|
||||
type ExecutionsGetClientLike,
|
||||
getRenderProgress,
|
||||
} from "./getRenderProgress.js";
|
||||
|
||||
function fakeExecutions(record: ExecutionRecord): ExecutionsGetClientLike {
|
||||
return {
|
||||
async getExecution(_req: { name: string }) {
|
||||
return [record] as [ExecutionRecord];
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const accumulated = JSON.stringify({
|
||||
Plan: { TotalFrames: 90, DurationMs: 4000 },
|
||||
Chunks: [
|
||||
{ FramesEncoded: 30, DurationMs: 8000 },
|
||||
{ FramesEncoded: 30, DurationMs: 8000 },
|
||||
{ FramesEncoded: 30, DurationMs: 8000 },
|
||||
],
|
||||
Assemble: { OutputGcsUri: "gs://b/renders/r1/output.mp4", FileSize: 123456, DurationMs: 3000 },
|
||||
});
|
||||
|
||||
describe("getRenderProgress", () => {
|
||||
it("reports running with no frame data while ACTIVE", async () => {
|
||||
const p = await getRenderProgress({
|
||||
executionName: "x",
|
||||
executions: fakeExecutions({ state: "ACTIVE", startTime: { seconds: 1700000000 } }),
|
||||
});
|
||||
expect(p.status).toBe("running");
|
||||
expect(p.overallProgress).toBe(0);
|
||||
expect(p.totalFrames).toBeNull();
|
||||
expect(p.fatalErrorEncountered).toBe(false);
|
||||
});
|
||||
|
||||
it("reports succeeded with parsed frames + cost", async () => {
|
||||
const p = await getRenderProgress({
|
||||
executionName: "x",
|
||||
vcpu: 4,
|
||||
memoryGib: 16,
|
||||
executions: fakeExecutions({
|
||||
state: "SUCCEEDED",
|
||||
result: accumulated,
|
||||
startTime: { seconds: 1700000000 },
|
||||
endTime: { seconds: 1700000031 },
|
||||
}),
|
||||
});
|
||||
expect(p.status).toBe("succeeded");
|
||||
expect(p.overallProgress).toBe(1);
|
||||
expect(p.totalFrames).toBe(90);
|
||||
expect(p.framesRendered).toBe(90);
|
||||
expect(p.invocationsObserved).toBe(5); // plan + 3 chunks + assemble
|
||||
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);
|
||||
});
|
||||
|
||||
it("maps FAILED to a fatal error and surfaces the error payload", async () => {
|
||||
const p = await getRenderProgress({
|
||||
executionName: "x",
|
||||
executions: fakeExecutions({
|
||||
state: "FAILED",
|
||||
error: { payload: "boom", context: "renderChunk" },
|
||||
}),
|
||||
});
|
||||
expect(p.status).toBe("failed");
|
||||
expect(p.fatalErrorEncountered).toBe(true);
|
||||
expect(p.errors[0]?.cause).toBe("boom");
|
||||
expect(p.errors[0]?.state).toBe("renderChunk");
|
||||
});
|
||||
|
||||
it("extracts the handler error name from a wrapped http failure payload", async () => {
|
||||
// Workflows wraps an http step failure as { code, message, body }, where
|
||||
// body is the handler's JSON { error, message }.
|
||||
const payload = JSON.stringify({
|
||||
code: 400,
|
||||
message: "HTTP server responded with error code 400",
|
||||
body: JSON.stringify({ error: "PLAN_HASH_MISMATCH", message: "mismatch" }),
|
||||
});
|
||||
const p = await getRenderProgress({
|
||||
executionName: "x",
|
||||
executions: fakeExecutions({ state: "FAILED", error: { payload, context: "renderChunk" } }),
|
||||
});
|
||||
expect(p.errors[0]?.error).toBe("PLAN_HASH_MISMATCH");
|
||||
});
|
||||
|
||||
it("maps CANCELLED", async () => {
|
||||
const p = await getRenderProgress({
|
||||
executionName: "x",
|
||||
executions: fakeExecutions({ state: "CANCELLED" }),
|
||||
});
|
||||
expect(p.status).toBe("cancelled");
|
||||
expect(p.fatalErrorEncountered).toBe(true);
|
||||
});
|
||||
|
||||
it("requires an executionName", async () => {
|
||||
await expect(
|
||||
getRenderProgress({ executionName: "", executions: fakeExecutions({}) }),
|
||||
).rejects.toThrow(/executionName is required/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,268 @@
|
||||
/**
|
||||
* `getRenderProgress` — read-only progress + cost snapshot for a single
|
||||
* render started by {@link renderToCloudRun}.
|
||||
*
|
||||
* Pulls one `GetExecution` per call. Cloud Workflows does not surface
|
||||
* per-step payloads through the basic Executions API the way Step Functions
|
||||
* exposes its history, so this reader takes a different tack than the AWS
|
||||
* adapter: the workflow definition **accumulates** each step's result body
|
||||
* (Plan + every RenderChunk + Assemble) and returns them as one structured
|
||||
* object. On success we parse that object for frame totals, the output
|
||||
* 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.
|
||||
*/
|
||||
|
||||
import {
|
||||
type BilledCloudRunInvocation,
|
||||
computeRenderCost,
|
||||
type RenderCost,
|
||||
} from "./costAccounting.js";
|
||||
|
||||
/** Normalised render status. Maps from Cloud Workflows execution states. */
|
||||
export type RenderStatus = "running" | "succeeded" | "failed" | "cancelled" | "unknown";
|
||||
|
||||
/** One error surfaced by the execution. */
|
||||
export interface RenderError {
|
||||
/** Step the failure surfaced in, when recoverable from the error context; else `<execution>`. */
|
||||
state: string;
|
||||
/** Error class / type. */
|
||||
error: string;
|
||||
/** Cause string (often a stringified JSON payload from the handler). */
|
||||
cause: string;
|
||||
}
|
||||
|
||||
/** 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. */
|
||||
overallProgress: number;
|
||||
framesRendered: number;
|
||||
/** `null` until the execution succeeds and the accumulated plan result is read. */
|
||||
totalFrames: number | null;
|
||||
/** Cloud Run invocations the workflow scheduled (Plan + chunks + Assemble), when known. */
|
||||
invocationsObserved: number;
|
||||
costs: RenderCost;
|
||||
/** Final output object if Assemble succeeded; `null` otherwise. */
|
||||
outputFile: { gcsUri: string; bytes: number | null } | null;
|
||||
errors: RenderError[];
|
||||
/** `true` once the execution has terminated in a non-success state. */
|
||||
fatalErrorEncountered: boolean;
|
||||
startedAt: string;
|
||||
endedAt: string | null;
|
||||
}
|
||||
|
||||
/** Protobuf Timestamp shape the gapic client returns for start/end times. */
|
||||
interface ProtoTimestamp {
|
||||
seconds?: number | string | null;
|
||||
nanos?: number | null;
|
||||
}
|
||||
|
||||
/** Subset of a Cloud Workflows Execution this reader consumes. */
|
||||
export interface ExecutionRecord {
|
||||
name?: string | null;
|
||||
state?: string | null;
|
||||
result?: string | null;
|
||||
error?: { payload?: string | null; context?: string | null } | null;
|
||||
startTime?: ProtoTimestamp | string | null;
|
||||
endTime?: ProtoTimestamp | string | null;
|
||||
}
|
||||
|
||||
/** Minimal surface of `@google-cloud/workflows`' `ExecutionsClient` for reads. */
|
||||
export interface ExecutionsGetClientLike {
|
||||
getExecution(req: { name: string }): Promise<[ExecutionRecord, ...unknown[]]>;
|
||||
}
|
||||
|
||||
/** Options for {@link getRenderProgress}. */
|
||||
export interface GetRenderProgressOptions {
|
||||
/** Server-assigned execution resource name from a {@link renderToCloudRun} call. */
|
||||
executionName: string;
|
||||
/** vCPU the Cloud Run service is configured with (for cost). Default 4. */
|
||||
vcpu?: number;
|
||||
/** Memory in GiB the Cloud Run service is configured with (for cost). Default 16. */
|
||||
memoryGib?: number;
|
||||
/** Test injection seam — production callers leave unset. */
|
||||
executions?: ExecutionsGetClientLike;
|
||||
}
|
||||
|
||||
const DEFAULT_VCPU = 4;
|
||||
const DEFAULT_MEMORY_GIB = 16;
|
||||
|
||||
/** Result body the handler returns for each action; the workflow accumulates these. */
|
||||
interface AccumulatedResult {
|
||||
Plan?: { TotalFrames?: number; DurationMs?: number } | null;
|
||||
Chunks?: Array<{ FramesEncoded?: number; DurationMs?: number } | null> | null;
|
||||
Assemble?: {
|
||||
OutputGcsUri?: string;
|
||||
FileSize?: number;
|
||||
FramesEncoded?: number;
|
||||
DurationMs?: number;
|
||||
} | null;
|
||||
}
|
||||
|
||||
/** Pull a current progress snapshot for one render. */
|
||||
// fallow-ignore-next-line complexity
|
||||
export async function getRenderProgress(opts: GetRenderProgressOptions): Promise<RenderProgress> {
|
||||
if (!opts.executionName) {
|
||||
throw new Error("[getRenderProgress] executionName is required");
|
||||
}
|
||||
const executions = opts.executions ?? (await defaultExecutionsClient());
|
||||
const vcpu = opts.vcpu ?? DEFAULT_VCPU;
|
||||
const memoryGib = opts.memoryGib ?? DEFAULT_MEMORY_GIB;
|
||||
|
||||
const [execution] = await executions.getExecution({ name: opts.executionName });
|
||||
const status = mapState(execution.state);
|
||||
const startedAt = toIso(execution.startTime) ?? new Date(0).toISOString();
|
||||
const endedAt = toIso(execution.endTime);
|
||||
|
||||
const errors: RenderError[] = [];
|
||||
if (execution.error) {
|
||||
errors.push({
|
||||
state: execution.error.context ?? "<execution>",
|
||||
error: extractErrorName(execution.error.payload) ?? "ExecutionError",
|
||||
cause: execution.error.payload ?? "",
|
||||
});
|
||||
}
|
||||
|
||||
// Default snapshot: running / unknown — no frame or cost data until the
|
||||
// accumulated result is available on success.
|
||||
if (status !== "succeeded") {
|
||||
return {
|
||||
status,
|
||||
overallProgress: 0,
|
||||
framesRendered: 0,
|
||||
totalFrames: null,
|
||||
invocationsObserved: 0,
|
||||
costs: computeRenderCost([], 0),
|
||||
outputFile: null,
|
||||
errors,
|
||||
fatalErrorEncountered: status === "failed" || status === "cancelled",
|
||||
startedAt,
|
||||
endedAt,
|
||||
};
|
||||
}
|
||||
|
||||
const acc = parseAccumulated(execution.result);
|
||||
const chunks = acc.Chunks?.filter((c): c is NonNullable<typeof c> => c != null) ?? [];
|
||||
const framesRendered = chunks.reduce((sum, c) => sum + (c.FramesEncoded ?? 0), 0);
|
||||
const totalFrames = typeof acc.Plan?.TotalFrames === "number" ? acc.Plan.TotalFrames : null;
|
||||
|
||||
const invocations: BilledCloudRunInvocation[] = [];
|
||||
const pushInv = (durationMs: number | undefined): void => {
|
||||
invocations.push({
|
||||
durationMs: typeof durationMs === "number" ? durationMs : 0,
|
||||
vcpu,
|
||||
memoryGib,
|
||||
estimated: typeof durationMs !== "number",
|
||||
});
|
||||
};
|
||||
if (acc.Plan) pushInv(acc.Plan.DurationMs);
|
||||
for (const c of chunks) pushInv(c.DurationMs);
|
||||
if (acc.Assemble) pushInv(acc.Assemble.DurationMs);
|
||||
|
||||
// Workflow step count: Plan + N chunks + Assemble + a small constant of
|
||||
// control steps (BuildChunkList, AssertChunkCount, the map scaffold).
|
||||
const workflowSteps = invocations.length + 4;
|
||||
const costs = computeRenderCost(invocations, workflowSteps);
|
||||
|
||||
const outputGcsUri = acc.Assemble?.OutputGcsUri;
|
||||
const outputFile = outputGcsUri
|
||||
? {
|
||||
gcsUri: outputGcsUri,
|
||||
bytes: typeof acc.Assemble?.FileSize === "number" ? acc.Assemble.FileSize : null,
|
||||
}
|
||||
: null;
|
||||
|
||||
return {
|
||||
status,
|
||||
overallProgress: 1,
|
||||
framesRendered,
|
||||
totalFrames,
|
||||
invocationsObserved: invocations.length,
|
||||
costs,
|
||||
outputFile,
|
||||
errors,
|
||||
fatalErrorEncountered: false,
|
||||
startedAt,
|
||||
endedAt,
|
||||
};
|
||||
}
|
||||
|
||||
// fallow-ignore-next-line complexity
|
||||
function mapState(state: string | null | undefined): RenderStatus {
|
||||
switch (state) {
|
||||
case "ACTIVE":
|
||||
case "QUEUED":
|
||||
return "running";
|
||||
case "SUCCEEDED":
|
||||
return "succeeded";
|
||||
case "FAILED":
|
||||
case "UNAVAILABLE":
|
||||
return "failed";
|
||||
case "CANCELLED":
|
||||
return "cancelled";
|
||||
default:
|
||||
return "unknown";
|
||||
}
|
||||
}
|
||||
|
||||
// fallow-ignore-next-line complexity
|
||||
function parseAccumulated(result: string | null | undefined): AccumulatedResult {
|
||||
if (!result) return {};
|
||||
try {
|
||||
const parsed = JSON.parse(result) as unknown;
|
||||
if (parsed && typeof parsed === "object") return parsed as AccumulatedResult;
|
||||
} catch {
|
||||
// Non-JSON result — treat as empty so cost/frames degrade to zero
|
||||
// rather than throwing on a snapshot read.
|
||||
}
|
||||
return {};
|
||||
}
|
||||
|
||||
/**
|
||||
* Best-effort pull of the handler's error name out of a Workflows failure
|
||||
* payload. On an http step failure, Workflows wraps the response as
|
||||
* `{ code, message, body, ... }` where `body` is the handler's JSON
|
||||
* `{ error, message }`. We dig out `error` (the typed name like
|
||||
* `PLAN_HASH_MISMATCH`) so triage sees the real cause, not a generic label.
|
||||
* Returns undefined for any shape we don't recognise — never throws.
|
||||
*/
|
||||
// fallow-ignore-next-line complexity
|
||||
function extractErrorName(payload: string | null | undefined): string | undefined {
|
||||
if (!payload) return undefined;
|
||||
try {
|
||||
const outer = JSON.parse(payload) as { error?: unknown; body?: unknown };
|
||||
if (typeof outer.error === "string") return outer.error;
|
||||
if (typeof outer.body === "string") {
|
||||
const inner = JSON.parse(outer.body) as { error?: unknown };
|
||||
if (typeof inner.error === "string") return inner.error;
|
||||
} else if (outer.body && typeof outer.body === "object") {
|
||||
const inner = outer.body as { error?: unknown };
|
||||
if (typeof inner.error === "string") return inner.error;
|
||||
}
|
||||
} catch {
|
||||
// Non-JSON / unexpected shape — fall through to the generic label.
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
// fallow-ignore-next-line complexity
|
||||
function toIso(ts: ProtoTimestamp | string | null | undefined): string | null {
|
||||
if (ts == null) return null;
|
||||
if (typeof ts === "string") return ts;
|
||||
const seconds = ts.seconds == null ? null : Number(ts.seconds);
|
||||
if (seconds == null || !Number.isFinite(seconds)) return null;
|
||||
const ms = seconds * 1000 + (ts.nanos ?? 0) / 1e6;
|
||||
return new Date(ms).toISOString();
|
||||
}
|
||||
|
||||
async function defaultExecutionsClient(): Promise<ExecutionsGetClientLike> {
|
||||
const mod = await import("@google-cloud/workflows");
|
||||
const client = new mod.ExecutionsClient();
|
||||
return client as unknown as ExecutionsGetClientLike;
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
/**
|
||||
* SDK subpath export — `@hyperframes/gcp-cloud-run/sdk`.
|
||||
*
|
||||
* Pulled into its own subpath so consumers that only drive renders (CLI, CI
|
||||
* scripts, adopter tooling) don't pay the cost of importing `./server.js`,
|
||||
* which transitively pulls `puppeteer-core` into the module graph. The SDK
|
||||
* files here are GCS + Workflows clients only — safe to load in any Node
|
||||
* environment.
|
||||
*/
|
||||
|
||||
export { deploySite, type DeploySiteOptions, type SiteHandle } from "./deploySite.js";
|
||||
export {
|
||||
type ExecutionsClientLike,
|
||||
renderToCloudRun,
|
||||
type RenderHandle,
|
||||
type RenderToCloudRunOptions,
|
||||
} from "./renderToCloudRun.js";
|
||||
export {
|
||||
type ExecutionRecord,
|
||||
type ExecutionsGetClientLike,
|
||||
getRenderProgress,
|
||||
type GetRenderProgressOptions,
|
||||
type RenderError,
|
||||
type RenderProgress,
|
||||
type RenderStatus,
|
||||
} from "./getRenderProgress.js";
|
||||
export {
|
||||
type BilledCloudRunInvocation,
|
||||
computeRenderCost,
|
||||
type RenderCost,
|
||||
} from "./costAccounting.js";
|
||||
export {
|
||||
InvalidConfigError,
|
||||
MAX_WORKFLOWS_INPUT_BYTES,
|
||||
validateDistributedRenderConfig,
|
||||
validateVariablesPayload,
|
||||
validateWorkflowsInputSize,
|
||||
} from "./validateConfig.js";
|
||||
export type { SerializableDistributedRenderConfig } from "../events.js";
|
||||
export type { DistributedFormat } from "../formatExtension.js";
|
||||
@@ -0,0 +1,127 @@
|
||||
/**
|
||||
* `renderToCloudRun` unit tests — argument assembly, required-field
|
||||
* validation, and the CreateExecution call over a fake ExecutionsClient.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import type { SerializableDistributedRenderConfig } from "../events.js";
|
||||
import { type ExecutionsClientLike, renderToCloudRun } from "./renderToCloudRun.js";
|
||||
import type { SiteHandle } from "./deploySite.js";
|
||||
|
||||
const config = {
|
||||
fps: 30,
|
||||
width: 1920,
|
||||
height: 1080,
|
||||
format: "mp4",
|
||||
} as SerializableDistributedRenderConfig;
|
||||
|
||||
const site: SiteHandle = {
|
||||
siteId: "abc",
|
||||
bucketName: "b",
|
||||
projectGcsUri: "gs://b/sites/abc/project.tar.gz",
|
||||
bytes: 100,
|
||||
uploadedAt: "2026-06-06T00:00:00Z",
|
||||
uploaded: true,
|
||||
};
|
||||
|
||||
class FakeExecutions implements ExecutionsClientLike {
|
||||
lastArgument: string | null = null;
|
||||
lastParent: string | null = null;
|
||||
|
||||
workflowPath(project: string, location: string, workflow: string): string {
|
||||
return `projects/${project}/locations/${location}/workflows/${workflow}`;
|
||||
}
|
||||
|
||||
async createExecution(req: {
|
||||
parent: string;
|
||||
execution: { argument: string };
|
||||
}): Promise<[{ name?: string | null; state?: string | null }]> {
|
||||
this.lastParent = req.parent;
|
||||
this.lastArgument = req.execution.argument;
|
||||
return [{ name: `${req.parent}/executions/exec-123`, state: "ACTIVE" }];
|
||||
}
|
||||
}
|
||||
|
||||
function opts(executions: ExecutionsClientLike) {
|
||||
return {
|
||||
siteHandle: site,
|
||||
config,
|
||||
bucketName: "b",
|
||||
projectId: "proj",
|
||||
location: "us-central1",
|
||||
workflowId: "hyperframes-render",
|
||||
serviceUrl: "https://render-abc.run.app",
|
||||
renderId: "hf-render-fixed",
|
||||
executions,
|
||||
};
|
||||
}
|
||||
|
||||
describe("renderToCloudRun", () => {
|
||||
it("starts an execution and returns a handle", async () => {
|
||||
const fake = new FakeExecutions();
|
||||
const handle = await renderToCloudRun(opts(fake));
|
||||
expect(handle.renderId).toBe("hf-render-fixed");
|
||||
expect(handle.executionName).toBe(
|
||||
"projects/proj/locations/us-central1/workflows/hyperframes-render/executions/exec-123",
|
||||
);
|
||||
expect(handle.outputGcsUri).toBe("gs://b/renders/hf-render-fixed/output.mp4");
|
||||
expect(handle.projectGcsUri).toBe("gs://b/sites/abc/project.tar.gz");
|
||||
});
|
||||
|
||||
it("builds the workflow argument the YAML expects", async () => {
|
||||
const fake = new FakeExecutions();
|
||||
await renderToCloudRun(opts(fake));
|
||||
const arg = JSON.parse(fake.lastArgument ?? "{}");
|
||||
expect(arg.RenderId).toBe("hf-render-fixed");
|
||||
expect(arg.ProjectGcsUri).toBe("gs://b/sites/abc/project.tar.gz");
|
||||
expect(arg.PlanOutputGcsPrefix).toBe("gs://b/renders/hf-render-fixed/");
|
||||
expect(arg.OutputGcsUri).toBe("gs://b/renders/hf-render-fixed/output.mp4");
|
||||
expect(arg.ServiceUrl).toBe("https://render-abc.run.app");
|
||||
expect(arg.Config.format).toBe("mp4");
|
||||
expect(fake.lastParent).toBe(
|
||||
"projects/proj/locations/us-central1/workflows/hyperframes-render",
|
||||
);
|
||||
});
|
||||
|
||||
it("derives the output extension from the format", async () => {
|
||||
const fake = new FakeExecutions();
|
||||
const handle = await renderToCloudRun({
|
||||
...opts(fake),
|
||||
config: { ...config, format: "webm" } as SerializableDistributedRenderConfig,
|
||||
});
|
||||
expect(handle.outputGcsUri).toBe("gs://b/renders/hf-render-fixed/output.webm");
|
||||
});
|
||||
|
||||
it("requires serviceUrl", async () => {
|
||||
const fake = new FakeExecutions();
|
||||
await expect(renderToCloudRun({ ...opts(fake), serviceUrl: "" })).rejects.toThrow(
|
||||
/serviceUrl is required/,
|
||||
);
|
||||
});
|
||||
|
||||
it("requires a siteHandle or projectDir", async () => {
|
||||
const fake = new FakeExecutions();
|
||||
const { siteHandle, ...rest } = opts(fake);
|
||||
void siteHandle;
|
||||
await expect(renderToCloudRun(rest)).rejects.toThrow(/siteHandle or projectDir/);
|
||||
});
|
||||
|
||||
it("validates the config before any GCP call", async () => {
|
||||
const fake = new FakeExecutions();
|
||||
await expect(
|
||||
renderToCloudRun({ ...opts(fake), config: { ...config, fps: 25 } as never }),
|
||||
).rejects.toThrow(/config\.fps/);
|
||||
expect(fake.lastArgument).toBeNull();
|
||||
});
|
||||
|
||||
it("rejects a renderId that could escape the GCS key prefix", async () => {
|
||||
const fake = new FakeExecutions();
|
||||
await expect(renderToCloudRun({ ...opts(fake), renderId: "../escape" })).rejects.toThrow(
|
||||
/renderId must match/,
|
||||
);
|
||||
await expect(renderToCloudRun({ ...opts(fake), renderId: "has/slash" })).rejects.toThrow(
|
||||
/renderId must match/,
|
||||
);
|
||||
expect(fake.lastArgument).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,188 @@
|
||||
/**
|
||||
* `renderToCloudRun` — start a distributed render against an already-deployed
|
||||
* Cloud Run service + Cloud Workflows definition and return a handle the
|
||||
* caller can poll with {@link getRenderProgress}.
|
||||
*
|
||||
* The function does *not* wait for the render to finish. Cloud Workflows
|
||||
* executions can run for hours; blocking the caller's process on the
|
||||
* execution is the wrong default. The returned `RenderHandle` carries
|
||||
* everything the progress / cost / download paths need.
|
||||
*
|
||||
* Wire order:
|
||||
* 1. Validate config (typed throw before any GCP call).
|
||||
* 2. `deploySite` if no `siteHandle` was provided.
|
||||
* 3. `CreateExecution` against the workflow with the argument shape the
|
||||
* `packages/gcp-cloud-run/terraform/workflow.yaml` definition expects.
|
||||
* 4. Return handle. The GCS `outputKey` is deterministic from the
|
||||
* client-generated `renderId` so the caller can predict the final
|
||||
* object URL before the (server-assigned) execution id exists.
|
||||
*
|
||||
* Unlike Step Functions, Cloud Workflows assigns the execution id
|
||||
* server-side, so we cannot use it as the GCS prefix. We mint a `renderId`
|
||||
* (uuid) client-side, use it for every GCS path, and pass it into the
|
||||
* workflow argument; the server-assigned execution resource name is tracked
|
||||
* separately for polling.
|
||||
*/
|
||||
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type { Storage } from "@google-cloud/storage";
|
||||
import type { SerializableDistributedRenderConfig } from "../events.js";
|
||||
import { formatExtension } from "../formatExtension.js";
|
||||
import { formatGcsUri } from "../gcsTransport.js";
|
||||
import { deploySite, type SiteHandle } from "./deploySite.js";
|
||||
import { validateDistributedRenderConfig, validateWorkflowsInputSize } from "./validateConfig.js";
|
||||
|
||||
/**
|
||||
* Minimal surface of `@google-cloud/workflows`' `ExecutionsClient` that
|
||||
* this module needs. The real client satisfies this; tests inject a double.
|
||||
*/
|
||||
export interface ExecutionsClientLike {
|
||||
workflowPath(project: string, location: string, workflow: string): string;
|
||||
createExecution(req: {
|
||||
parent: string;
|
||||
execution: { argument: string };
|
||||
}): Promise<[{ name?: string | null; state?: string | null }, ...unknown[]]>;
|
||||
}
|
||||
|
||||
/** Options for {@link renderToCloudRun}. */
|
||||
export interface RenderToCloudRunOptions {
|
||||
/** Local project directory. Required when `siteHandle` is not supplied. */
|
||||
projectDir?: string;
|
||||
/** Re-use an existing `deploySite` upload (skips tar+GCS upload). */
|
||||
siteHandle?: SiteHandle;
|
||||
/** Validated `SerializableDistributedRenderConfig` (no logger / abortSignal). */
|
||||
config: SerializableDistributedRenderConfig;
|
||||
/** GCS bucket from the Terraform output (`render_bucket_name`). */
|
||||
bucketName: string;
|
||||
/** GCP project id hosting the workflow. */
|
||||
projectId: string;
|
||||
/** Workflow location, e.g. `us-central1`. */
|
||||
location: string;
|
||||
/** Workflow id from the Terraform output (`workflow_name`). */
|
||||
workflowId: string;
|
||||
/**
|
||||
* HTTPS URL of the deployed Cloud Run render service (Terraform output
|
||||
* `service_url`). The workflow POSTs every step (plan / renderChunk /
|
||||
* assemble) to this URL; passed as an execution argument so the workflow
|
||||
* definition stays free of hard-coded URLs.
|
||||
*/
|
||||
serviceUrl: string;
|
||||
/**
|
||||
* Final output GCS key. Defaults to `renders/<renderId>/output.<ext>`
|
||||
* where `<ext>` is derived from `config.format`.
|
||||
*/
|
||||
outputKey?: string;
|
||||
/**
|
||||
* Client-generated render id. Defaults to `hf-render-<uuid>`. Used as the
|
||||
* GCS key prefix and echoed into the workflow argument; not the same as
|
||||
* the server-assigned execution id.
|
||||
*/
|
||||
renderId?: string;
|
||||
/** Test injection seam — production callers leave unset. */
|
||||
executions?: ExecutionsClientLike;
|
||||
/** Test injection seam — propagated to `deploySite` when applicable. */
|
||||
storage?: Storage;
|
||||
}
|
||||
|
||||
/** Stable identifier + every URL/name the caller needs to follow the render. */
|
||||
export interface RenderHandle {
|
||||
/** Client-generated render id; the GCS prefix everything lands under. */
|
||||
renderId: string;
|
||||
/** Server-assigned execution resource name; pass to {@link getRenderProgress}. */
|
||||
executionName: string;
|
||||
bucketName: string;
|
||||
workflowId: string;
|
||||
outputGcsUri: string;
|
||||
projectGcsUri: string;
|
||||
startedAt: string;
|
||||
}
|
||||
|
||||
// fallow-ignore-next-line complexity
|
||||
export async function renderToCloudRun(opts: RenderToCloudRunOptions): Promise<RenderHandle> {
|
||||
validateDistributedRenderConfig(opts.config);
|
||||
|
||||
if (!opts.bucketName) throw new Error("[renderToCloudRun] bucketName is required");
|
||||
if (!opts.projectId) throw new Error("[renderToCloudRun] projectId is required");
|
||||
if (!opts.location) throw new Error("[renderToCloudRun] location is required");
|
||||
if (!opts.workflowId) throw new Error("[renderToCloudRun] workflowId is required");
|
||||
if (!opts.serviceUrl) throw new Error("[renderToCloudRun] serviceUrl is required");
|
||||
if (!opts.siteHandle && !opts.projectDir) {
|
||||
throw new Error("[renderToCloudRun] either siteHandle or projectDir must be supplied");
|
||||
}
|
||||
|
||||
const renderId = opts.renderId ?? `hf-render-${randomUUID()}`;
|
||||
// `renderId` is interpolated directly into GCS object keys
|
||||
// (`renders/<renderId>/…`). Reject anything that could escape that prefix
|
||||
// or build a malformed key — `..`, slashes, or other path metacharacters —
|
||||
// so a caller-supplied id can't collide with or overwrite another render's
|
||||
// artifacts elsewhere in the bucket.
|
||||
if (!/^[A-Za-z0-9._-]+$/.test(renderId) || renderId.includes("..")) {
|
||||
throw new Error(
|
||||
`[renderToCloudRun] renderId must match [A-Za-z0-9._-]+ and not contain "..": ${JSON.stringify(renderId)}`,
|
||||
);
|
||||
}
|
||||
const ext = formatExtension(opts.config.format);
|
||||
const outputKey = opts.outputKey ?? `renders/${renderId}/output${ext}`;
|
||||
const planOutputGcsPrefix = formatGcsUri({
|
||||
bucket: opts.bucketName,
|
||||
key: `renders/${renderId}/`,
|
||||
});
|
||||
const outputGcsUri = formatGcsUri({ bucket: opts.bucketName, key: outputKey });
|
||||
|
||||
const site =
|
||||
opts.siteHandle ??
|
||||
(await deploySite({
|
||||
projectDir: opts.projectDir as string,
|
||||
bucketName: opts.bucketName,
|
||||
storage: opts.storage,
|
||||
}));
|
||||
|
||||
const argument = {
|
||||
RenderId: renderId,
|
||||
ProjectGcsUri: site.projectGcsUri,
|
||||
PlanOutputGcsPrefix: planOutputGcsPrefix,
|
||||
OutputGcsUri: outputGcsUri,
|
||||
ServiceUrl: opts.serviceUrl,
|
||||
Config: opts.config,
|
||||
};
|
||||
|
||||
// Reject oversize input client-side. Cloud Workflows caps the execution
|
||||
// argument at 512 KiB; without this check, input bloat (typically from
|
||||
// `config.variables` containing inlined media) surfaces as an opaque
|
||||
// server-side error after the execution starts, far from the caller's
|
||||
// stack frame.
|
||||
validateWorkflowsInputSize(argument);
|
||||
|
||||
const executions = opts.executions ?? (await defaultExecutionsClient());
|
||||
const parent = executions.workflowPath(opts.projectId, opts.location, opts.workflowId);
|
||||
const startedAt = new Date().toISOString();
|
||||
const [execution] = await executions.createExecution({
|
||||
parent,
|
||||
execution: { argument: JSON.stringify(argument) },
|
||||
});
|
||||
|
||||
if (!execution.name) {
|
||||
throw new Error("[renderToCloudRun] CreateExecution returned no execution name");
|
||||
}
|
||||
|
||||
return {
|
||||
renderId,
|
||||
executionName: execution.name,
|
||||
bucketName: opts.bucketName,
|
||||
workflowId: opts.workflowId,
|
||||
outputGcsUri,
|
||||
projectGcsUri: site.projectGcsUri,
|
||||
startedAt,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Lazily import the real `@google-cloud/workflows` ExecutionsClient. Dynamic
|
||||
* so SDK consumers that only call `validateDistributedRenderConfig` (or
|
||||
* inject their own client) don't pay the import cost.
|
||||
*/
|
||||
async function defaultExecutionsClient(): Promise<ExecutionsClientLike> {
|
||||
const mod = await import("@google-cloud/workflows");
|
||||
const client = new mod.ExecutionsClient();
|
||||
return client as unknown as ExecutionsClientLike;
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
/**
|
||||
* `validateDistributedRenderConfig` + `validateWorkflowsInputSize` unit
|
||||
* tests. Pins the shape rejections the SDK surfaces synchronously before a
|
||||
* Cloud Workflows execution starts.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from "bun:test";
|
||||
import type { SerializableDistributedRenderConfig } from "../events.js";
|
||||
import {
|
||||
InvalidConfigError,
|
||||
MAX_WORKFLOWS_INPUT_BYTES,
|
||||
validateDistributedRenderConfig,
|
||||
validateVariablesPayload,
|
||||
validateWorkflowsInputSize,
|
||||
} from "./validateConfig.js";
|
||||
|
||||
function base(): SerializableDistributedRenderConfig {
|
||||
return {
|
||||
fps: 30,
|
||||
width: 1920,
|
||||
height: 1080,
|
||||
format: "mp4",
|
||||
} as SerializableDistributedRenderConfig;
|
||||
}
|
||||
|
||||
describe("validateDistributedRenderConfig", () => {
|
||||
it("accepts a minimal valid config", () => {
|
||||
expect(validateDistributedRenderConfig(base())).toBeDefined();
|
||||
});
|
||||
|
||||
it("rejects a bad fps", () => {
|
||||
expect(() => validateDistributedRenderConfig({ ...base(), fps: 25 } as never)).toThrow(
|
||||
/config\.fps/,
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects odd dimensions (yuv420p)", () => {
|
||||
expect(() => validateDistributedRenderConfig({ ...base(), width: 1921 })).toThrow(/even/);
|
||||
});
|
||||
|
||||
it("rejects an out-of-range dimension", () => {
|
||||
expect(() => validateDistributedRenderConfig({ ...base(), height: 8 })).toThrow(/\[16, 7680\]/);
|
||||
});
|
||||
|
||||
it("rejects an unknown format", () => {
|
||||
expect(() => validateDistributedRenderConfig({ ...base(), format: "gif" as never })).toThrow(
|
||||
/config\.format/,
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects codec with a non-mp4 format", () => {
|
||||
expect(() =>
|
||||
validateDistributedRenderConfig({ ...base(), format: "webm", codec: "h264" } as never),
|
||||
).toThrow(/only valid with format="mp4"/);
|
||||
});
|
||||
|
||||
it("rejects crf + bitrate together", () => {
|
||||
expect(() =>
|
||||
validateDistributedRenderConfig({ ...base(), crf: 20, bitrate: "10M" } as never),
|
||||
).toThrow(/mutually exclusive/);
|
||||
});
|
||||
|
||||
it("rejects force-hdr", () => {
|
||||
expect(() =>
|
||||
validateDistributedRenderConfig({ ...base(), hdrMode: "force-hdr" as never }),
|
||||
).toThrow(/force-sdr/);
|
||||
});
|
||||
|
||||
it("rejects an over-cap chunkSize", () => {
|
||||
expect(() => validateDistributedRenderConfig({ ...base(), chunkSize: 5000 } as never)).toThrow(
|
||||
/<= 3600/,
|
||||
);
|
||||
});
|
||||
|
||||
it("throws InvalidConfigError with a field pointer", () => {
|
||||
try {
|
||||
validateDistributedRenderConfig({ ...base(), fps: 1 } as never);
|
||||
throw new Error("should have thrown");
|
||||
} catch (err) {
|
||||
expect(err).toBeInstanceOf(InvalidConfigError);
|
||||
expect((err as InvalidConfigError).field).toBe("config.fps");
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("validateVariablesPayload", () => {
|
||||
it("accepts a plain JSON object", () => {
|
||||
expect(() =>
|
||||
validateVariablesPayload({ title: "Hi", count: 3, nested: { ok: true } }),
|
||||
).not.toThrow();
|
||||
});
|
||||
|
||||
it("rejects undefined leaves", () => {
|
||||
expect(() => validateVariablesPayload({ a: undefined })).toThrow(/undefined leaves/);
|
||||
});
|
||||
|
||||
it("rejects a top-level array", () => {
|
||||
expect(() => validateVariablesPayload([1, 2])).toThrow(/plain JSON object/);
|
||||
});
|
||||
|
||||
it("rejects NaN", () => {
|
||||
expect(() => validateVariablesPayload({ x: NaN })).toThrow(/non-finite/);
|
||||
});
|
||||
|
||||
it("rejects a Date (non-plain object)", () => {
|
||||
expect(() => validateVariablesPayload({ when: new Date(0) })).toThrow(/non-plain objects/);
|
||||
});
|
||||
});
|
||||
|
||||
describe("validateWorkflowsInputSize", () => {
|
||||
it("accepts a small payload", () => {
|
||||
expect(() => validateWorkflowsInputSize({ a: "b" })).not.toThrow();
|
||||
});
|
||||
|
||||
it("rejects a payload over the 512 KiB cap", () => {
|
||||
const big = { blob: "x".repeat(MAX_WORKFLOWS_INPUT_BYTES + 1) };
|
||||
expect(() => validateWorkflowsInputSize(big)).toThrow(/512 KiB/);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,77 @@
|
||||
/**
|
||||
* Client-side validation for the Cloud Run adapter.
|
||||
*
|
||||
* The cloud-agnostic config-shape validation (`validateDistributedRenderConfig`,
|
||||
* `validateVariablesPayload`, `InvalidConfigError`) lives in
|
||||
* `@hyperframes/producer/distributed` and is shared with the other adapters.
|
||||
* This module re-exports those and adds the one piece that is specific to
|
||||
* Cloud Workflows: the 512 KiB execution-argument size cap.
|
||||
*/
|
||||
|
||||
import { InvalidConfigError } from "@hyperframes/producer/distributed";
|
||||
|
||||
export {
|
||||
InvalidConfigError,
|
||||
validateDistributedRenderConfig,
|
||||
validateVariablesPayload,
|
||||
} from "@hyperframes/producer/distributed";
|
||||
|
||||
/**
|
||||
* Hard cap on Cloud Workflows execution arguments — 512 KiB per the Workflows
|
||||
* quotas page (maximum size of arguments passed when an execution starts).
|
||||
* The cap is on the entire serialized argument, not just the variables,
|
||||
* because users hit it at the wire boundary regardless of which field caused
|
||||
* the bloat.
|
||||
*
|
||||
* Specific to Cloud Workflows. Other runtimes (Lambda + Step Functions,
|
||||
* Temporal) have different caps; don't reuse this constant for those without
|
||||
* confirming the limit.
|
||||
*/
|
||||
export const MAX_WORKFLOWS_INPUT_BYTES = 512 * 1024;
|
||||
|
||||
/** Pointer to the docs section that explains the URL-your-assets convention. */
|
||||
const LARGE_VARIABLES_DOCS_URL =
|
||||
"https://hyperframes.heygen.com/deploy/templates-on-lambda#working-with-large-variables";
|
||||
|
||||
/**
|
||||
* Validate that the serialized Cloud Workflows execution argument fits inside
|
||||
* the 512 KiB cap. Measured in UTF-8 bytes (the format the API uses on the
|
||||
* wire) — JS strings count UTF-16 code units, which under-reports for any
|
||||
* multi-byte character.
|
||||
*
|
||||
* Throws {@link InvalidConfigError} with a clear message naming the actual
|
||||
* byte count, the cap, and a pointer to the "working with large variables"
|
||||
* docs section, so users hit the limit at the SDK boundary with actionable
|
||||
* guidance instead of as an opaque argument-too-large error after the
|
||||
* execution starts.
|
||||
*/
|
||||
// fallow-ignore-next-line complexity
|
||||
export function validateWorkflowsInputSize(input: unknown): void {
|
||||
let serialized: string | undefined;
|
||||
try {
|
||||
serialized = JSON.stringify(input);
|
||||
} catch (err) {
|
||||
throw new InvalidConfigError(
|
||||
"config",
|
||||
`Cloud Workflows execution argument is not JSON-serializable: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
}
|
||||
if (serialized === undefined) {
|
||||
throw new InvalidConfigError(
|
||||
"config",
|
||||
"Cloud Workflows execution argument is not JSON-serializable (JSON.stringify returned undefined). " +
|
||||
"Check that all fields, including config.variables, are plain JSON values.",
|
||||
);
|
||||
}
|
||||
const byteLength = Buffer.byteLength(serialized, "utf8");
|
||||
if (byteLength > MAX_WORKFLOWS_INPUT_BYTES) {
|
||||
throw new InvalidConfigError(
|
||||
"config",
|
||||
`Cloud Workflows execution argument is ${byteLength} bytes, which exceeds the ` +
|
||||
`${MAX_WORKFLOWS_INPUT_BYTES}-byte (512 KiB) limit. Variables are for typed data ` +
|
||||
`(strings, numbers, structured records); media assets (images, audio, video) should ` +
|
||||
`be passed as URL references the composition resolves at render time, not inlined as ` +
|
||||
`base64. See ${LARGE_VARIABLES_DOCS_URL} for the URL-your-assets convention.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user