diff --git a/.fallowrc.jsonc b/.fallowrc.jsonc index 41ba72dbed..8982222c9b 100644 --- a/.fallowrc.jsonc +++ b/.fallowrc.jsonc @@ -4,6 +4,7 @@ "packages/producer/src/**/*.test.ts", "packages/aws-lambda/src/**/*.test.ts", "packages/gcp-cloud-run/src/**/*.test.ts", + "packages/gcp-cloud-run/terraform/*.test.ts", "packages/producer/src/regression-harness.ts", "packages/producer/src/regression-harness-distributed.test.ts", "packages/producer/src/regression-harness-lambda-local.ts", @@ -403,6 +404,12 @@ // extracting a shared cloud abstraction would couple independent packages. "packages/aws-lambda/src/handler.ts", "packages/aws-lambda/src/s3Transport.ts", + // The GCP handler deliberately mirrors the AWS protocol lifecycle while + // retaining provider-specific GCS, HTTP, and Cloud Workflows semantics. + // Its tests also mirror the same wire-contract cases; a cross-provider + // test abstraction would hide the adapter boundary being asserted. + "packages/gcp-cloud-run/src/server.ts", + "packages/gcp-cloud-run/src/server.test.ts", // sourcePatcher.ts: pre-existing internal clones between the inline-style // and attribute tag-patchers; only the PatchOperation type gained two // optional fields here, but the line shift makes fallow re-flag them. diff --git a/bun.lock b/bun.lock index 9643719f33..a3a23fc379 100644 --- a/bun.lock +++ b/bun.lock @@ -23,7 +23,7 @@ }, "packages/aws-lambda": { "name": "@hyperframes/aws-lambda", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@aws-sdk/client-s3": "^3.700.0", "@aws-sdk/client-sfn": "^3.700.0", @@ -56,7 +56,7 @@ }, "packages/cli": { "name": "@hyperframes/cli", - "version": "0.7.60", + "version": "0.7.71", "bin": { "hyperframes": "./bin/hyperframes.mjs", }, @@ -107,7 +107,7 @@ }, "packages/core": { "name": "@hyperframes/core", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@chenglou/pretext": "^0.0.5", "@hyperframes/lint": "workspace:*", @@ -132,7 +132,7 @@ }, "packages/engine": { "name": "@hyperframes/engine", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@hono/node-server": "^1.13.0", "@hyperframes/core": "workspace:^", @@ -151,7 +151,7 @@ }, "packages/gcp-cloud-run": { "name": "@hyperframes/gcp-cloud-run", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@google-cloud/storage": "^7.14.0", "@google-cloud/workflows": "^4.2.0", @@ -167,11 +167,12 @@ "esbuild": "^0.25.12", "tsx": "^4.21.0", "typescript": "^5.7.2", + "yaml": "^2.9.0", }, }, "packages/lint": { "name": "@hyperframes/lint", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@hyperframes/parsers": "workspace:*", "htmlparser2": "^10.1.0", @@ -189,7 +190,7 @@ }, "packages/parsers": { "name": "@hyperframes/parsers", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@babel/parser": "^7.27.0", "acorn": "^8.17.0", @@ -209,7 +210,7 @@ }, "packages/player": { "name": "@hyperframes/player", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@hyperframes/core": "workspace:*", }, @@ -224,7 +225,7 @@ }, "packages/producer": { "name": "@hyperframes/producer", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@fontsource/archivo-black": "^5.2.8", "@fontsource/eb-garamond": "^5.2.7", @@ -269,7 +270,7 @@ }, "packages/sdk": { "name": "@hyperframes/sdk", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@hyperframes/core": "workspace:*", "@hyperframes/parsers": "workspace:*", @@ -299,7 +300,7 @@ }, "packages/shader-transitions": { "name": "@hyperframes/shader-transitions", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "html2canvas": "^1.4.1", }, @@ -311,7 +312,7 @@ }, "packages/studio": { "name": "@hyperframes/studio", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@codemirror/autocomplete": "^6.20.1", "@codemirror/commands": "^6.10.3", @@ -359,7 +360,7 @@ }, "packages/studio-server": { "name": "@hyperframes/studio-server", - "version": "0.7.60", + "version": "0.7.71", "dependencies": { "@hyperframes/core": "workspace:*", "@hyperframes/parsers": "workspace:*", diff --git a/examples/gcp-cloud-run/README.md b/examples/gcp-cloud-run/README.md index f7be4eda93..62732f9928 100644 --- a/examples/gcp-cloud-run/README.md +++ b/examples/gcp-cloud-run/README.md @@ -1,51 +1,136 @@ # Google Cloud Run example -End-to-end deployment + smoke for [`@hyperframes/gcp-cloud-run`](../../packages/gcp-cloud-run) — the Cloud Run + Cloud Workflows adapter for HyperFrames distributed rendering. +End-to-end deployment and parity testing for +[`@hyperframes/gcp-cloud-run`](../../packages/gcp-cloud-run), the Cloud Run + +Cloud Workflows adapter for HyperFrames distributed rendering. ## Layout +```text +scripts/smoke.sh Owner-isolated real-GCP deploy, render, parity, cleanup +sample-events/ v1 and v2 handler request examples ``` -scripts/smoke.sh Real-GCP smoke: build → deploy → render → PSNR → destroy -sample-events/ Example request bodies for the Cloud Run handler - (plan.json, render-chunk.json, assemble.json) -``` -The Terraform module and the Cloud Workflows definition that the smoke deploys live with the package, at `packages/gcp-cloud-run/terraform/` (including `workflow.yaml`). +The Terraform module and Cloud Workflows definition live in +`packages/gcp-cloud-run/terraform/`. + +## Protocol rollout + +The workflow defaults to plan protocol v1 when `PlanProtocol` is absent. V2 is +accepted only when the caller explicitly sends `PlanProtocol: "v2"`. + +V1 and v2 use disjoint plan locators: + +- v1: `PlanGcsUri` +- v2: `PlanV2ManifestGcsUri` and `PlanV2ArtifactGcsPrefix` + +The workflow validates that the plan response matches the selected protocol +before starting chunk fan-out. It never silently falls back from v2 to v1. +Deploy the v2 workflow only with a Cloud Run image whose handler implements +the matching v2 request/response contract. An older v1-only handler will keep +serving default v1 requests, but explicit v2 smoke executions will fail closed. ## Prerequisites -- `gcloud` authenticated, with a project that has **billing enabled** -- `terraform` (≥ 1.5), `docker`, `ffmpeg`, `jq` on PATH +- `gcloud` authenticated to a project with billing enabled +- `terraform` (>= 1.5), `ffmpeg`, `ffprobe`, `jq`, `tar`, and `sha256sum` +- the required project APIs already enabled, plus permission to run Cloud + Build and manage Cloud Run, Workflows, GCS, IAM service accounts, + Monitoring, and Artifact Registry resources ## Run the smoke +V1 remains the safe default: + ```bash -# Renders the mp4-h264-sdr fixture through the workflow and PSNR-compares it -# against the in-process baseline, then tears the stack down. -./scripts/smoke.sh --project YOUR_GCP_PROJECT --region us-central1 +./scripts/smoke.sh \ + --project YOUR_GCP_PROJECT \ + --region us-central1 +``` -# Keep the stack up to poke at it: -./scripts/smoke.sh --project YOUR_GCP_PROJECT --keep-stack +Explicitly run v1/v2 end-to-end parity at one or more chunk sizes: -# Render at several chunk sizes to see the fan-out scaling: -./scripts/smoke.sh --project YOUR_GCP_PROJECT --chunk-sizes 30,15,10 +```bash +./scripts/smoke.sh \ + --project YOUR_GCP_PROJECT \ + --region us-central1 \ + --protocols v1,v2 \ + --chunk-sizes 30,15,10 \ + --owner plan-v2-parity ``` -Outputs land in `scripts/gcp-smoke-artifacts/`: `results.json` -(`chunkSize × wallClockMs × psnrAvgDb`), the rendered MP4s, and each -workflow execution's describe output. +For each chunk size, parity requires exact equality of: + +- decoded RGBA video frames +- decoded 48 kHz stereo PCM audio +- normalized `ffprobe` stream and duration metadata + +The encoded MP4 hash and byte count are recorded but are not the equality +oracle because mux metadata can differ without changing decoded output. +Each render is also PSNR-compared with the checked-in in-process fixture +baseline. + +## Isolation and cleanup + +Every invocation hashes the owner, project, region, and a fresh invocation +nonce into a unique resource prefix such as `hf-smoke-a1b2c3d4e5`. Reusing an +owner label does not reuse old Terraform state or cloud resources. This prefix +stays within GCP service account naming limits. The smoke: + +- never uses the static `hyperframes` prefix +- copies the Terraform module into an owner-scoped work directory and uses an + isolated Terraform data directory and state file +- scopes GCS keys, render outputs, the image package/tag, and the default + Artifact Registry repository to that owner +- deletes only an image it built +- deletes the Artifact Registry repository only when that invocation created it +- refuses to enable project APIs, because APIs are shared project state +- stages the bounded Cloud Build source archive in an owner-scoped bucket, + writes build logs to Cloud Logging, and deletes the staging bucket + +Cleanup is on by default. It empties and destroys the owner-scoped bucket and +stack, deletes owned image/repository/build-staging resources, then verifies +the Cloud Run service, workflow, buckets, both service accounts, image, and any +test-created repository are absent. Cleanup fails on API or authentication +errors rather than interpreting them as successful deletion. GCP retains the +Cloud Build execution record and Cloud Logging audit entries as project-level +operational history; the smoke test does not attempt to erase audit records. + +`--keep-stack` deliberately retains the stack, image, and repository and +prints the exact isolated state directory and Terraform cleanup commands. +Never use it for unattended CI. + +Evidence lands under: + +```text +scripts/gcp-smoke-artifacts// + results.json + parity.json + renders/ + terraform/ + terraform-data/ +``` + +Use `--image` to test a caller-owned existing image. That image is never +deleted. `--skip-build` requires `--image`; new invocations never inherit an +old invocation's state or image implicitly. ## Test the handler locally -The sample events exercise the same body shape Cloud Workflows sends. With the -container running locally (`PORT=8080`) and credentials that can reach a GCS -bucket, you can drive a single action: +The sample events mirror the request bodies sent by Cloud Workflows: ```bash +# V1 curl -sX POST localhost:8080/ \ -H 'content-type: application/json' \ --data @sample-events/plan.json | jq . + +# Explicit v2 +curl -sX POST localhost:8080/ \ + -H 'content-type: application/json' \ + --data @sample-events/plan-v2.json | jq . ``` -Replace the `PROJECT` placeholder bucket names and `REPLACE_WITH_PLAN_HASH` -with real values from a prior `plan` response. +Replace `PROJECT`, locator placeholders, and plan hashes with values returned +by the preceding plan action. A complete action sequence is +`plan → renderChunk(s) → assemble`. diff --git a/examples/gcp-cloud-run/sample-events/assemble-v2.json b/examples/gcp-cloud-run/sample-events/assemble-v2.json new file mode 100644 index 0000000000..23966b50dc --- /dev/null +++ b/examples/gcp-cloud-run/sample-events/assemble-v2.json @@ -0,0 +1,14 @@ +{ + "Action": "assemble", + "PlanProtocol": "v2", + "PlanV2ManifestGcsUri": "gs://hyperframes-render-PROJECT/renders/hf-render-demo/v2/manifest.json", + "PlanV2ArtifactGcsPrefix": "gs://hyperframes-render-PROJECT/renders/hf-render-demo/v2/artifacts/sha256", + "PlanHash": "REPLACE_WITH_PLAN_HASH", + "ChunkGcsUris": [ + "gs://hyperframes-render-PROJECT/renders/hf-render-demo/chunks/0000.mp4", + "gs://hyperframes-render-PROJECT/renders/hf-render-demo/chunks/0001.mp4" + ], + "AudioGcsUri": null, + "OutputGcsUri": "gs://hyperframes-render-PROJECT/renders/hf-render-demo/output.mp4", + "Format": "mp4" +} diff --git a/examples/gcp-cloud-run/sample-events/assemble.json b/examples/gcp-cloud-run/sample-events/assemble.json index 115a538bdb..4be98aa21d 100644 --- a/examples/gcp-cloud-run/sample-events/assemble.json +++ b/examples/gcp-cloud-run/sample-events/assemble.json @@ -1,5 +1,6 @@ { "Action": "assemble", + "PlanProtocol": "v1", "PlanGcsUri": "gs://hyperframes-render-PROJECT/renders/hf-render-demo/plan.tar.gz", "ChunkGcsUris": [ "gs://hyperframes-render-PROJECT/renders/hf-render-demo/chunks/0000.mp4", diff --git a/examples/gcp-cloud-run/sample-events/plan-v2.json b/examples/gcp-cloud-run/sample-events/plan-v2.json new file mode 100644 index 0000000000..d17c345e9b --- /dev/null +++ b/examples/gcp-cloud-run/sample-events/plan-v2.json @@ -0,0 +1,7 @@ +{ + "Action": "plan", + "PlanProtocol": "v2", + "ProjectGcsUri": "gs://hyperframes-render-PROJECT/sites/abc123/project.tar.gz", + "PlanOutputGcsPrefix": "gs://hyperframes-render-PROJECT/renders/hf-render-demo/", + "Config": { "fps": 30, "width": 1920, "height": 1080, "format": "mp4" } +} diff --git a/examples/gcp-cloud-run/sample-events/plan.json b/examples/gcp-cloud-run/sample-events/plan.json index 667988b413..6f0856668f 100644 --- a/examples/gcp-cloud-run/sample-events/plan.json +++ b/examples/gcp-cloud-run/sample-events/plan.json @@ -1,5 +1,6 @@ { "Action": "plan", + "PlanProtocol": "v1", "ProjectGcsUri": "gs://hyperframes-render-PROJECT/sites/abc123/project.tar.gz", "PlanOutputGcsPrefix": "gs://hyperframes-render-PROJECT/renders/hf-render-demo/", "Config": { "fps": 30, "width": 1920, "height": 1080, "format": "mp4" } diff --git a/examples/gcp-cloud-run/sample-events/render-chunk-v2.json b/examples/gcp-cloud-run/sample-events/render-chunk-v2.json new file mode 100644 index 0000000000..63d4bc21e7 --- /dev/null +++ b/examples/gcp-cloud-run/sample-events/render-chunk-v2.json @@ -0,0 +1,10 @@ +{ + "Action": "renderChunk", + "PlanProtocol": "v2", + "PlanV2ManifestGcsUri": "gs://hyperframes-render-PROJECT/renders/hf-render-demo/v2/manifest.json", + "PlanV2ArtifactGcsPrefix": "gs://hyperframes-render-PROJECT/renders/hf-render-demo/v2/artifacts/sha256", + "PlanHash": "REPLACE_WITH_PLAN_HASH", + "ChunkIndex": 0, + "ChunkOutputGcsPrefix": "gs://hyperframes-render-PROJECT/renders/hf-render-demo/", + "Format": "mp4" +} diff --git a/examples/gcp-cloud-run/sample-events/render-chunk.json b/examples/gcp-cloud-run/sample-events/render-chunk.json index b98ad63adc..c11bd441ae 100644 --- a/examples/gcp-cloud-run/sample-events/render-chunk.json +++ b/examples/gcp-cloud-run/sample-events/render-chunk.json @@ -1,5 +1,6 @@ { "Action": "renderChunk", + "PlanProtocol": "v1", "PlanGcsUri": "gs://hyperframes-render-PROJECT/renders/hf-render-demo/plan.tar.gz", "PlanHash": "REPLACE_WITH_PLAN_HASH", "ChunkIndex": 0, diff --git a/examples/gcp-cloud-run/scripts/smoke.sh b/examples/gcp-cloud-run/scripts/smoke.sh index 4cd4dbf1db..ff14309eb1 100755 --- a/examples/gcp-cloud-run/scripts/smoke.sh +++ b/examples/gcp-cloud-run/scripts/smoke.sh @@ -1,239 +1,671 @@ #!/usr/bin/env bash -# Real-GCP smoke + benchmark for the HyperFrames Cloud Run adapter. +# Owner-isolated real-GCP smoke + v1/v2 parity test for the HyperFrames +# Cloud Run adapter. # -# Run from a workstation with `gcloud` credentials. Builds the render -# container, pushes it to Artifact Registry, applies the Terraform module at -# packages/gcp-cloud-run/terraform to your project, renders a fixture -# composition through the Cloud Workflows definition, PSNR-compares the -# output against the in-process baseline, and tears the stack down. +# The default is intentionally v1-only. Plan protocol v2 must be opted into +# explicitly with --protocols v1,v2. Every invocation derives a unique, +# length-safe resource prefix and uses an isolated Terraform working directory +# and state file. Cleanup verifies every owned resource is absent and fails +# closed on API/authentication errors. # # Usage: # ./smoke.sh --project -# ./smoke.sh --project p --fixture mp4-h264-sdr --chunk-sizes 15,30 -# ./smoke.sh --project p --keep-stack +# ./smoke.sh --project p --protocols v1,v2 --chunk-sizes 15,30 +# ./smoke.sh --project p --owner james-plan-v2 --keep-stack # -# Required tools on PATH: -# - gcloud (authenticated; the target project must have billing enabled) -# - terraform (>= 1.5) -# - docker -# - ffmpeg (PSNR computation) -# - jq -# -# Inputs (flags or env vars): -# --project (required; or $GCP_PROJECT) -# --region (default: us-central1) -# --fixture (default: mp4-h264-sdr — under packages/producer/tests/distributed/) -# --chunk-sizes (default: from the fixture meta; CSV of chunkSize overrides) -# --psnr-threshold (default: 35) -# --repo (Artifact Registry repo name, default: hyperframes) -# --keep-stack (skip `terraform destroy` at the end) -# --skip-build (reuse the last-pushed image tag in ./gcp-smoke-artifacts/image.txt) -# -# Outputs: -# ./gcp-smoke-artifacts/results.json (chunkSize x wallClockMs x psnrAvgDb) -# ./gcp-smoke-artifacts/renders/c-output.mp4 -# ./gcp-smoke-artifacts/renders/c-execution.json +# Required tools: +# gcloud, terraform (>= 1.5), ffmpeg, ffprobe, jq, tar, sha256sum # # Exit codes: -# 0 all good 1 arg/pre-flight 2 build/push 3 terraform apply -# 4 a render failed 5 PSNR below threshold +# 0 success 1 arguments/pre-flight +# 2 image build/push failed 3 terraform apply failed +# 4 render failed 5 baseline PSNR failed +# 6 v1/v2 parity failed 7 cleanup or cleanup verification failed set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" REPO_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)" -TF_DIR="$REPO_ROOT/packages/gcp-cloud-run/terraform" +TF_SOURCE_DIR="$REPO_ROOT/packages/gcp-cloud-run/terraform" -# ── Defaults ────────────────────────────────────────────────────────────── PROJECT="${GCP_PROJECT:-}" REGION="${GCP_REGION:-us-central1}" FIXTURE="${FIXTURE:-mp4-h264-sdr}" CHUNK_SIZES="${CHUNK_SIZES:-}" PSNR_THRESHOLD="${PSNR_THRESHOLD:-35}" -AR_REPO="${AR_REPO:-hyperframes}" +PROTOCOLS="${PROTOCOLS:-v1}" +OWNER="${HYPERFRAMES_SMOKE_OWNER:-}" +AR_REPO="${AR_REPO:-}" +AR_REPO_WAS_EXPLICIT=0 +[ -z "$AR_REPO" ] || AR_REPO_WAS_EXPLICIT=1 +EXPLICIT_IMAGE="${HYPERFRAMES_GCP_IMAGE:-}" KEEP_STACK=0 SKIP_BUILD=0 while [ $# -gt 0 ]; do case "$1" in - --project) PROJECT="$2"; shift 2 ;; - --region) REGION="$2"; shift 2 ;; - --fixture) FIXTURE="$2"; shift 2 ;; - --chunk-sizes) CHUNK_SIZES="$2"; shift 2 ;; - --psnr-threshold) PSNR_THRESHOLD="$2"; shift 2 ;; - --repo) AR_REPO="$2"; shift 2 ;; - --keep-stack) KEEP_STACK=1; shift ;; - --skip-build) SKIP_BUILD=1; shift ;; - -h|--help) sed -n '2,40p' "$0"; exit 0 ;; - *) echo "Unknown arg: $1" >&2; exit 1 ;; + --project) PROJECT="$2"; shift 2 ;; + --region) REGION="$2"; shift 2 ;; + --fixture) FIXTURE="$2"; shift 2 ;; + --chunk-sizes) CHUNK_SIZES="$2"; shift 2 ;; + --psnr-threshold) PSNR_THRESHOLD="$2"; shift 2 ;; + --protocols) PROTOCOLS="$2"; shift 2 ;; + --owner) OWNER="$2"; shift 2 ;; + --repo) AR_REPO="$2"; AR_REPO_WAS_EXPLICIT=1; shift 2 ;; + --image) EXPLICIT_IMAGE="$2"; shift 2 ;; + --keep-stack) KEEP_STACK=1; shift ;; + --skip-build) SKIP_BUILD=1; shift ;; + -h|--help) sed -n '2,34p' "$0"; exit 0 ;; + *) echo "ERROR: unknown argument: $1" >&2; exit 1 ;; + esac +done + +[ -n "$PROJECT" ] || { + echo "ERROR: --project (or GCP_PROJECT) is required" >&2 + exit 1 +} + +for tool in gcloud terraform ffmpeg ffprobe jq tar sha256sum; do + command -v "$tool" >/dev/null || { + echo "ERROR: $tool is not available on PATH" >&2 + exit 1 + } +done + +PROTOCOLS="${PROTOCOLS//[[:space:]]/}" +case ",$PROTOCOLS," in + *,v1,*|*,v2,*) ;; + *) echo "ERROR: --protocols must contain v1 and/or v2" >&2; exit 1 ;; +esac +IFS=',' read -ra PROTOCOL_LIST <<< "$PROTOCOLS" +declare -A SEEN_PROTOCOLS=() +for protocol in "${PROTOCOL_LIST[@]}"; do + case "$protocol" in + v1|v2) ;; + *) echo "ERROR: unsupported protocol '$protocol'; expected v1 or v2" >&2; exit 1 ;; esac + [ -z "${SEEN_PROTOCOLS[$protocol]:-}" ] || { + echo "ERROR: duplicate protocol '$protocol'" >&2 + exit 1 + } + SEEN_PROTOCOLS[$protocol]=1 done -[ -n "$PROJECT" ] || { echo "ERROR: --project (or \$GCP_PROJECT) is required" >&2; exit 1; } -for tool in gcloud terraform docker ffmpeg jq; do - command -v "$tool" >/dev/null || { echo "ERROR: $tool not on PATH" >&2; exit 1; } +if [ -z "$OWNER" ]; then + OWNER="$(id -un)-$(date -u +%Y%m%dT%H%M%SZ)-$$" +fi +# Include an invocation nonce even when the human-readable owner is reused. +# This prevents a new run from ever inheriting an old run's Terraform state or +# colliding with old resources. +RUN_NONCE="$(date -u +%Y%m%dT%H%M%S)-$$-$RANDOM" +OWNER_HASH="$(printf '%s' "$OWNER:$PROJECT:$REGION:$RUN_NONCE" | sha256sum | cut -c1-10)" +# 19 chars. With the longest "-run"/"-wf" suffix, service-account IDs stay +# comfortably under GCP's 30-character limit. +STACK_NAME="hf-smoke-$OWNER_HASH" +[ "$STACK_NAME" != "hyperframes" ] || { + echo "ERROR: refusing to use the shared static resource prefix" >&2 + exit 1 +} +[ -n "$AR_REPO" ] || AR_REPO="$STACK_NAME" + +ARTIFACT_ROOT="$SCRIPT_DIR/gcp-smoke-artifacts" +ARTIFACT_DIR="$ARTIFACT_ROOT/$OWNER_HASH" +RENDER_DIR="$ARTIFACT_DIR/renders" +TF_WORK_DIR="$ARTIFACT_DIR/terraform" +TF_DATA_DIR="$ARTIFACT_DIR/terraform-data" +mkdir -p "$RENDER_DIR" "$TF_WORK_DIR" "$TF_DATA_DIR" +export TF_DATA_DIR + +for module_file in "$TF_SOURCE_DIR"/*.tf "$TF_SOURCE_DIR/workflow.yaml"; do + cp "$module_file" "$TF_WORK_DIR/" done FIXTURE_DIR="$REPO_ROOT/packages/producer/tests/distributed/$FIXTURE" FIXTURE_META="$FIXTURE_DIR/meta.json" BASELINE_MP4="$FIXTURE_DIR/output/output.mp4" -[ -d "$FIXTURE_DIR/src" ] || { echo "ERROR: fixture src missing: $FIXTURE_DIR/src" >&2; exit 1; } -[ -f "$BASELINE_MP4" ] || { echo "ERROR: baseline mp4 missing: $BASELINE_MP4" >&2; exit 1; } +[ -d "$FIXTURE_DIR/src" ] || { + echo "ERROR: fixture source missing: $FIXTURE_DIR/src" >&2 + exit 1 +} +[ -f "$FIXTURE_META" ] || { + echo "ERROR: fixture metadata missing: $FIXTURE_META" >&2 + exit 1 +} +[ -f "$BASELINE_MP4" ] || { + echo "ERROR: baseline video missing: $BASELINE_MP4" >&2 + exit 1 +} + +BUCKET="$STACK_NAME-render-$PROJECT" +SERVICE_NAME="$STACK_NAME-render" +WORKFLOW_NAME="$STACK_NAME-render" +RUN_SA="$STACK_NAME-run@$PROJECT.iam.gserviceaccount.com" +WORKFLOW_SA="$STACK_NAME-wf@$PROJECT.iam.gserviceaccount.com" +IMAGE="" +IMAGE_PACKAGE="" +CREATED_IMAGE=0 +CREATED_REPO=0 +CREATED_BUILD_BUCKET=0 +STACK_APPLIED=0 +BUILD_BUCKET="$STACK_NAME-build-$PROJECT" + +is_not_found() { + grep -Eqi 'NOT_FOUND|not found|does not exist|was not found|could not be found|cannot find' "$1" +} + +verify_absent() { + local label="$1" + shift + local error_file="$ARTIFACT_DIR/cleanup-${label//[^a-zA-Z0-9]/-}.stderr" + if "$@" >/dev/null 2>"$error_file"; then + echo " ✗ $label still exists" >&2 + return 1 + fi + if is_not_found "$error_file"; then + echo " ✓ $label absent" + return 0 + fi + echo " ✗ could not verify $label absence (API/auth error):" >&2 + sed -n '1,12p' "$error_file" >&2 + return 1 +} + +verify_service_account_absent() { + local label="$1" + local email="$2" + local error_file="$ARTIFACT_DIR/cleanup-${label//[^a-zA-Z0-9]/-}.stderr" + local matches + if ! matches="$(gcloud iam service-accounts list \ + --project "$PROJECT" \ + --filter "email:$email" \ + --format 'value(email)' 2>"$error_file")"; then + echo " ✗ could not verify $label absence (API/auth error):" >&2 + sed -n '1,12p' "$error_file" >&2 + return 1 + fi + if [ -n "$matches" ]; then + echo " ✗ $label still exists" >&2 + return 1 + fi + echo " ✓ $label absent" +} + +delete_owned_image() { + local error_file="$ARTIFACT_DIR/cleanup-image-delete.stderr" + [ "$CREATED_IMAGE" -eq 1 ] || return 0 + echo "→ Deleting owner-scoped test image package $IMAGE_PACKAGE" + if gcloud artifacts docker images delete "$IMAGE_PACKAGE" --delete-tags --quiet \ + --project "$PROJECT" >/dev/null 2>"$error_file"; then + return 0 + fi + if is_not_found "$error_file"; then + return 0 + fi + sed -n '1,12p' "$error_file" >&2 + return 1 +} + +cleanup() { + local original_rc=$? + local cleanup_rc=0 + trap - EXIT + set +e + + if [ "$KEEP_STACK" -eq 1 ]; then + echo "→ --keep-stack set; retaining this invocation's stack, image, and repository." + echo " Owner: $OWNER ($OWNER_HASH)" + echo " Isolated Terraform state: $TF_WORK_DIR/terraform.tfstate" + echo " Destroy with TF_DATA_DIR=$TF_DATA_DIR terraform -chdir=$TF_WORK_DIR apply -auto-approve -var project_id=$PROJECT -var region=$REGION -var project_name=$STACK_NAME -var image=$IMAGE -var bucket_force_destroy=true" + echo " Then run the matching terraform destroy command with the same variables." + exit "$original_rc" + fi + + if [ "$STACK_APPLIED" -eq 1 ]; then + echo "→ Destroying owner-scoped Terraform stack $STACK_NAME" + # Update force_destroy only when this isolated state already owns the + # bucket. A partial apply that failed before bucket creation must not make + # cleanup create a new bucket merely to destroy it. + if terraform -chdir="$TF_WORK_DIR" state show google_storage_bucket.render \ + >/dev/null 2>&1; then + terraform -chdir="$TF_WORK_DIR" apply -input=false -auto-approve \ + -target=google_storage_bucket.render \ + -var "project_id=$PROJECT" \ + -var "region=$REGION" \ + -var "project_name=$STACK_NAME" \ + -var "image=$IMAGE" \ + -var "bucket_force_destroy=true" >/dev/null + [ $? -eq 0 ] || cleanup_rc=1 + fi + terraform -chdir="$TF_WORK_DIR" destroy -input=false -auto-approve \ + -var "project_id=$PROJECT" \ + -var "region=$REGION" \ + -var "project_name=$STACK_NAME" \ + -var "image=$IMAGE" \ + -var "bucket_force_destroy=true" + [ $? -eq 0 ] || cleanup_rc=1 + + verify_absent "cloud-run-service" \ + gcloud run services describe "$SERVICE_NAME" --region "$REGION" --project "$PROJECT" + [ $? -eq 0 ] || cleanup_rc=1 + verify_absent "workflow" \ + gcloud workflows describe "$WORKFLOW_NAME" --location "$REGION" --project "$PROJECT" + [ $? -eq 0 ] || cleanup_rc=1 + verify_absent "render-bucket" \ + gcloud storage buckets describe "gs://$BUCKET" --project "$PROJECT" + [ $? -eq 0 ] || cleanup_rc=1 + verify_service_account_absent "run-service-account" "$RUN_SA" + [ $? -eq 0 ] || cleanup_rc=1 + verify_service_account_absent "workflow-service-account" "$WORKFLOW_SA" + [ $? -eq 0 ] || cleanup_rc=1 + fi + + delete_owned_image + [ $? -eq 0 ] || cleanup_rc=1 + if [ "$CREATED_IMAGE" -eq 1 ]; then + verify_absent "artifact-image" \ + gcloud artifacts docker images describe "$IMAGE_PACKAGE" --project "$PROJECT" + [ $? -eq 0 ] || cleanup_rc=1 + fi -ARTIFACT_DIR="$SCRIPT_DIR/gcp-smoke-artifacts" -mkdir -p "$ARTIFACT_DIR/renders" + if [ "$CREATED_BUILD_BUCKET" -eq 1 ]; then + echo "→ Deleting owner-scoped Cloud Build staging bucket $BUILD_BUCKET" + gcloud storage rm --recursive "gs://$BUILD_BUCKET" --project "$PROJECT" >/dev/null + [ $? -eq 0 ] || cleanup_rc=1 + verify_absent "cloud-build-staging-bucket" \ + gcloud storage buckets describe "gs://$BUILD_BUCKET" --project "$PROJECT" + [ $? -eq 0 ] || cleanup_rc=1 + fi + + if [ "$CREATED_REPO" -eq 1 ]; then + echo "→ Deleting test-created Artifact Registry repository $AR_REPO" + gcloud artifacts repositories delete "$AR_REPO" --location "$REGION" \ + --project "$PROJECT" --quiet + [ $? -eq 0 ] || cleanup_rc=1 + verify_absent "artifact-repository" \ + gcloud artifacts repositories describe "$AR_REPO" --location "$REGION" --project "$PROJECT" + [ $? -eq 0 ] || cleanup_rc=1 + fi + + if [ "$cleanup_rc" -ne 0 ]; then + echo "ERROR: cleanup or cleanup verification failed; see $ARTIFACT_DIR/cleanup-*.stderr" >&2 + exit 7 + fi + exit "$original_rc" +} +trap cleanup EXIT -echo "→ Project: $PROJECT Region: $REGION Fixture: $FIXTURE" +echo "→ Project: $PROJECT" +echo " Region: $REGION" +echo " Owner: $OWNER ($OWNER_HASH)" +echo " Resource prefix: $STACK_NAME" +echo " Protocols: $PROTOCOLS" +echo " Isolated Terraform directory: $TF_WORK_DIR" -# ── 1. Enable APIs ────────────────────────────────────────────────────────── -echo "→ Enabling required APIs (idempotent)" -gcloud services enable \ - run.googleapis.com workflows.googleapis.com workflowexecutions.googleapis.com \ - artifactregistry.googleapis.com cloudbuild.googleapis.com monitoring.googleapis.com \ - --project "$PROJECT" >/dev/null +echo "→ Verifying required project APIs are already enabled" +for api in \ + run.googleapis.com \ + workflows.googleapis.com \ + workflowexecutions.googleapis.com \ + artifactregistry.googleapis.com \ + cloudbuild.googleapis.com \ + monitoring.googleapis.com; do + enabled_api="$(gcloud services list \ + --enabled \ + --project "$PROJECT" \ + --filter "config.name=$api" \ + --format 'value(config.name)')" || { + echo "ERROR: could not verify required API $api" >&2 + exit 1 + } + [ "$enabled_api" = "$api" ] || { + echo "ERROR: required API $api is not enabled; refusing to mutate project-shared API state" >&2 + exit 1 + } +done + +echo "→ Verifying the unique resource names are unused" +verify_absent "preflight-cloud-run-service" \ + gcloud run services describe "$SERVICE_NAME" --region "$REGION" --project "$PROJECT" || exit 1 +verify_absent "preflight-workflow" \ + gcloud workflows describe "$WORKFLOW_NAME" --location "$REGION" --project "$PROJECT" || exit 1 +verify_absent "preflight-render-bucket" \ + gcloud storage buckets describe "gs://$BUCKET" --project "$PROJECT" || exit 1 +verify_absent "preflight-run-service-account" \ + gcloud iam service-accounts describe "$RUN_SA" --project "$PROJECT" || exit 1 +verify_absent "preflight-workflow-service-account" \ + gcloud iam service-accounts describe "$WORKFLOW_SA" --project "$PROJECT" || exit 1 +verify_absent "preflight-cloud-build-staging-bucket" \ + gcloud storage buckets describe "gs://$BUILD_BUCKET" --project "$PROJECT" || exit 1 -# ── 2. Build + push the render image ──────────────────────────────────────── IMAGE_TXT="$ARTIFACT_DIR/image.txt" -if [ "$SKIP_BUILD" -eq 1 ] && [ -f "$IMAGE_TXT" ]; then - IMAGE="$(cat "$IMAGE_TXT")" - echo "→ Reusing image $IMAGE" +if [ -n "$EXPLICIT_IMAGE" ]; then + IMAGE="$EXPLICIT_IMAGE" + echo "→ Using caller-owned image $IMAGE" +elif [ "$SKIP_BUILD" -eq 1 ]; then + echo "ERROR: --skip-build requires --image; new runs never reuse prior owner state" >&2 + exit 1 else - gcloud artifacts repositories describe "$AR_REPO" --location "$REGION" --project "$PROJECT" >/dev/null 2>&1 || \ - gcloud artifacts repositories create "$AR_REPO" --repository-format docker \ - --location "$REGION" --project "$PROJECT" >/dev/null - TAG="$(date +%Y%m%d-%H%M%S)" - IMAGE="$REGION-docker.pkg.dev/$PROJECT/$AR_REPO/hyperframes-render:$TAG" - echo "→ Building + pushing $IMAGE via Cloud Build" - # The Dockerfile lives at packages/gcp-cloud-run/Dockerfile, not the repo - # root, so we drive the build with an inline cloudbuild config rather than - # `--tag` (which assumes a root Dockerfile). - CB_CONFIG="$ARTIFACT_DIR/cloudbuild.yaml" - cat > "$CB_CONFIG" </dev/null 2>"$REPO_ERROR"; then + if [ "$AR_REPO_WAS_EXPLICIT" -eq 0 ]; then + echo "ERROR: generated repository $AR_REPO already exists; refusing to reuse owner state" >&2 + exit 1 + fi + echo "→ Reusing caller-selected Artifact Registry repository $AR_REPO" + elif is_not_found "$REPO_ERROR"; then + echo "→ Creating owner-scoped Artifact Registry repository $AR_REPO" + gcloud artifacts repositories create "$AR_REPO" \ + --repository-format docker \ + --location "$REGION" \ + --project "$PROJECT" >/dev/null + CREATED_REPO=1 + else + echo "ERROR: repository lookup failed (not a NOT_FOUND response)" >&2 + sed -n '1,12p' "$REPO_ERROR" >&2 + exit 1 + fi + + IMAGE_PACKAGE="$REGION-docker.pkg.dev/$PROJECT/$AR_REPO/$STACK_NAME-render" + IMAGE="$IMAGE_PACKAGE:$OWNER_HASH" + verify_absent "preflight-artifact-image" \ + gcloud artifacts docker images describe "$IMAGE_PACKAGE" --project "$PROJECT" || exit 1 + CREATED_IMAGE=1 + BUILD_CONFIG="$ARTIFACT_DIR/cloudbuild.json" + BUILD_IGNORE_FILE="$ARTIFACT_DIR/gcloudignore" + cat > "$BUILD_IGNORE_FILE" <<'EOF' +** +!package.json +!bun.lock +!scripts/ +!scripts/package-subpaths.mjs +!packages/ +!packages/core/ +!packages/core/** +!packages/engine/ +!packages/engine/** +!packages/producer/ +!packages/producer/** +!packages/gcp-cloud-run/ +!packages/gcp-cloud-run/** +!packages/lint/ +!packages/lint/** +!packages/parsers/ +!packages/parsers/** +!packages/sdk/ +!packages/sdk/** +!packages/sdk-playground/ +!packages/sdk-playground/** +!packages/studio-server/ +!packages/studio-server/** +!packages/player/ +!packages/player/package.json +!packages/cli/ +!packages/cli/package.json +!packages/studio/ +!packages/studio/package.json +!packages/shader-transitions/ +!packages/shader-transitions/package.json +!packages/aws-lambda/ +!packages/aws-lambda/package.json +packages/**/node_modules/** +packages/**/dist/** +packages/**/coverage/** +packages/**/output/** +packages/**/tests/** +packages/**/*.test.ts +packages/**/*.test.tsx +packages/gcp-cloud-run/terraform/** EOF - gcloud builds submit "$REPO_ROOT" --project "$PROJECT" --config "$CB_CONFIG" \ - || { echo "ERROR: image build/push failed" >&2; exit 2; } - echo "$IMAGE" > "$IMAGE_TXT" + jq -n --arg image "$IMAGE" '{ + steps: [{ + name: "gcr.io/cloud-builders/docker", + args: ["build", "-f", "packages/gcp-cloud-run/Dockerfile", "-t", $image, "."] + }], + images: [$image], + timeout: "3600s", + options: { + machineType: "E2_HIGHCPU_8", + logging: "CLOUD_LOGGING_ONLY" + } + }' > "$BUILD_CONFIG" + echo "→ Creating owner-scoped Cloud Build staging bucket $BUILD_BUCKET" + gcloud storage buckets create "gs://$BUILD_BUCKET" \ + --location "$REGION" \ + --uniform-bucket-level-access \ + --project "$PROJECT" >/dev/null + CREATED_BUILD_BUCKET=1 + echo "→ Building and pushing owner-scoped image $IMAGE via Cloud Build" + gcloud builds submit "$REPO_ROOT" \ + --project "$PROJECT" \ + --config "$BUILD_CONFIG" \ + --ignore-file "$BUILD_IGNORE_FILE" \ + --gcs-source-staging-dir "gs://$BUILD_BUCKET/source" || { + echo "ERROR: image build/push failed" >&2 + exit 2 + } + printf '%s\n' "$IMAGE" > "$IMAGE_TXT" fi -# ── 3. terraform apply ────────────────────────────────────────────────────── -# The google provider authenticates via Application Default Credentials. If -# ADC isn't configured (common on a box set up with only `gcloud auth login`), -# fall back to a short-lived access token from the active gcloud account. +# Terraform can use ADC or a short-lived token from the active gcloud login. if ! gcloud auth application-default print-access-token >/dev/null 2>&1; then - echo "→ ADC not configured; using a gcloud access token for Terraform" - export GOOGLE_OAUTH_ACCESS_TOKEN="$(gcloud auth print-access-token)" + echo "→ ADC not configured; using a short-lived gcloud token for Terraform" + export GOOGLE_OAUTH_ACCESS_TOKEN + GOOGLE_OAUTH_ACCESS_TOKEN="$(gcloud auth print-access-token)" export GOOGLE_PROJECT="$PROJECT" fi -echo "→ terraform apply" -terraform -chdir="$TF_DIR" init -input=false >/dev/null -terraform -chdir="$TF_DIR" apply -input=false -auto-approve \ - -var "project_id=$PROJECT" -var "region=$REGION" -var "image=$IMAGE" \ - || { echo "ERROR: terraform apply failed" >&2; exit 3; } - -BUCKET="$(terraform -chdir="$TF_DIR" output -raw render_bucket_name)" -SERVICE_URL="$(terraform -chdir="$TF_DIR" output -raw service_url)" -WORKFLOW="$(terraform -chdir="$TF_DIR" output -raw workflow_name)" -echo " bucket=$BUCKET service=$SERVICE_URL workflow=$WORKFLOW" -cleanup() { - if [ "$KEEP_STACK" -eq 0 ]; then - echo "→ terraform destroy" - # Apply force_destroy=true into state FIRST. Terraform reads the bucket's - # force_destroy from prior state during the destroy step, so a destroy - # alone can't flip it; a quick apply updates the attribute, then destroy - # can empty + remove the (scratch) bucket. - terraform -chdir="$TF_DIR" apply -input=false -auto-approve \ - -var "project_id=$PROJECT" -var "region=$REGION" -var "image=$IMAGE" \ - -var "bucket_force_destroy=true" >/dev/null 2>&1 || true - terraform -chdir="$TF_DIR" destroy -input=false -auto-approve \ - -var "project_id=$PROJECT" -var "region=$REGION" -var "image=$IMAGE" \ - -var "bucket_force_destroy=true" || true - else - echo "→ --keep-stack set; leaving the stack up. Destroy with:" - echo " terraform -chdir=$TF_DIR destroy -var project_id=$PROJECT -var region=$REGION -var image=$IMAGE -var bucket_force_destroy=true" - fi +echo "→ Applying isolated Terraform stack" +terraform -chdir="$TF_WORK_DIR" init -input=false >/dev/null +# Mark the stack cleanup-eligible before apply. Terraform may persist a +# partially-created stack even when apply itself exits nonzero. +STACK_APPLIED=1 +terraform -chdir="$TF_WORK_DIR" apply -input=false -auto-approve \ + -var "project_id=$PROJECT" \ + -var "region=$REGION" \ + -var "project_name=$STACK_NAME" \ + -var "image=$IMAGE" || { + echo "ERROR: terraform apply failed" >&2 + exit 3 } -trap cleanup EXIT -# ── 4. Upload the fixture as a project tarball ────────────────────────────── +BUCKET="$(terraform -chdir="$TF_WORK_DIR" output -raw render_bucket_name)" +SERVICE_URL="$(terraform -chdir="$TF_WORK_DIR" output -raw service_url)" +WORKFLOW_NAME="$(terraform -chdir="$TF_WORK_DIR" output -raw workflow_name)" +echo " bucket=$BUCKET service=$SERVICE_URL workflow=$WORKFLOW_NAME" + SITE_TAR="$ARTIFACT_DIR/project.tar.gz" tar -czf "$SITE_TAR" -C "$FIXTURE_DIR/src" . -PROJECT_GCS="gs://$BUCKET/sites/$FIXTURE/project.tar.gz" +PROJECT_GCS="gs://$BUCKET/sites/smoke/$STACK_NAME/$FIXTURE/project.tar.gz" gcloud storage cp "$SITE_TAR" "$PROJECT_GCS" --project "$PROJECT" >/dev/null echo "→ Uploaded fixture to $PROJECT_GCS" -BASE_FPS=$(jq -r '.renderConfig.fps // 30' "$FIXTURE_META") -META_CHUNK=$(jq -r '.renderConfig.chunkSize // empty' "$FIXTURE_META") +BASE_FPS="$(jq -r '.renderConfig.fps // 30' "$FIXTURE_META")" +META_CHUNK="$(jq -r '.renderConfig.chunkSize // empty' "$FIXTURE_META")" [ -n "$CHUNK_SIZES" ] || CHUNK_SIZES="${META_CHUNK:-15}" +IFS=',' read -ra SIZES <<< "$CHUNK_SIZES" +for chunk_size in "${SIZES[@]}"; do + [[ "$chunk_size" =~ ^[1-9][0-9]*$ ]] || { + echo "ERROR: invalid positive integer chunk size '$chunk_size'" >&2 + exit 1 + } +done -echo "[]" > "$ARTIFACT_DIR/results.json" +jq -n '[]' > "$ARTIFACT_DIR/results.json" +jq -n '[]' > "$ARTIFACT_DIR/parity.json" OVERALL_RC=0 -IFS=',' read -ra SIZES <<< "$CHUNK_SIZES" -for CS in "${SIZES[@]}"; do - RENDER_ID="hf-smoke-c${CS}-$(date +%s)" - OUT_GCS="gs://$BUCKET/renders/$RENDER_ID/output.mp4" - ARG=$(jq -n \ - --arg svc "$SERVICE_URL" \ - --arg proj "$PROJECT_GCS" \ - --arg prefix "gs://$BUCKET/renders/$RENDER_ID/" \ - --arg out "$OUT_GCS" \ - --argjson fps "$BASE_FPS" \ - --argjson cs "$CS" \ - '{ServiceUrl:$svc, ProjectGcsUri:$proj, PlanOutputGcsPrefix:$prefix, OutputGcsUri:$out, - Config:{fps:$fps, width:640, height:360, format:"mp4", chunkSize:$cs}}') - - echo "→ Render chunkSize=$CS (renderId=$RENDER_ID)" - START_MS=$(date +%s%3N) - EXEC=$(gcloud workflows execute "$WORKFLOW" --location "$REGION" --project "$PROJECT" \ - --data "$ARG" --format='value(name)') - # Poll until terminal. - STATE="ACTIVE" - while [ "$STATE" = "ACTIVE" ] || [ "$STATE" = "QUEUED" ]; do - sleep 5 - STATE=$(gcloud workflows executions describe "$EXEC" --location "$REGION" \ - --project "$PROJECT" --format='value(state)') - done - END_MS=$(date +%s%3N) - WALL=$((END_MS - START_MS)) +for protocol in "${PROTOCOL_LIST[@]}"; do + for chunk_size in "${SIZES[@]}"; do + RENDER_ID="$STACK_NAME-$protocol-c$chunk_size" + OUTPUT_GCS="gs://$BUCKET/renders/smoke/$STACK_NAME/$protocol/c$chunk_size/output.mp4" + PLAN_PREFIX="gs://$BUCKET/renders/smoke/$STACK_NAME/$protocol/c$chunk_size/plan/" + ARGUMENTS="$(jq -n \ + --arg service "$SERVICE_URL" \ + --arg project_uri "$PROJECT_GCS" \ + --arg plan_prefix "$PLAN_PREFIX" \ + --arg output_uri "$OUTPUT_GCS" \ + --arg protocol "$protocol" \ + --argjson fps "$BASE_FPS" \ + --argjson chunk_size "$chunk_size" \ + '{ + ServiceUrl: $service, + ProjectGcsUri: $project_uri, + PlanOutputGcsPrefix: $plan_prefix, + OutputGcsUri: $output_uri, + PlanProtocol: $protocol, + Config: {fps: $fps, width: 640, height: 360, format: "mp4", chunkSize: $chunk_size} + }')" - gcloud workflows executions describe "$EXEC" --location "$REGION" --project "$PROJECT" \ - --format=json > "$ARTIFACT_DIR/renders/c$CS-execution.json" + echo "→ Render protocol=$protocol chunkSize=$chunk_size (renderId=$RENDER_ID)" + START_MS="$(date +%s%3N)" + EXECUTION="$(gcloud workflows execute "$WORKFLOW_NAME" \ + --location "$REGION" \ + --project "$PROJECT" \ + --data "$ARGUMENTS" \ + --format='value(name)')" + STATE="ACTIVE" + while [ "$STATE" = "ACTIVE" ] || [ "$STATE" = "QUEUED" ]; do + sleep 5 + STATE="$(gcloud workflows executions describe "$EXECUTION" \ + --location "$REGION" \ + --project "$PROJECT" \ + --format='value(state)')" + done + END_MS="$(date +%s%3N)" + WALL_MS=$((END_MS - START_MS)) - if [ "$STATE" != "SUCCEEDED" ]; then - echo " ✗ execution state=$STATE" - jq -r '.error.payload // empty' "$ARTIFACT_DIR/renders/c$CS-execution.json" | head -c 800 - OVERALL_RC=4 - continue - fi + EXECUTION_JSON="$RENDER_DIR/$protocol-c$chunk_size-execution.json" + gcloud workflows executions describe "$EXECUTION" \ + --location "$REGION" \ + --project "$PROJECT" \ + --format=json > "$EXECUTION_JSON" - OUT_LOCAL="$ARTIFACT_DIR/renders/c$CS-output.mp4" - gcloud storage cp "$OUT_GCS" "$OUT_LOCAL" --project "$PROJECT" >/dev/null + if [ "$STATE" != "SUCCEEDED" ]; then + echo " ✗ execution state=$STATE" + jq -r '.error.payload // empty' "$EXECUTION_JSON" | head -c 800 + [ "$OVERALL_RC" -ne 0 ] || OVERALL_RC=4 + continue + fi - # PSNR vs the in-process baseline. - PSNR_LOG="$ARTIFACT_DIR/renders/c$CS-psnr.log" - ffmpeg -y -i "$OUT_LOCAL" -i "$BASELINE_MP4" \ - -lavfi "psnr=stats_file=$PSNR_LOG" -f null - 2>/dev/null || true - PSNR_AVG=$(awk -F'psnr_avg:' '/psnr_avg:/{split($2,a," "); s+=a[1]; n++} END{if(n>0) printf "%.2f", s/n; else print "0"}' "$PSNR_LOG" 2>/dev/null || echo "0") + OUTPUT_LOCAL="$RENDER_DIR/$protocol-c$chunk_size-output.mp4" + gcloud storage cp "$OUTPUT_GCS" "$OUTPUT_LOCAL" --project "$PROJECT" >/dev/null - echo " ✓ state=SUCCEEDED wall=${WALL}ms psnr_avg=${PSNR_AVG}dB" - jq --argjson cs "$CS" --argjson wall "$WALL" --arg psnr "$PSNR_AVG" \ - '. += [{chunkSize:$cs, wallClockMs:$wall, psnrAvgDb:($psnr|tonumber)}]' \ - "$ARTIFACT_DIR/results.json" > "$ARTIFACT_DIR/results.json.tmp" && \ + PSNR_LOG="$RENDER_DIR/$protocol-c$chunk_size-psnr.log" + ffmpeg -y -i "$OUTPUT_LOCAL" -i "$BASELINE_MP4" \ + -lavfi "psnr=stats_file=$PSNR_LOG" -f null - 2>/dev/null || true + PSNR_AVG="$(awk -F'psnr_avg:' \ + '/psnr_avg:/{split($2,a," "); s+=a[1]; n++} END{if(n>0) printf "%.2f", s/n; else print "0"}' \ + "$PSNR_LOG" 2>/dev/null || echo "0")" + ENCODED_SHA="$(sha256sum "$OUTPUT_LOCAL" | cut -d' ' -f1)" + ENCODED_BYTES="$(wc -c < "$OUTPUT_LOCAL" | tr -d ' ')" + + echo " ✓ state=SUCCEEDED wall=${WALL_MS}ms psnr_avg=${PSNR_AVG}dB" + jq \ + --arg protocol "$protocol" \ + --argjson chunk_size "$chunk_size" \ + --argjson wall_ms "$WALL_MS" \ + --arg psnr "$PSNR_AVG" \ + --arg encoded_sha "$ENCODED_SHA" \ + --argjson encoded_bytes "$ENCODED_BYTES" \ + '. += [{ + planProtocol: $protocol, + chunkSize: $chunk_size, + wallClockMs: $wall_ms, + psnrAvgDb: ($psnr | tonumber), + encodedSha256: $encoded_sha, + encodedBytes: $encoded_bytes + }]' \ + "$ARTIFACT_DIR/results.json" > "$ARTIFACT_DIR/results.json.tmp" mv "$ARTIFACT_DIR/results.json.tmp" "$ARTIFACT_DIR/results.json" - if awk "BEGIN{exit !($PSNR_AVG < $PSNR_THRESHOLD)}"; then - echo " ✗ PSNR ${PSNR_AVG}dB below threshold ${PSNR_THRESHOLD}dB" - OVERALL_RC=5 - fi + if awk "BEGIN{exit !($PSNR_AVG < $PSNR_THRESHOLD)}"; then + echo " ✗ PSNR ${PSNR_AVG}dB below threshold ${PSNR_THRESHOLD}dB" + [ "$OVERALL_RC" -ne 0 ] || OVERALL_RC=5 + fi + done done -echo "→ Results:"; cat "$ARTIFACT_DIR/results.json" | jq . -exit $OVERALL_RC +canonicalize_output() { + local media_file="$1" + local output_prefix="$2" + ffmpeg -v error -i "$media_file" -map 0:v:0 -pix_fmt rgba \ + -f framemd5 "$output_prefix.frames.md5" + grep -v '^#' "$output_prefix.frames.md5" > "$output_prefix.frames.data" + + if ffprobe -v error -select_streams a:0 -show_entries stream=index \ + -of csv=p=0 "$media_file" | grep -q .; then + ffmpeg -v error -i "$media_file" -map 0:a:0 -vn -ac 2 -ar 48000 \ + -f hash -hash sha256 - > "$output_prefix.audio.sha256" + else + printf '%s\n' "NO_AUDIO" > "$output_prefix.audio.sha256" + fi + + ffprobe -v error \ + -show_entries \ + format=duration:stream=codec_type,codec_name,width,height,pix_fmt,avg_frame_rate,r_frame_rate,nb_frames,sample_rate,channels,channel_layout \ + -of json "$media_file" | jq -S . > "$output_prefix.probe.json" +} + +if [ -n "${SEEN_PROTOCOLS[v1]:-}" ] && [ -n "${SEEN_PROTOCOLS[v2]:-}" ]; then + echo "→ Comparing v1 and v2 decoded outputs" + for chunk_size in "${SIZES[@]}"; do + V1_OUTPUT="$RENDER_DIR/v1-c$chunk_size-output.mp4" + V2_OUTPUT="$RENDER_DIR/v2-c$chunk_size-output.mp4" + if [ ! -f "$V1_OUTPUT" ] || [ ! -f "$V2_OUTPUT" ]; then + echo " ✗ c$chunk_size parity unavailable because one protocol did not render" + [ "$OVERALL_RC" -ne 0 ] || OVERALL_RC=6 + continue + fi + + V1_PREFIX="$RENDER_DIR/v1-c$chunk_size-canonical" + V2_PREFIX="$RENDER_DIR/v2-c$chunk_size-canonical" + canonicalize_output "$V1_OUTPUT" "$V1_PREFIX" + canonicalize_output "$V2_OUTPUT" "$V2_PREFIX" + + FRAMES_EQUAL=false + AUDIO_EQUAL=false + METADATA_EQUAL=false + cmp -s "$V1_PREFIX.frames.data" "$V2_PREFIX.frames.data" && FRAMES_EQUAL=true + cmp -s "$V1_PREFIX.audio.sha256" "$V2_PREFIX.audio.sha256" && AUDIO_EQUAL=true + cmp -s "$V1_PREFIX.probe.json" "$V2_PREFIX.probe.json" && METADATA_EQUAL=true + + jq \ + --argjson chunk_size "$chunk_size" \ + --argjson frames_equal "$FRAMES_EQUAL" \ + --argjson audio_equal "$AUDIO_EQUAL" \ + --argjson metadata_equal "$METADATA_EQUAL" \ + --arg v1_frame_sha "$(sha256sum "$V1_PREFIX.frames.data" | cut -d' ' -f1)" \ + --arg v2_frame_sha "$(sha256sum "$V2_PREFIX.frames.data" | cut -d' ' -f1)" \ + --arg v1_audio_sha "$(sed -n '1p' "$V1_PREFIX.audio.sha256")" \ + --arg v2_audio_sha "$(sed -n '1p' "$V2_PREFIX.audio.sha256")" \ + '. += [{ + chunkSize: $chunk_size, + decodedFramesEqual: $frames_equal, + decodedAudioEqual: $audio_equal, + normalizedMetadataEqual: $metadata_equal, + v1FrameManifestSha256: $v1_frame_sha, + v2FrameManifestSha256: $v2_frame_sha, + v1AudioSha256: $v1_audio_sha, + v2AudioSha256: $v2_audio_sha + }]' \ + "$ARTIFACT_DIR/parity.json" > "$ARTIFACT_DIR/parity.json.tmp" + mv "$ARTIFACT_DIR/parity.json.tmp" "$ARTIFACT_DIR/parity.json" + + if [ "$FRAMES_EQUAL" = true ] && [ "$AUDIO_EQUAL" = true ] && [ "$METADATA_EQUAL" = true ]; then + echo " ✓ c$chunk_size decoded video, audio, and metadata match" + else + echo " ✗ c$chunk_size parity mismatch: frames=$FRAMES_EQUAL audio=$AUDIO_EQUAL metadata=$METADATA_EQUAL" + [ "$OVERALL_RC" -ne 0 ] || OVERALL_RC=6 + fi + done +fi + +echo "→ Results" +jq . "$ARTIFACT_DIR/results.json" +if [ -n "${SEEN_PROTOCOLS[v1]:-}" ] && [ -n "${SEEN_PROTOCOLS[v2]:-}" ]; then + echo "→ Parity" + jq . "$ARTIFACT_DIR/parity.json" +fi +echo "→ Evidence: $ARTIFACT_DIR" +exit "$OVERALL_RC" diff --git a/packages/gcp-cloud-run/Dockerfile b/packages/gcp-cloud-run/Dockerfile index 20bf5aa7e5..a084849e87 100644 --- a/packages/gcp-cloud-run/Dockerfile +++ b/packages/gcp-cloud-run/Dockerfile @@ -97,6 +97,7 @@ COPY packages/parsers/package.json packages/parsers/package.json COPY packages/sdk/package.json packages/sdk/package.json COPY packages/sdk-playground/package.json packages/sdk-playground/package.json COPY packages/studio-server/package.json packages/studio-server/package.json +COPY scripts/package-subpaths.mjs scripts/package-subpaths.mjs RUN bun install --frozen-lockfile # Copy source for the packages the render path needs. diff --git a/packages/gcp-cloud-run/package.json b/packages/gcp-cloud-run/package.json index a0b2d1d688..2d3a24abf4 100644 --- a/packages/gcp-cloud-run/package.json +++ b/packages/gcp-cloud-run/package.json @@ -81,7 +81,8 @@ "@types/tar": "^6.1.13", "esbuild": "^0.25.12", "tsx": "^4.21.0", - "typescript": "^5.7.2" + "typescript": "^5.7.2", + "yaml": "^2.9.0" }, "engines": { "node": ">=22" diff --git a/packages/gcp-cloud-run/src/__fixtures__/fakeGcs.ts b/packages/gcp-cloud-run/src/__fixtures__/fakeGcs.ts index f49dc31b2d..980b81f9ab 100644 --- a/packages/gcp-cloud-run/src/__fixtures__/fakeGcs.ts +++ b/packages/gcp-cloud-run/src/__fixtures__/fakeGcs.ts @@ -24,6 +24,9 @@ export interface FakeGcsOp { export class FakeGcs { ops: FakeGcsOp[] = []; objects = new Map(); + metadata = new Map>(); + /** Optional one-shot race hook used to model another writer before a create-only upload. */ + beforeUpload?: (uri: string) => void; // Accessed only through the `Storage` cast in tests, so fallow's static // analysis can't see the reference. @@ -57,13 +60,30 @@ class FakeBucket { return new FakeFile(this.gcs, this.bucketName, key); } + // The stateful command matrix intentionally stays inline so tests exercise + // generation preconditions and metadata updates as one fake GCS operation. + // fallow-ignore-next-line complexity async upload( localPath: string, - opts: { destination: string; contentType?: string }, + opts: { + destination: string; + contentType?: string; + metadata?: { metadata?: Record }; + preconditionOpts?: { ifGenerationMatch?: number }; + }, ): Promise { const uri = `gs://${this.bucketName}/${opts.destination}`; + this.gcs.beforeUpload?.(uri); + if (opts.preconditionOpts?.ifGenerationMatch === 0 && this.gcs.objects.has(uri)) { + const error = new Error(`FakeGcs: precondition failed: ${uri}`) as Error & { + code: number; + }; + error.code = 412; + throw error; + } const bytes = readFileSync(localPath); this.gcs.objects.set(uri, bytes); + this.gcs.metadata.set(uri, opts.metadata?.metadata ?? {}); this.gcs.ops.push({ kind: "upload", uri, bytes: bytes.length }); return [{}]; } @@ -96,10 +116,18 @@ class FakeFile { return [has]; } - async getMetadata(): Promise<[{ size?: string | number; updated?: string }]> { + async getMetadata(): Promise< + [{ size?: string | number; updated?: string; metadata?: Record }] + > { const bytes = this.gcs.objects.get(this.uri); this.gcs.ops.push({ kind: "getMetadata", uri: this.uri }); - return [{ size: bytes?.length ?? 0, updated: "2026-06-06T00:00:00.000Z" }]; + return [ + { + size: bytes?.length ?? 0, + updated: "2026-06-06T00:00:00.000Z", + metadata: this.gcs.metadata.get(this.uri) ?? {}, + }, + ]; } /** Helper for tests that want to materialize an object to disk. */ diff --git a/packages/gcp-cloud-run/src/events.ts b/packages/gcp-cloud-run/src/events.ts index bafde86d37..98ab840aa1 100644 --- a/packages/gcp-cloud-run/src/events.ts +++ b/packages/gcp-cloud-run/src/events.ts @@ -31,6 +31,8 @@ export type { SerializableDistributedRenderConfig } from "@hyperframes/producer/ /** Discriminator for the three roles the one Cloud Run image fulfills. */ export type CloudRunAction = "plan" | "renderChunk" | "assemble"; +/** Transport protocol selected for one complete distributed render. */ +export type CloudRunPlanProtocol = "v1" | "v2"; /** * Top-level shape of any request body the handler may receive. @@ -48,7 +50,7 @@ export type CloudRunEvent = | { Input: CloudRunEvent }; /** Activity A: produce a planDir, upload to GCS. */ -export interface PlanEvent { +interface PlanEventBase { Action: "plan"; /** GCS URI pointing at a `tar -czf`-archived project directory (`gs://bucket/key.tar.gz`). */ ProjectGcsUri: string; @@ -58,11 +60,21 @@ export interface PlanEvent { Config: SerializableDistributedRenderConfig; } +/** Legacy/default plan transport. Absence is deliberately interpreted as v1. */ +export interface PlanV1Event extends PlanEventBase { + PlanProtocol?: "v1"; +} + +/** Explicit opt-in to the content-addressed v2 plan transport. */ +export interface PlanV2Event extends PlanEventBase { + PlanProtocol: "v2"; +} + +export type PlanEvent = PlanV1Event | PlanV2Event; + /** Activity B: fetch planDir, render one chunk, upload result. */ -export interface RenderChunkEvent { +interface RenderChunkEventBase { Action: "renderChunk"; - /** GCS URI of the plan tar produced by a PlanEvent invocation. */ - PlanGcsUri: string; /** * `PlanResult.planHash` from the Plan invocation. The handler verifies * this against the untarred planDir's `plan.json` before invoking the @@ -79,15 +91,33 @@ export interface RenderChunkEvent { Format: DistributedFormat; } +/** Legacy/default chunk event. */ +export interface RenderChunkV1Event extends RenderChunkEventBase { + PlanProtocol?: "v1"; + /** GCS URI of the v1 plan tar produced by a PlanEvent invocation. */ + PlanGcsUri: string; + PlanV2ManifestGcsUri?: never; + PlanV2ArtifactGcsPrefix?: never; +} + +/** + * V2 chunk event. It intentionally cannot carry `PlanGcsUri`: the manifest + * describes the exact content-addressed artifacts needed by this chunk. + */ +export interface RenderChunkV2Event extends RenderChunkEventBase { + PlanProtocol: "v2"; + PlanV2ManifestGcsUri: string; + PlanV2ArtifactGcsPrefix: string; + PlanGcsUri?: never; +} + +export type RenderChunkEvent = RenderChunkV1Event | RenderChunkV2Event; + /** Activity C: fetch planDir + all chunks + audio, assemble, upload final. */ -export interface AssembleEvent { +interface AssembleEventBase { Action: "assemble"; - /** GCS URI of the plan tar produced by a PlanEvent invocation. */ - PlanGcsUri: string; /** GCS URIs of every chunk, ordered by chunk index. Length must equal `chunkCount`. */ ChunkGcsUris: string[]; - /** GCS URI of the planDir's `audio.aac` if the composition has audio; `null` otherwise. */ - AudioGcsUri: string | null; /** Final output GCS URI (`gs://bucket/key.mp4`). */ OutputGcsUri: string; /** Output container format; drives file vs frame-dir handling. */ @@ -104,12 +134,35 @@ export interface AssembleEvent { Cfr?: boolean; } +/** Legacy/default assemble event. */ +export interface AssembleV1Event extends AssembleEventBase { + PlanProtocol?: "v1"; + /** GCS URI of the v1 plan tar produced by a PlanEvent invocation. */ + PlanGcsUri: string; + /** Legacy standalone audio locator; `null` when audio is embedded in the v1 plan tar. */ + AudioGcsUri: string | null; + PlanV2ManifestGcsUri?: never; + PlanV2ArtifactGcsPrefix?: never; +} + +/** V2 assemble event, scoped to manifest-declared assembler artifacts. */ +export interface AssembleV2Event extends AssembleEventBase { + PlanProtocol: "v2"; + PlanV2ManifestGcsUri: string; + PlanV2ArtifactGcsPrefix: string; + PlanHash: string; + PlanGcsUri?: never; + /** V2 audio is a manifest artifact materialized only for the assembler. */ + AudioGcsUri: null; +} + +export type AssembleEvent = AssembleV1Event | AssembleV2Event; + // ── Result types — kept small to fit Cloud Workflows step budgets ──────────── /** Result of a `plan` invocation. Carries enough to size the Map(N) state. */ -export interface PlanResultBody { +interface PlanResultBodyBase { Action: "plan"; - PlanGcsUri: string; PlanHash: string; ChunkCount: number; TotalFrames: number; @@ -124,6 +177,25 @@ export interface PlanResultBody { DurationMs: number; } +/** Existing v1 result. Kept unchanged for wire compatibility. */ +export interface PlanV1ResultBody extends PlanResultBodyBase { + PlanGcsUri: string; + PlanProtocol?: never; + PlanV2ManifestGcsUri?: never; + PlanV2ArtifactGcsPrefix?: never; +} + +/** V2 result. The two v2 locators are never aliases for `PlanGcsUri`. */ +export interface PlanV2ResultBody extends PlanResultBodyBase { + PlanProtocol: "v2"; + PlanV2ManifestGcsUri: string; + PlanV2ArtifactGcsPrefix: string; + PlanGcsUri?: never; + AudioGcsUri: null; +} + +export type PlanResultBody = PlanV1ResultBody | PlanV2ResultBody; + /** Result of a `renderChunk` invocation. Sized ≤200 bytes. */ export interface RenderChunkResultBody { Action: "renderChunk"; diff --git a/packages/gcp-cloud-run/src/gcsTransport.test.ts b/packages/gcp-cloud-run/src/gcsTransport.test.ts index 8b12de0759..e43d683c75 100644 --- a/packages/gcp-cloud-run/src/gcsTransport.test.ts +++ b/packages/gcp-cloud-run/src/gcsTransport.test.ts @@ -10,10 +10,13 @@ import { join } from "node:path"; import { asStorage, FakeGcs } from "./__fixtures__/fakeGcs.js"; import { downloadGcsObjectToFile, + downloadGcsObjectToFileVerified, formatGcsUri, parseGcsUri, + sha256File, tarDirectory, untarDirectory, + uploadContentAddressedFileToGcs, uploadFileToGcs, } from "./gcsTransport.js"; @@ -112,3 +115,71 @@ describe("download/upload bridge", () => { ); }); }); + +describe("content-addressed v2 artifacts", () => { + it("uploads once and reuses an object with matching digest metadata", async () => { + const gcs = new FakeGcs(); + const source = join(mkTmp("hf-cas-upload-"), "artifact.bin"); + writeFileSync(source, "immutable bytes"); + const digest = await sha256File(source); + const uri = `gs://bucket/v2/artifacts/sha256/${digest.slice(0, 2)}/${digest}`; + + expect(await uploadContentAddressedFileToGcs(asStorage(gcs), source, uri, digest)).toBe( + "uploaded", + ); + expect(await uploadContentAddressedFileToGcs(asStorage(gcs), source, uri, digest)).toBe( + "reused", + ); + expect(gcs.ops.filter((op) => op.kind === "upload")).toHaveLength(1); + expect(gcs.metadata.get(uri)?.sha256).toBe(digest); + }); + + it("refuses to overwrite an immutable key with conflicting metadata", async () => { + const gcs = new FakeGcs(); + const source = join(mkTmp("hf-cas-conflict-"), "artifact.bin"); + writeFileSync(source, "expected bytes"); + const digest = await sha256File(source); + const uri = `gs://bucket/v2/artifacts/sha256/${digest.slice(0, 2)}/${digest}`; + gcs.seed(uri, Buffer.from("same length!!!")); + gcs.metadata.set(uri, { sha256: "0".repeat(64) }); + + await expect( + uploadContentAddressedFileToGcs(asStorage(gcs), source, uri, digest), + ).rejects.toMatchObject({ name: "PLAN_ARTIFACT_DIGEST_MISMATCH" }); + expect(gcs.ops.some((op) => op.kind === "upload")).toBe(false); + }); + + it("reuses an identical object that wins the create-only generation race", async () => { + const gcs = new FakeGcs(); + const source = join(mkTmp("hf-cas-race-"), "artifact.bin"); + writeFileSync(source, "racing bytes"); + const digest = await sha256File(source); + const uri = `gs://bucket/v2/artifacts/sha256/${digest.slice(0, 2)}/${digest}`; + gcs.beforeUpload = (uploadUri) => { + gcs.beforeUpload = undefined; + gcs.seed(uploadUri, readFileSync(source)); + gcs.metadata.set(uploadUri, { sha256: digest }); + }; + + expect(await uploadContentAddressedFileToGcs(asStorage(gcs), source, uri, digest)).toBe( + "reused", + ); + expect(gcs.ops.some((op) => op.kind === "upload")).toBe(false); + }); + + it("deletes a downloaded artifact when digest verification fails", async () => { + const gcs = new FakeGcs(); + const work = mkTmp("hf-cas-download-"); + const expectedSource = join(work, "expected.bin"); + const destination = join(work, "download.bin"); + writeFileSync(expectedSource, "expected"); + const expected = await sha256File(expectedSource); + const uri = "gs://bucket/v2/artifacts/corrupt"; + gcs.seed(uri, Buffer.from("corrupt")); + + await expect( + downloadGcsObjectToFileVerified(asStorage(gcs), uri, destination, expected), + ).rejects.toMatchObject({ name: "PLAN_ARTIFACT_DIGEST_MISMATCH" }); + expect(existsSync(destination)).toBe(false); + }); +}); diff --git a/packages/gcp-cloud-run/src/gcsTransport.ts b/packages/gcp-cloud-run/src/gcsTransport.ts index a3a0913c4f..4147ad09ae 100644 --- a/packages/gcp-cloud-run/src/gcsTransport.ts +++ b/packages/gcp-cloud-run/src/gcsTransport.ts @@ -19,7 +19,15 @@ * is the same shape as `@hyperframes/aws-lambda`'s `s3Transport.ts`. */ -import { createWriteStream, existsSync, mkdirSync, rmSync, statSync } from "node:fs"; +import { + createReadStream, + createWriteStream, + existsSync, + mkdirSync, + rmSync, + statSync, +} from "node:fs"; +import { createHash } from "node:crypto"; import { dirname } from "node:path"; import { pipeline } from "node:stream/promises"; import type { Storage } from "@google-cloud/storage"; @@ -70,6 +78,26 @@ export async function downloadGcsObjectToFile( await pipeline(file.createReadStream(), createWriteStream(destPath)); } +/** Download and verify an immutable plan-v2 artifact before materialization. */ +export async function downloadGcsObjectToFileVerified( + storage: Storage, + uri: string, + destPath: string, + expectedSha256: string, +): Promise { + assertSha256(expectedSha256); + await downloadGcsObjectToFile(storage, uri, destPath); + const actual = await sha256File(destPath); + if (actual !== expectedSha256) { + rmSync(destPath, { force: true }); + const error = new Error( + `[gcsTransport] PLAN_ARTIFACT_DIGEST_MISMATCH: ${uri} expected ${expectedSha256}, got ${actual}`, + ); + error.name = "PLAN_ARTIFACT_DIGEST_MISMATCH"; + throw error; + } +} + /** * Upload a local file's contents to a GCS URI using a resumable upload. * GCS objects have no practical size ceiling for the artifacts this adapter @@ -96,6 +124,118 @@ export async function uploadFileToGcs( }); } +/** + * Upload one content-addressed plan-v2 artifact exactly once. + * + * The zero-generation precondition makes creation atomic. Existing objects + * are reused only when their immutable digest metadata and byte length agree; + * a conflict is never overwritten because another render may already consume + * that object. + */ +export async function uploadContentAddressedFileToGcs( + storage: Storage, + localPath: string, + uri: string, + expectedSha256: string, + contentType?: string, +): Promise<"uploaded" | "reused"> { + assertSha256(expectedSha256); + if (!existsSync(localPath)) { + throw new Error(`[gcsTransport] upload source missing: ${localPath}`); + } + const actualSha256 = await sha256File(localPath); + if (actualSha256 !== expectedSha256) { + throwDigestMismatch( + `local artifact ${localPath} expected ${expectedSha256}, got ${actualSha256}`, + ); + } + + const { bucket, key } = parseGcsUri(uri); + const bucketHandle = storage.bucket(bucket); + const file = bucketHandle.file(key); + const size = statSync(localPath).size; + if (await isReusableContentAddressedObject(file, uri, size, expectedSha256)) { + return "reused"; + } + + try { + await bucketHandle.upload(localPath, { + destination: key, + contentType, + metadata: { metadata: { sha256: expectedSha256 } }, + preconditionOpts: { ifGenerationMatch: 0 }, + }); + return "uploaded"; + } catch (error) { + // A concurrent planner may win the create-only race. Reuse only after + // verifying that the winning object is exactly the immutable CAS value. + if ( + isGcsPreconditionFailed(error) && + (await isReusableContentAddressedObject(file, uri, size, expectedSha256)) + ) { + return "reused"; + } + throw error; + } +} + +interface GcsFileLike { + exists(): Promise<[boolean, ...unknown[]]>; + getMetadata(): Promise< + [ + { + size?: string | number; + metadata?: Record; + }, + ...unknown[], + ] + >; +} + +async function isReusableContentAddressedObject( + file: GcsFileLike, + uri: string, + expectedSize: number, + expectedSha256: string, +): Promise { + const [exists] = await file.exists(); + if (!exists) return false; + const [metadata] = await file.getMetadata(); + if (Number(metadata.size) === expectedSize && metadata.metadata?.sha256 === expectedSha256) { + return true; + } + throwDigestMismatch( + `immutable object ${uri} already exists with different digest metadata or size`, + ); +} + +export async function sha256File(path: string): Promise { + const hash = createHash("sha256"); + for await (const chunk of createReadStream(path)) { + hash.update(chunk as Buffer); + } + return hash.digest("hex"); +} + +function assertSha256(value: string): void { + if (!/^[a-f0-9]{64}$/.test(value)) { + throw new Error( + `[gcsTransport] expected lowercase SHA-256 digest, got ${JSON.stringify(value)}`, + ); + } +} + +function throwDigestMismatch(detail: string): never { + const error = new Error(`[gcsTransport] PLAN_ARTIFACT_DIGEST_MISMATCH: ${detail}`); + error.name = "PLAN_ARTIFACT_DIGEST_MISMATCH"; + throw error; +} + +function isGcsPreconditionFailed(error: unknown): boolean { + if (!error || typeof error !== "object") return false; + return (error as { code?: unknown }).code === 412; +} + /** * Pack a directory into a `.tar.gz` at `destTarball`. Uses the `tar` npm * package (pure JS over `node:zlib`) rather than spawning a system tar diff --git a/packages/gcp-cloud-run/src/index.ts b/packages/gcp-cloud-run/src/index.ts index cb34289e7a..4967d5a02b 100644 --- a/packages/gcp-cloud-run/src/index.ts +++ b/packages/gcp-cloud-run/src/index.ts @@ -21,24 +21,36 @@ export { createApp, dispatch, type HandlerDeps, startServer, unwrapEvent } from "./server.js"; export { type AssembleEvent, + type AssembleV1Event, + type AssembleV2Event, type AssembleResultBody, type CloudRunAction, type CloudRunEvent, + type CloudRunPlanProtocol, type CloudRunResult, type PlanEvent, type PlanResultBody, + type PlanV1Event, + type PlanV1ResultBody, + type PlanV2Event, + type PlanV2ResultBody, type RenderChunkEvent, type RenderChunkResultBody, + type RenderChunkV1Event, + type RenderChunkV2Event, type SerializableDistributedRenderConfig, } from "./events.js"; export { ChromeBinaryUnavailableError, resolveChromeExecutablePath } from "./chromium.js"; export { downloadGcsObjectToFile, + downloadGcsObjectToFileVerified, formatGcsUri, type GcsLocation, parseGcsUri, + sha256File, tarDirectory, untarDirectory, + uploadContentAddressedFileToGcs, uploadFileToGcs, } from "./gcsTransport.js"; diff --git a/packages/gcp-cloud-run/src/sdk/renderToCloudRun.test.ts b/packages/gcp-cloud-run/src/sdk/renderToCloudRun.test.ts index 8a41d5c125..e7d2e39818 100644 --- a/packages/gcp-cloud-run/src/sdk/renderToCloudRun.test.ts +++ b/packages/gcp-cloud-run/src/sdk/renderToCloudRun.test.ts @@ -78,11 +78,19 @@ describe("renderToCloudRun", () => { 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(arg.PlanProtocol).toBe("v1"); expect(fake.lastParent).toBe( "projects/proj/locations/us-central1/workflows/hyperframes-render", ); }); + it("forwards an explicit v2 whole-render opt-in", async () => { + const fake = new FakeExecutions(); + await renderToCloudRun({ ...opts(fake), planProtocol: "v2" }); + const arg = JSON.parse(fake.lastArgument ?? "{}"); + expect(arg.PlanProtocol).toBe("v2"); + }); + it("derives the output extension from the format", async () => { const fake = new FakeExecutions(); const handle = await renderToCloudRun({ diff --git a/packages/gcp-cloud-run/src/sdk/renderToCloudRun.ts b/packages/gcp-cloud-run/src/sdk/renderToCloudRun.ts index 116d290716..1fc8937802 100644 --- a/packages/gcp-cloud-run/src/sdk/renderToCloudRun.ts +++ b/packages/gcp-cloud-run/src/sdk/renderToCloudRun.ts @@ -26,7 +26,7 @@ import { randomUUID } from "node:crypto"; import type { Storage } from "@google-cloud/storage"; -import type { SerializableDistributedRenderConfig } from "../events.js"; +import type { CloudRunPlanProtocol, SerializableDistributedRenderConfig } from "../events.js"; import { formatExtension } from "../formatExtension.js"; import { formatGcsUri } from "../gcsTransport.js"; import { deploySite, type SiteHandle } from "./deploySite.js"; @@ -52,6 +52,11 @@ export interface RenderToCloudRunOptions { siteHandle?: SiteHandle; /** Validated `SerializableDistributedRenderConfig` (no logger / abortSignal). */ config: SerializableDistributedRenderConfig; + /** + * Distributed plan transport. Defaults to `"v1"` for backwards + * compatibility; v2 is always an explicit whole-render opt-in. + */ + planProtocol?: CloudRunPlanProtocol; /** GCS bucket from the Terraform output (`render_bucket_name`). */ bucketName: string; /** GCP project id hosting the workflow. */ @@ -144,6 +149,7 @@ export async function renderToCloudRun(opts: RenderToCloudRunOptions): Promisegcp v2 fixture"); + const planJson = { + planHash: "a".repeat(64), + chunkCount: 1, + totalFrames: 30, + dimensions: { fpsNum: 30, fpsDen: 1, width: 640, height: 360, format: "mp4" }, + ffmpegVersion: "6.0", + producerVersion: "test", + fontSnapshotSha: "font-snapshot-test", + }; + writeFileSync(join(dir, "plan.json"), JSON.stringify(planJson)); + writeFileSync( + join(dir, "meta", "chunks.json"), + JSON.stringify([{ index: 0, startFrame: 0, endFrame: 30 }]), + ); + writeFileSync(join(dir, "meta", "encoder.json"), "{}"); + if (withAudio) writeFileSync(join(dir, "audio.aac"), "AAC"); + planJson.planHash = recomputePlanHashFromPlanDir(dir); + writeFileSync(join(dir, "plan.json"), JSON.stringify(planJson)); +} + const planResult: PlanResult = { planDir: "(set at call time)", planProtocol: CURRENT_PLAN_PROTOCOL, @@ -207,6 +237,139 @@ describe("dispatch", () => { expect(gcs.objects.has("gs://b/renders/r1/output.mp4")).toBe(true); }); + // This end-to-end adapter contract is intentionally one narrative test: it + // verifies ordering and target isolation across all three handler roles. + // fallow-ignore-next-line complexity + it("runs v2 plan → target-scoped chunk → assemble with manifest-last CAS", async () => { + const gcs = new FakeGcs(); + await seedProjectTar(gcs, "gs://b/sites/v2/project.tar.gz"); + const root = mkTmp("hf-v2-e2e-"); + const planV2 = async ( + _projectDir: string, + _config: unknown, + planV2Dir: string, + ): Promise => { + const v1Dir = join(root, "v1"); + makeMinimalV1PlanDir(v1Dir, true); + return createPlanV2FromV1(v1Dir, planV2Dir); + }; + const renderChunk = async ( + planDir: string, + chunkIndex: number, + outputBase: string, + ): Promise => { + expect(existsSync(join(planDir, "audio.aac"))).toBe(false); + writeFileSync(outputBase, `chunk-${chunkIndex}`); + return { + outputPath: outputBase, + outputKind: "file", + framesEncoded: 30, + sha256: "b".repeat(64), + }; + }; + const assemble = async ( + _planDir: string, + _chunks: string[], + audioPath: string | null, + finalOutput: string, + ): Promise => { + expect(audioPath).not.toBeNull(); + expect(readFileSync(audioPath as string, "utf8")).toBe("AAC"); + writeFileSync(finalOutput, "v2-output"); + return { framesEncoded: 30, fileSize: 9 }; + }; + const deps = depsWith(gcs, { planV2, renderChunk, assemble }); + + const planned = await dispatch( + { + Action: "plan", + PlanProtocol: "v2", + ProjectGcsUri: "gs://b/sites/v2/project.tar.gz", + PlanOutputGcsPrefix: "gs://b/renders/v2/", + Config: { fps: 30, width: 640, height: 360, format: "mp4" }, + }, + deps, + ); + expect(planned).toMatchObject({ + PlanProtocol: "v2", + PlanV2ManifestGcsUri: "gs://b/renders/v2/v2/manifest.json", + PlanV2ArtifactGcsPrefix: "gs://b/renders/v2/v2/artifacts/sha256", + AudioGcsUri: null, + }); + expect("PlanGcsUri" in planned).toBe(false); + const uploadUris = gcs.ops.filter((op) => op.kind === "upload").map((op) => op.uri); + expect(uploadUris.at(-1)).toBe("gs://b/renders/v2/v2/manifest.json"); + expect(gcs.metadata.get(uploadUris.at(-1) ?? "")?.sha256).toMatch(/^[a-f0-9]{64}$/); + await dispatch( + { + Action: "plan", + PlanProtocol: "v2", + ProjectGcsUri: "gs://b/sites/v2/project.tar.gz", + PlanOutputGcsPrefix: "gs://b/renders/v2/", + Config: { fps: 30, width: 640, height: 360, format: "mp4" }, + }, + deps, + ); + expect(gcs.ops.filter((op) => op.kind === "upload").map((op) => op.uri)).toEqual(uploadUris); + if (!("PlanProtocol" in planned) || planned.PlanProtocol !== "v2") { + throw new Error("expected v2 plan result"); + } + + const beforeChunk = gcs.ops.length; + const chunk = await dispatch( + { + Action: "renderChunk", + PlanProtocol: "v2", + PlanV2ManifestGcsUri: planned.PlanV2ManifestGcsUri, + PlanV2ArtifactGcsPrefix: planned.PlanV2ArtifactGcsPrefix, + PlanHash: planned.PlanHash, + ChunkIndex: 0, + ChunkOutputGcsPrefix: "gs://b/renders/v2/", + Format: "mp4", + }, + deps, + ); + if (chunk.Action !== "renderChunk") throw new Error("expected chunk result"); + const audioDigest = createHash("sha256").update("AAC").digest("hex"); + const audioUri = `${planned.PlanV2ArtifactGcsPrefix}/${audioDigest.slice(0, 2)}/${audioDigest}`; + expect(gcs.ops.slice(beforeChunk).some((operation) => operation.uri === audioUri)).toBe(false); + + await dispatch( + { + Action: "assemble", + PlanProtocol: "v2", + PlanV2ManifestGcsUri: planned.PlanV2ManifestGcsUri, + PlanV2ArtifactGcsPrefix: planned.PlanV2ArtifactGcsPrefix, + PlanHash: planned.PlanHash, + ChunkGcsUris: [chunk.ChunkGcsUri], + AudioGcsUri: null, + OutputGcsUri: "gs://b/renders/v2/output.mp4", + Format: "mp4", + }, + deps, + ); + expect( + gcs.ops.some((operation) => operation.kind === "download" && operation.uri === audioUri), + ).toBe(true); + }); + + it("rejects mixed v1/v2 locators at runtime", async () => { + const event = { + Action: "renderChunk", + PlanProtocol: "v2", + PlanGcsUri: "gs://b/renders/r1/plan.tar.gz", + PlanV2ManifestGcsUri: "gs://b/renders/r1/v2/manifest.json", + PlanV2ArtifactGcsPrefix: "gs://b/renders/r1/v2/artifacts/sha256", + PlanHash: PLAN_HASH, + ChunkIndex: 0, + ChunkOutputGcsPrefix: "gs://b/renders/r1/", + Format: "mp4", + } as unknown as RenderChunkEvent; + await expect(dispatch(event, depsWith(new FakeGcs()))).rejects.toMatchObject({ + name: "PLAN_PROTOCOL_UNSUPPORTED", + }); + }); + it("rejects an unknown action", async () => { const gcs = new FakeGcs(); await expect( @@ -236,6 +399,27 @@ describe("bucket allowlist guard", () => { } }); + it("checks both v2 manifest and artifact-prefix buckets", async () => { + const prev = process.env.HYPERFRAMES_RENDER_BUCKET; + process.env.HYPERFRAMES_RENDER_BUCKET = "allowed-bucket"; + try { + const event: RenderChunkEvent = { + Action: "renderChunk", + PlanProtocol: "v2", + PlanV2ManifestGcsUri: "gs://allowed-bucket/v2/manifest.json", + PlanV2ArtifactGcsPrefix: "gs://evil-bucket/v2/artifacts/sha256", + PlanHash: PLAN_HASH, + ChunkIndex: 0, + ChunkOutputGcsPrefix: "gs://allowed-bucket/renders/r1/", + Format: "mp4", + }; + await expect(dispatch(event, depsWith(new FakeGcs()))).rejects.toThrow(/GCS_URI_NOT_ALLOWED/); + } finally { + if (prev === undefined) delete process.env.HYPERFRAMES_RENDER_BUCKET; + else process.env.HYPERFRAMES_RENDER_BUCKET = prev; + } + }); + it('treats HYPERFRAMES_RENDER_BUCKET="*" as an explicit opt-out (off-bucket allowed)', async () => { const gcs = new FakeGcs(); await seedPlanTar(gcs, "gs://any-bucket/renders/r1/plan.tar.gz", PLAN_HASH); @@ -327,7 +511,45 @@ describe("createApp HTTP mapping", () => { expect(res.status).toBe(400); const body = (await res.json()) as { error: string }; - expect(body.error).toBe("PlanProtocolUnsupportedError"); + expect(body.error).toBe("PLAN_PROTOCOL_UNSUPPORTED"); + }); + + it("returns 400 for plan v2 integrity error names and code aliases", async () => { + for (const error of [ + new PlanV2IntegrityError("corrupt test artifact"), + Object.assign(new Error("corrupt test artifact"), { + name: PLAN_V2_INTEGRITY_UNRECOVERABLE, + }), + Object.assign(new Error("corrupt test artifact"), { + code: PLAN_V2_INTEGRITY_UNRECOVERABLE, + }), + ]) { + const gcs = new FakeGcs(); + await seedPlanTar(gcs, "gs://b/renders/r1/plan.tar.gz", PLAN_HASH); + const app = createApp( + depsWith(gcs, { + renderChunk: async () => { + throw error; + }, + }), + ); + const res = await app.request("/", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + Action: "renderChunk", + PlanGcsUri: "gs://b/renders/r1/plan.tar.gz", + PlanHash: PLAN_HASH, + ChunkIndex: 0, + ChunkOutputGcsPrefix: "gs://b/renders/r1/", + Format: "mp4", + }), + }); + + expect(res.status).toBe(400); + const body = (await res.json()) as { error: string }; + expect(body.error).toBe(PLAN_V2_INTEGRITY_UNRECOVERABLE); + } }); it("returns 500 for a retryable/unknown error", async () => { diff --git a/packages/gcp-cloud-run/src/server.ts b/packages/gcp-cloud-run/src/server.ts index 7aa7f84878..98120a8699 100644 --- a/packages/gcp-cloud-run/src/server.ts +++ b/packages/gcp-cloud-run/src/server.ts @@ -28,8 +28,15 @@ import { type AssembleResult, type ChunkResult, type DistributedRenderConfig, + listPlanV2ArtifactsForTarget, + materializePlanV2Target, plan, + planV2, type PlanResult, + type PlanV2Artifact, + type PlanV2MaterializationTarget, + type PlanV2Result, + readPlanV2Manifest, renderChunk, } from "@hyperframes/producer/distributed"; import { resolveChromeExecutablePath } from "./chromium.js"; @@ -47,9 +54,12 @@ import type { import { type DistributedFormat, formatExtension } from "./formatExtension.js"; import { downloadGcsObjectToFile, + downloadGcsObjectToFileVerified, parseGcsUri, + sha256File, tarDirectory, untarDirectory, + uploadContentAddressedFileToGcs, uploadFileToGcs, } from "./gcsTransport.js"; @@ -74,6 +84,7 @@ export interface HandlerDeps { storage?: Storage; primitives?: { plan: typeof plan; + planV2?: typeof planV2; renderChunk: typeof renderChunk; assemble: typeof assemble; }; @@ -91,6 +102,7 @@ export interface HandlerDeps { // fallow-ignore-next-line complexity export async function dispatch(event: CloudRunEvent, deps?: HandlerDeps): Promise { const unwrapped = unwrapEvent(event); + validatePlanProtocolShape(unwrapped); validateEventGcsUris(unwrapped); logEvent({ event: "handler_start", action: unwrapped.Action, input: summarizeEvent(unwrapped) }); try { @@ -113,9 +125,11 @@ export async function dispatch(event: CloudRunEvent, deps?: HandlerDeps): Promis } } } catch (err) { + normalizeTerminalErrorName(err); logEvent({ event: "handler_error", action: unwrapped.Action, + input: summarizeEvent(unwrapped), message: err instanceof Error ? err.message : String(err), name: err instanceof Error ? err.name : undefined, }); @@ -123,6 +137,57 @@ export async function dispatch(event: CloudRunEvent, deps?: HandlerDeps): Promis } } +// This is the single fail-closed boundary for the wire union. Keeping all +// forbidden locator combinations together makes mixed-protocol input auditable. +// fallow-ignore-next-line complexity +function validatePlanProtocolShape(event: PlanEvent | RenderChunkEvent | AssembleEvent): void { + const raw = event as unknown as Record; + const protocol = raw.PlanProtocol; + if (protocol !== undefined && protocol !== "v1" && protocol !== "v2") { + const error = new Error( + `[handler] unsupported PlanProtocol ${JSON.stringify(protocol)}; expected "v1", "v2", or absent`, + ); + error.name = "PLAN_PROTOCOL_UNSUPPORTED"; + throw error; + } + if (event.Action === "plan") return; + + const hasV1Locator = typeof raw.PlanGcsUri === "string"; + const hasV2Manifest = typeof raw.PlanV2ManifestGcsUri === "string"; + const hasV2Prefix = typeof raw.PlanV2ArtifactGcsPrefix === "string"; + const valid = + protocol === "v2" + ? !hasV1Locator && hasV2Manifest && hasV2Prefix + : hasV1Locator && !hasV2Manifest && !hasV2Prefix; + if (!valid) { + const error = new Error( + `[handler] ${protocol === "v2" ? "v2" : "v1"} ${event.Action} event has mixed or missing plan locators`, + ); + error.name = "PLAN_PROTOCOL_UNSUPPORTED"; + throw error; + } + if (protocol === "v2" && event.Action === "assemble" && event.AudioGcsUri !== null) { + const error = new Error("[handler] v2 assemble audio must be materialized from the manifest"); + error.name = "PLAN_PROTOCOL_UNSUPPORTED"; + throw error; + } +} + +/** Normalize producer error codes to the stable HTTP/workflow discriminator. */ +// The explicit mapping is the public Cloud Workflows retry contract. +// fallow-ignore-next-line complexity +function normalizeTerminalErrorName(error: unknown): void { + if (!error || typeof error !== "object") return; + const candidate = error as { code?: unknown; name?: string }; + if ( + candidate.code === "PLAN_PROTOCOL_UNSUPPORTED" || + candidate.code === "PLAN_TOO_LARGE" || + candidate.code === "PLAN_V2_INTEGRITY_UNRECOVERABLE" + ) { + candidate.name = candidate.code; + } +} + // At most `{Payload: {Input: ...}}` is expected; 4 levels is 2× headroom // and prevents infinite loops on malformed input. const MAX_ENVELOPE_DEPTH = 4; @@ -171,6 +236,8 @@ function logEvent(payload: Record): void { * include the entire project config; we only emit the routable fields * needed to triage a failure from Cloud Logging. */ +// Keep event variants together so Cloud Logging has one redaction boundary. +// fallow-ignore-next-line complexity function summarizeEvent( event: PlanEvent | RenderChunkEvent | AssembleEvent, ): Record { @@ -179,18 +246,25 @@ function summarizeEvent( return { projectGcsUri: event.ProjectGcsUri, planOutputGcsPrefix: event.PlanOutputGcsPrefix, + planProtocol: event.PlanProtocol ?? "v1", format: event.Config.format, fps: event.Config.fps, }; case "renderChunk": return { - planGcsUri: event.PlanGcsUri, + planProtocol: event.PlanProtocol ?? "v1", + ...(event.PlanProtocol === "v2" + ? { planV2ManifestGcsUri: event.PlanV2ManifestGcsUri } + : { planGcsUri: event.PlanGcsUri }), chunkIndex: event.ChunkIndex, format: event.Format, }; case "assemble": return { - planGcsUri: event.PlanGcsUri, + planProtocol: event.PlanProtocol ?? "v1", + ...(event.PlanProtocol === "v2" + ? { planV2ManifestGcsUri: event.PlanV2ManifestGcsUri } + : { planGcsUri: event.PlanGcsUri }), chunkCount: event.ChunkGcsUris.length, hasAudio: event.AudioGcsUri !== null, outputGcsUri: event.OutputGcsUri, @@ -215,6 +289,9 @@ function primeChrome(deps?: HandlerDeps): void { // fallow-ignore-next-line complexity async function handlePlan(event: PlanEvent, deps?: HandlerDeps): Promise { + if (event.PlanProtocol === "v2") { + return handlePlanV2(event, deps); + } const started = Date.now(); const storage = deps?.storage ?? getStorage(); const primitive = deps?.primitives?.plan ?? plan; @@ -274,6 +351,79 @@ async function handlePlan(event: PlanEvent, deps?: HandlerDeps): Promise, + deps?: HandlerDeps, +): Promise> { + const started = Date.now(); + const storage = deps?.storage ?? getStorage(); + const primitive = deps?.primitives?.planV2 ?? planV2; + primeChrome(deps); + + const work = mkdtempSync(join(deps?.tmpRoot ?? tmpdir(), "hf-cr-plan-v2-")); + const projectArchive = join(work, "project.tar.gz"); + const projectDir = join(work, "project"); + const planV2Dir = join(work, "plan-v2"); + try { + await downloadGcsObjectToFile(storage, event.ProjectGcsUri, projectArchive); + await untarDirectory(projectArchive, projectDir); + const result: PlanV2Result = await primitive(projectDir, { ...event.Config }, planV2Dir); + const manifest = readPlanV2Manifest(planV2Dir); + if (manifest.planHash !== result.planHash) { + throwPlanHashMismatch(result.planHash, manifest.planHash); + } + + const outputPrefix = `${trimTrailingSlash(event.PlanOutputGcsPrefix)}/v2`; + const artifactPrefix = `${outputPrefix}/artifacts/sha256`; + const uniqueArtifacts = [ + ...new Map(manifest.artifacts.map((artifact) => [artifact.sha256, artifact])).values(), + ]; + await mapConcurrent(uniqueArtifacts, 16, async (artifact) => { + await uploadContentAddressedFileToGcs( + storage, + planV2BlobPath(planV2Dir, artifact.sha256), + planV2BlobUri(artifactPrefix, artifact.sha256), + artifact.sha256, + ); + }); + + const manifestUri = `${outputPrefix}/manifest.json`; + await uploadContentAddressedFileToGcs( + storage, + result.manifestPath, + manifestUri, + await sha256File(result.manifestPath), + "application/json", + ); + + return { + Action: "plan", + PlanProtocol: "v2", + PlanV2ManifestGcsUri: manifestUri, + PlanV2ArtifactGcsPrefix: artifactPrefix, + PlanHash: result.planHash, + ChunkCount: result.chunkCount, + TotalFrames: result.totalFrames, + Fps: result.fps, + Width: result.width, + Height: result.height, + Format: result.format, + HasAudio: manifest.artifacts.some((artifact) => artifact.path === "audio.aac"), + AudioGcsUri: null, + FfmpegVersion: result.ffmpegVersion, + ProducerVersion: result.producerVersion, + DurationMs: Date.now() - started, + }; + } finally { + cleanupDir(work); + } +} + // ── RenderChunk ───────────────────────────────────────────────────────────── // fallow-ignore-next-line complexity @@ -281,6 +431,9 @@ async function handleRenderChunk( event: RenderChunkEvent, deps?: HandlerDeps, ): Promise { + if (event.PlanProtocol === "v2") { + return handleRenderChunkV2(event, deps); + } const started = Date.now(); const storage = deps?.storage ?? getStorage(); const primitive = deps?.primitives?.renderChunk ?? renderChunk; @@ -331,6 +484,51 @@ async function handleRenderChunk( } } +/** Materialize only this chunk's verified v2 dependencies before rendering. */ +// fallow-ignore-next-line complexity +async function handleRenderChunkV2( + event: Extract, + deps?: HandlerDeps, +): Promise { + const started = Date.now(); + const storage = deps?.storage ?? getStorage(); + const primitive = deps?.primitives?.renderChunk ?? renderChunk; + primeChrome(deps); + + const work = mkdtempSync(join(deps?.tmpRoot ?? tmpdir(), "hf-cr-chunk-v2-")); + try { + const planDir = await downloadAndMaterializePlanV2( + storage, + event, + { role: "chunk", chunkIndex: event.ChunkIndex }, + work, + ); + const chunkOutputBase = join( + work, + event.Format === "png-sequence" + ? `chunk-${pad(event.ChunkIndex)}` + : `chunk-${pad(event.ChunkIndex)}${formatExtension(event.Format)}`, + ); + const result = await primitive(planDir, event.ChunkIndex, chunkOutputBase); + const chunkUri = await uploadChunkOutput( + storage, + result, + event.ChunkOutputGcsPrefix, + event.ChunkIndex, + ); + return { + Action: "renderChunk", + ChunkGcsUri: chunkUri, + ChunkIndex: event.ChunkIndex, + Sha256: result.sha256, + FramesEncoded: result.framesEncoded, + DurationMs: Date.now() - started, + }; + } finally { + cleanupDir(work); + } +} + async function uploadChunkOutput( storage: Storage, result: ChunkResult, @@ -361,6 +559,9 @@ async function handleAssemble( event: AssembleEvent, deps?: HandlerDeps, ): Promise { + if (event.PlanProtocol === "v2") { + return handleAssembleV2(event, deps); + } const started = Date.now(); const storage = deps?.storage ?? getStorage(); const primitive = deps?.primitives?.assemble ?? assemble; @@ -417,6 +618,127 @@ async function handleAssemble( } } +/** + * Materialize the assembler target. Audio is declared assembler-only by the + * v2 manifest and therefore is never downloaded by chunk workers. + */ +// fallow-ignore-next-line complexity +async function handleAssembleV2( + event: Extract, + deps?: HandlerDeps, +): Promise { + const started = Date.now(); + const storage = deps?.storage ?? getStorage(); + const primitive = deps?.primitives?.assemble ?? assemble; + const work = mkdtempSync(join(deps?.tmpRoot ?? tmpdir(), "hf-cr-assemble-v2-")); + try { + const planDir = await downloadAndMaterializePlanV2(storage, event, { role: "assembler" }, work); + const audioPath = existsSync(join(planDir, "audio.aac")) ? join(planDir, "audio.aac") : null; + const chunkPaths = await downloadChunkObjects(storage, event.ChunkGcsUris, work, event.Format); + const finalOutput = + event.Format === "png-sequence" + ? join(work, "output-frames") + : join(work, `output${formatExtension(event.Format)}`); + const result = await primitive(planDir, chunkPaths, audioPath, finalOutput, { + cfr: event.Cfr === true, + }); + if (event.Format === "png-sequence") { + const tarball = `${finalOutput}.tar.gz`; + await tarDirectory(finalOutput, tarball); + await uploadFileToGcs(storage, tarball, event.OutputGcsUri, "application/gzip"); + } else { + await uploadFileToGcs(storage, finalOutput, event.OutputGcsUri); + } + return { + Action: "assemble", + OutputGcsUri: event.OutputGcsUri, + FramesEncoded: result.framesEncoded, + FileSize: result.fileSize, + DurationMs: Date.now() - started, + }; + } finally { + cleanupDir(work); + } +} + +async function downloadAndMaterializePlanV2( + storage: Storage, + event: { + PlanV2ManifestGcsUri: string; + PlanV2ArtifactGcsPrefix: string; + PlanHash: string; + }, + target: PlanV2MaterializationTarget, + work: string, +): Promise { + const transportDir = join(work, "plan-v2"); + mkdirSync(transportDir, { recursive: true }); + await downloadGcsObjectToFile( + storage, + event.PlanV2ManifestGcsUri, + join(transportDir, "plan.json"), + ); + const manifest = readPlanV2Manifest(transportDir); + if (manifest.planHash !== event.PlanHash) { + throwPlanHashMismatch(event.PlanHash, manifest.planHash); + } + const artifacts = listPlanV2ArtifactsForTarget(manifest, target); + const uniqueArtifacts = [ + ...new Map(artifacts.map((artifact) => [artifact.sha256, artifact])).values(), + ]; + await mapConcurrent(uniqueArtifacts, 16, async (artifact) => { + await downloadPlanV2Artifact(storage, event.PlanV2ArtifactGcsPrefix, transportDir, artifact); + }); + const planDir = join(work, "plan"); + materializePlanV2Target(transportDir, target, planDir); + return planDir; +} + +async function downloadPlanV2Artifact( + storage: Storage, + artifactPrefix: string, + planV2Dir: string, + artifact: Readonly, +): Promise { + await downloadGcsObjectToFileVerified( + storage, + planV2BlobUri(artifactPrefix, artifact.sha256), + planV2BlobPath(planV2Dir, artifact.sha256), + artifact.sha256, + ); +} + +function planV2BlobPath(planV2Dir: string, digest: string): string { + return join(planV2Dir, "artifacts", "sha256", digest.slice(0, 2), digest); +} + +function planV2BlobUri(prefix: string, digest: string): string { + return `${trimTrailingSlash(prefix)}/${digest.slice(0, 2)}/${digest}`; +} + +function throwPlanHashMismatch(expected: string, actual: string): never { + const error = new Error( + `PLAN_HASH_MISMATCH: event PlanHash=${expected} did not match v2 manifest planHash=${actual}`, + ); + error.name = "PLAN_HASH_MISMATCH"; + throw error; +} + +async function mapConcurrent( + values: readonly T[], + concurrency: number, + fn: (value: T) => Promise, +): Promise { + let cursor = 0; + async function worker(): Promise { + while (cursor < values.length) { + const index = cursor++; + await fn(values[index]!); + } + } + await Promise.all(Array.from({ length: Math.min(concurrency, values.length) }, () => worker())); +} + async function downloadChunkObjects( storage: Storage, uris: string[], @@ -454,15 +776,21 @@ async function downloadChunkObjects( // ── Helpers ───────────────────────────────────────────────────────────────── /** Collect every GCS URI that the handler will touch for a given event. */ +// This exhaustive event projection is the bucket-allowlist security boundary. +// fallow-ignore-next-line complexity function getEventGcsUris(event: PlanEvent | RenderChunkEvent | AssembleEvent): string[] { switch (event.Action) { case "plan": return [event.ProjectGcsUri, event.PlanOutputGcsPrefix]; case "renderChunk": - return [event.PlanGcsUri, event.ChunkOutputGcsPrefix]; + return event.PlanProtocol === "v2" + ? [event.PlanV2ManifestGcsUri, event.PlanV2ArtifactGcsPrefix, event.ChunkOutputGcsPrefix] + : [event.PlanGcsUri, event.ChunkOutputGcsPrefix]; case "assemble": return [ - event.PlanGcsUri, + ...(event.PlanProtocol === "v2" + ? [event.PlanV2ManifestGcsUri, event.PlanV2ArtifactGcsPrefix] + : [event.PlanGcsUri]), ...event.ChunkGcsUris, event.OutputGcsUri, event.AudioGcsUri, @@ -578,6 +906,9 @@ const NON_RETRYABLE_ERROR_NAMES = new Set([ // Handler-boundary guards. "GCS_URI_NOT_ALLOWED", "PLAN_HASH_MISMATCH", + "PLAN_ARTIFACT_DIGEST_MISMATCH", + "PLAN_PROTOCOL_UNSUPPORTED", + "PLAN_V2_INTEGRITY_UNRECOVERABLE", // Producer error class names (`.name`) + their string code aliases — the // class sets `.name` to the class name but wraps a `code`; cover both so a // raw-code throw is caught too. Mirrors the AWS state machine's @@ -585,11 +916,11 @@ const NON_RETRYABLE_ERROR_NAMES = new Set([ "FormatNotSupportedInDistributedError", "PlanTooLargeError", "PlanProtocolUnsupportedError", + "PlanV2IntegrityError", "RenderChunkValidationError", "FFMPEG_VERSION_MISMATCH", "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED", "PLAN_TOO_LARGE", - "PLAN_PROTOCOL_UNSUPPORTED", "BROWSER_GPU_NOT_SOFTWARE", "FONT_FETCH_FAILED", "ChromeBinaryUnavailableError", diff --git a/packages/gcp-cloud-run/terraform/main.tf b/packages/gcp-cloud-run/terraform/main.tf index a0bb978697..460cb42746 100644 --- a/packages/gcp-cloud-run/terraform/main.tf +++ b/packages/gcp-cloud-run/terraform/main.tf @@ -133,6 +133,11 @@ resource "google_workflows_workflow" "render" { region = var.region service_account = google_service_account.workflow_sa.id source_contents = file(local.workflow_source) + # Do not publish an executable workflow until its identity has permission to + # reach Cloud Run. IAM propagation remains eventually consistent, so the + # workflow also retries the edge's transient 403 response with bounded + # backoff. + depends_on = [google_cloud_run_v2_service_iam_member.workflow_invokes_run] # Allow `terraform destroy` to remove the workflow without a manual step; # the definition is reproducible from this module. deletion_protection = false diff --git a/packages/gcp-cloud-run/terraform/outputs.tf b/packages/gcp-cloud-run/terraform/outputs.tf index f4b4ac9a78..a58d0ac809 100644 --- a/packages/gcp-cloud-run/terraform/outputs.tf +++ b/packages/gcp-cloud-run/terraform/outputs.tf @@ -3,6 +3,16 @@ output "render_bucket_name" { value = google_storage_bucket.render.name } +output "project_name" { + description = "Resource prefix used by this deployment." + value = var.project_name +} + +output "render_service_name" { + description = "Cloud Run service name." + value = google_cloud_run_v2_service.render.name +} + output "service_url" { description = "HTTPS URL of the Cloud Run render service. Pass as renderToCloudRun({ serviceUrl })." value = google_cloud_run_v2_service.render.uri diff --git a/packages/gcp-cloud-run/terraform/smoke-safety.test.ts b/packages/gcp-cloud-run/terraform/smoke-safety.test.ts new file mode 100644 index 0000000000..9b8bd0248d --- /dev/null +++ b/packages/gcp-cloud-run/terraform/smoke-safety.test.ts @@ -0,0 +1,71 @@ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "bun:test"; + +const smokePath = join(import.meta.dir, "../../../examples/gcp-cloud-run/scripts/smoke.sh"); +const smoke = readFileSync(smokePath, "utf-8"); +const dockerfile = readFileSync(join(import.meta.dir, "../Dockerfile"), "utf-8"); + +describe("GCP smoke ownership and protocol safety", () => { + it("defaults to v1 and requires an explicit v2 protocol argument", () => { + expect(smoke).toContain('PROTOCOLS="${PROTOCOLS:-v1}"'); + expect(smoke).toContain("--protocols)"); + expect(smoke).toContain("PlanProtocol: $protocol"); + expect(smoke).toContain("decodedFramesEqual"); + expect(smoke).toContain("decodedAudioEqual"); + expect(smoke).toContain("normalizedMetadataEqual"); + }); + + it("derives a length-safe owner prefix and isolates Terraform state", () => { + expect(smoke).toContain('OWNER_HASH="$(printf'); + expect(smoke).toContain("RUN_NONCE="); + expect(smoke).toContain('STACK_NAME="hf-smoke-$OWNER_HASH"'); + expect(smoke).toContain('TF_WORK_DIR="$ARTIFACT_DIR/terraform"'); + expect(smoke).toContain('TF_DATA_DIR="$ARTIFACT_DIR/terraform-data"'); + expect(smoke).toContain("export TF_DATA_DIR"); + expect(smoke).toContain('-var "project_name=$STACK_NAME"'); + expect(smoke).toContain('[ "$STACK_NAME" != "hyperframes" ]'); + }); + + it("tracks owned registry resources and verifies stack deletion", () => { + expect(smoke).toContain("CREATED_IMAGE=0"); + expect(smoke).toContain("CREATED_REPO=0"); + expect(smoke).toContain('if [ "$CREATED_REPO" -eq 1 ]'); + expect(smoke).toContain('if [ "$CREATED_IMAGE" -eq 1 ]'); + + for (const resource of [ + "cloud-run-service", + "workflow", + "render-bucket", + "artifact-image", + "artifact-repository", + "cloud-build-staging-bucket", + ]) { + expect(smoke).toContain(`verify_absent "${resource}"`); + } + expect(smoke).toContain('verify_service_account_absent "run-service-account"'); + expect(smoke).toContain('verify_service_account_absent "workflow-service-account"'); + expect(smoke).toContain('verify_absent "preflight-cloud-run-service"'); + expect(smoke).toContain('verify_absent "preflight-artifact-image"'); + expect(smoke).toContain("$STACK_NAME-render"); + expect(smoke).toContain("--ignore-file"); + expect(smoke).toContain("--gcs-source-staging-dir"); + expect(smoke).toContain("!scripts/package-subpaths.mjs"); + expect(dockerfile).toContain("COPY scripts/package-subpaths.mjs scripts/package-subpaths.mjs"); + expect(smoke).not.toContain("gcloud services enable"); + expect(smoke).toContain("gcloud services list"); + expect(smoke).toContain("--enabled"); + expect(smoke).not.toContain("gcloud services describe"); + expect(smoke).toContain("cannot find"); + expect(smoke).toContain("gcloud iam service-accounts list"); + expect(smoke).toContain('--filter "email:$email"'); + }); + + it("does not swallow Terraform cleanup failures", () => { + const cleanupStart = smoke.indexOf("cleanup() {"); + const cleanupEnd = smoke.indexOf("\ntrap cleanup EXIT", cleanupStart); + const cleanup = smoke.slice(cleanupStart, cleanupEnd); + expect(cleanup).not.toContain("|| true"); + expect(cleanup).toContain("exit 7"); + }); +}); diff --git a/packages/gcp-cloud-run/terraform/variables.tf b/packages/gcp-cloud-run/terraform/variables.tf index 5b9a41b576..80deb886f0 100644 --- a/packages/gcp-cloud-run/terraform/variables.tf +++ b/packages/gcp-cloud-run/terraform/variables.tf @@ -13,6 +13,15 @@ variable "project_name" { type = string description = "Name prefix applied to the service / workflow / bucket / service accounts." default = "hyperframes" + + validation { + condition = ( + length(var.project_name) >= 3 && + length(var.project_name) <= 23 && + can(regex("^[a-z][a-z0-9-]*[a-z0-9]$", var.project_name)) + ) + error_message = "project_name must be 3-23 lowercase letters, digits, or hyphens, begin with a letter, and end with a letter or digit so derived service-account IDs remain valid." + } } variable "image" { diff --git a/packages/gcp-cloud-run/terraform/workflow.test.ts b/packages/gcp-cloud-run/terraform/workflow.test.ts new file mode 100644 index 0000000000..b17e00772a --- /dev/null +++ b/packages/gcp-cloud-run/terraform/workflow.test.ts @@ -0,0 +1,148 @@ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { describe, expect, it } from "bun:test"; +import { parse } from "yaml"; + +type Step = Record; + +const source = readFileSync(join(import.meta.dir, "workflow.yaml"), "utf-8"); +// Cloud Workflows expressions are valid to Google's parser but `${...}` +// inside YAML flow collections is not valid generic YAML. Quote expressions +// for structural parsing while preserving their text for contract assertions. +const parseableSource = source.replace(/\$\{([^}]*)\}/g, (_match, expression: string) => + JSON.stringify(`\${${expression}}`), +); +const workflow = parse(parseableSource) as { + main: { + steps: Step[]; + }; + retryable: { + steps: Step[]; + }; +}; + +function namedStep(name: string, steps = workflow.main.steps): Record { + for (const step of steps) { + if (name in step) return step[name] as Record; + } + throw new Error(`missing workflow step ${name}`); +} + +function requestBody(stepName: string): Record { + const step = namedStep(stepName); + const attempt = step.try as { + args: { + body: Record; + }; + }; + return attempt.args.body; +} + +function requestAuth(stepName: string): Record { + const step = namedStep(stepName); + const attempt = step.try as { + args: { + auth: Record; + }; + }; + return attempt.args.auth; +} + +function chunkRequestBody(stepName: string): Record { + const renderChunks = namedStep("renderChunks"); + const parallel = renderChunks.parallel as { + for: { + steps: Step[]; + }; + }; + const step = namedStep(stepName, parallel.for.steps); + const attempt = step.try as { + args: { + body: Record; + }; + }; + return attempt.args.body; +} + +describe("Cloud Workflows plan protocol routing", () => { + it("pins OIDC tokens to the Cloud Run root and retries IAM propagation", () => { + for (const stepName of ["planV1", "planV2", "assembleV1", "assembleV2"]) { + expect(requestAuth(stepName)).toEqual({ + type: "OIDC", + audience: "${serviceUrl}", + }); + } + expect(JSON.stringify(workflow.retryable)).toContain("e.code == 403"); + expect(source.match(/max_retries: 6/g)).toHaveLength(2); + expect(source.match(/max_retries: 4/g)).toHaveLength(4); + }); + + it("keeps v1 as the default and rejects unknown protocols before plan", () => { + expect(source).toContain('default(map.get(args, "PlanProtocol"), "v1")'); + expect(namedStep("selectPlanProtocol")).toMatchObject({ + next: "unsupportedPlanProtocol", + }); + expect(namedStep("unsupportedPlanProtocol")).toMatchObject({ + raise: { + code: "PLAN_PROTOCOL_UNSUPPORTED", + }, + }); + }); + + it("uses disjoint v1 and v2 plan request/response contracts", () => { + expect(requestBody("planV1")).toMatchObject({ + Action: "plan", + PlanProtocol: "v1", + }); + expect(requestBody("planV2")).toMatchObject({ + Action: "plan", + PlanProtocol: "v2", + }); + const validation = JSON.stringify(namedStep("validatePlanResult")); + expect(validation).toContain("PlanGcsUri"); + expect(validation).toContain("PlanV2ManifestGcsUri"); + expect(validation).toContain("PlanV2ArtifactGcsPrefix"); + expect(source).toContain('not("PlanGcsUri" in planResult)'); + expect(source).toContain( + '(not("PlanProtocol" in planResult) or planResult.PlanProtocol == "v1")', + ); + }); + + it("never mixes v1 and v2 chunk locators", () => { + const v1 = chunkRequestBody("renderOneChunkV1"); + expect(v1).toMatchObject({ + Action: "renderChunk", + PlanProtocol: "v1", + }); + expect(v1).toHaveProperty("PlanGcsUri"); + expect(v1).not.toHaveProperty("PlanV2ManifestGcsUri"); + expect(v1).not.toHaveProperty("PlanV2ArtifactGcsPrefix"); + + const v2 = chunkRequestBody("renderOneChunkV2"); + expect(v2).toMatchObject({ + Action: "renderChunk", + PlanProtocol: "v2", + }); + expect(v2).not.toHaveProperty("PlanGcsUri"); + expect(v2).toHaveProperty("PlanV2ManifestGcsUri"); + expect(v2).toHaveProperty("PlanV2ArtifactGcsPrefix"); + expect(v2).toHaveProperty("PlanHash"); + }); + + it("never mixes v1 and v2 assembler locators", () => { + const v1 = requestBody("assembleV1"); + expect(v1).toHaveProperty("PlanGcsUri"); + expect(v1).not.toHaveProperty("PlanV2ManifestGcsUri"); + expect(v1).not.toHaveProperty("PlanV2ArtifactGcsPrefix"); + + const v2 = requestBody("assembleV2"); + expect(v2).not.toHaveProperty("PlanGcsUri"); + expect(v2).toHaveProperty("PlanV2ManifestGcsUri"); + expect(v2).toHaveProperty("PlanV2ArtifactGcsPrefix"); + expect(v2).toHaveProperty("PlanHash"); + expect(v2).toMatchObject({ + PlanProtocol: "v2", + AudioGcsUri: null, + }); + }); +}); diff --git a/packages/gcp-cloud-run/terraform/workflow.yaml b/packages/gcp-cloud-run/terraform/workflow.yaml index 072cd197fe..7901a5a2be 100644 --- a/packages/gcp-cloud-run/terraform/workflow.yaml +++ b/packages/gcp-cloud-run/terraform/workflow.yaml @@ -26,9 +26,23 @@ main: - planOutputGcsPrefix: ${args.PlanOutputGcsPrefix} - outputGcsUri: ${args.OutputGcsUri} - config: ${args.Config} + # Backward-compatible default. v2 is accepted only through an + # explicit top-level PlanProtocol opt-in. + - planProtocol: ${default(map.get(args, "PlanProtocol"), "v1")} # ── Plan (Activity A) ──────────────────────────────────────────────────── - - plan: + - selectPlanProtocol: + switch: + - condition: ${planProtocol == "v1"} + next: planV1 + - condition: ${planProtocol == "v2"} + next: planV2 + next: unsupportedPlanProtocol + - unsupportedPlanProtocol: + raise: + code: PLAN_PROTOCOL_UNSUPPORTED + message: ${"PlanProtocol must be v1 or v2; got " + string(planProtocol)} + - planV1: try: call: http.post args: @@ -36,22 +50,69 @@ main: timeout: 1800 auth: type: OIDC + audience: ${serviceUrl} body: Action: plan + PlanProtocol: v1 ProjectGcsUri: ${projectGcsUri} PlanOutputGcsPrefix: ${planOutputGcsPrefix} Config: ${config} - result: planResp + result: planRespV1 retry: predicate: ${retryable} - max_retries: 4 + max_retries: 6 backoff: initial_delay: 2 max_delay: 60 multiplier: 2 - - capturePlan: + next: capturePlanV1 + - capturePlanV1: + assign: + - planResult: ${planRespV1.body} + next: validatePlanResult + - planV2: + try: + call: http.post + args: + url: ${serviceUrl} + timeout: 1800 + auth: + type: OIDC + audience: ${serviceUrl} + body: + Action: plan + PlanProtocol: v2 + ProjectGcsUri: ${projectGcsUri} + PlanOutputGcsPrefix: ${planOutputGcsPrefix} + Config: ${config} + result: planRespV2 + retry: + predicate: ${retryable} + max_retries: 6 + backoff: + initial_delay: 2 + max_delay: 60 + multiplier: 2 + next: capturePlanV2 + - capturePlanV2: + assign: + - planResult: ${planRespV2.body} + next: validatePlanResult + - validatePlanResult: + # Fail closed before fan-out. A v2 render may never fall back to a + # v1 PlanGcsUri, and a v1 render may never consume v2 locators. + switch: + - condition: ${planProtocol == "v1" and (not("PlanProtocol" in planResult) or planResult.PlanProtocol == "v1") and ("PlanGcsUri" in planResult) and not("PlanV2ManifestGcsUri" in planResult) and not("PlanV2ArtifactGcsPrefix" in planResult)} + next: captureChunkCount + - condition: ${planProtocol == "v2" and ("PlanProtocol" in planResult) and planResult.PlanProtocol == "v2" and ("PlanV2ManifestGcsUri" in planResult) and ("PlanV2ArtifactGcsPrefix" in planResult) and not("PlanGcsUri" in planResult)} + next: captureChunkCount + next: planProtocolLocatorMismatch + - planProtocolLocatorMismatch: + raise: + code: PLAN_PROTOCOL_LOCATOR_MISMATCH + message: "Plan response did not match the selected protocol's disjoint locator contract." + - captureChunkCount: assign: - - planResult: ${planResp.body} - chunkCount: ${planResult.ChunkCount} # ── BuildChunkList + AssertChunkCount ────────────────────────────────────── @@ -98,7 +159,12 @@ main: value: idx in: ${chunkIndexes} steps: - - renderOneChunk: + - selectChunkProtocol: + switch: + - condition: ${planProtocol == "v2"} + next: renderOneChunkV2 + next: renderOneChunkV1 + - renderOneChunkV1: try: call: http.post args: @@ -106,8 +172,10 @@ main: timeout: 1800 auth: type: OIDC + audience: ${serviceUrl} body: Action: renderChunk + PlanProtocol: v1 ChunkIndex: ${idx} PlanGcsUri: ${planResult.PlanGcsUri} PlanHash: ${planResult.PlanHash} @@ -121,13 +189,46 @@ main: initial_delay: 2 max_delay: 60 multiplier: 2 + next: storeChunk + - renderOneChunkV2: + try: + call: http.post + args: + url: ${serviceUrl} + timeout: 1800 + auth: + type: OIDC + audience: ${serviceUrl} + body: + Action: renderChunk + PlanProtocol: v2 + ChunkIndex: ${idx} + PlanV2ManifestGcsUri: ${planResult.PlanV2ManifestGcsUri} + PlanV2ArtifactGcsPrefix: ${planResult.PlanV2ArtifactGcsPrefix} + PlanHash: ${planResult.PlanHash} + ChunkOutputGcsPrefix: ${planOutputGcsPrefix} + Format: ${planResult.Format} + result: chunkResp + retry: + predicate: ${retryable} + max_retries: 4 + backoff: + initial_delay: 2 + max_delay: 60 + multiplier: 2 + next: storeChunk - storeChunk: assign: - chunkUris[idx]: ${chunkResp.body.ChunkGcsUri} - chunkResults[idx]: ${chunkResp.body} # ── Assemble (Activity C) ────────────────────────────────────────────────── - - assemble: + - selectAssembleProtocol: + switch: + - condition: ${planProtocol == "v2"} + next: assembleV2 + next: assembleV1 + - assembleV1: try: call: http.post args: @@ -135,8 +236,10 @@ main: timeout: 1800 auth: type: OIDC + audience: ${serviceUrl} body: Action: assemble + PlanProtocol: v1 PlanGcsUri: ${planResult.PlanGcsUri} ChunkGcsUris: ${chunkUris} AudioGcsUri: ${planResult.AudioGcsUri} @@ -154,6 +257,38 @@ main: initial_delay: 2 max_delay: 60 multiplier: 2 + next: done + - assembleV2: + try: + call: http.post + args: + url: ${serviceUrl} + timeout: 1800 + auth: + type: OIDC + audience: ${serviceUrl} + body: + Action: assemble + PlanProtocol: v2 + PlanV2ManifestGcsUri: ${planResult.PlanV2ManifestGcsUri} + PlanV2ArtifactGcsPrefix: ${planResult.PlanV2ArtifactGcsPrefix} + PlanHash: ${planResult.PlanHash} + ChunkGcsUris: ${chunkUris} + # Audio is an assembler-scoped v2 artifact and is materialized + # from the manifest, never carried through a v1 AudioGcsUri. + AudioGcsUri: null + OutputGcsUri: ${outputGcsUri} + Format: ${planResult.Format} + Cfr: ${("cfr" in config) and config.cfr} + result: assembleResp + retry: + predicate: ${retryable} + max_retries: 4 + backoff: + initial_delay: 2 + max_delay: 60 + multiplier: 2 + next: done - done: return: @@ -161,9 +296,11 @@ main: Chunks: ${chunkResults} Assemble: ${assembleResp.body} -# Retry predicate: retry transient/server failures (429 + 5xx), never the -# handler's non-retryable 400s (bad input, plan-hash mismatch, unsupported -# format, …). Connection / timeout errors carry no `.code`; retry those too. +# Retry predicate: retry transient/server failures (403 from Cloud Run IAM +# propagation, 429, and 5xx), never the handler's non-retryable 400s (bad +# input, plan-hash mismatch, unsupported format, …). The handler does not emit +# 403, so that status is always from Cloud Run's authentication edge. +# Connection / timeout errors carry no `.code`; retry those too. retryable: params: [e] steps: @@ -173,6 +310,8 @@ retryable: return: true - condition: ${e.code == 429} return: true + - condition: ${e.code == 403} + return: true - condition: ${e.code >= 500 and e.code < 600} return: true - nonRetryable: