fix(producer): validate distributed video metadata

Author: jrusso1020

examples/aws-lambda/template.yaml

@@ -263,6 +263,8 @@ Resources:
                   - PlanTooLargeError
                   - PLAN_PROTOCOL_UNSUPPORTED
                   - PlanProtocolUnsupportedError
+                  - VIDEO_SOURCE_UNRENDERABLE
+                  - INVALID_VIDEO_METADATA
                   - PLAN_ARTIFACT_DIGEST_MISMATCH
                   - FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED
                 MaxAttempts: 0
@@ -305,6 +307,8 @@ Resources:
                   - PLAN_PROTOCOL_UNSUPPORTED
                   - PlanProtocolUnsupportedError
                   - PLAN_V2_INTEGRITY_UNRECOVERABLE
+                  - VIDEO_SOURCE_UNRENDERABLE
+                  - INVALID_VIDEO_METADATA
                   - PlanV2IntegrityError
                   - PLAN_ARTIFACT_DIGEST_MISMATCH
                   - FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED
@@ -401,6 +405,7 @@ Resources:
                         - PlanTooLargeError
                         - PLAN_PROTOCOL_UNSUPPORTED
                         - PlanProtocolUnsupportedError
+                        - INVALID_VIDEO_METADATA
                         - PLAN_ARTIFACT_DIGEST_MISMATCH
                       MaxAttempts: 0
                     - ErrorEquals: [States.ALL]
@@ -497,6 +502,7 @@ Resources:
                         - PLAN_PROTOCOL_UNSUPPORTED
                         - PlanProtocolUnsupportedError
                         - PLAN_V2_INTEGRITY_UNRECOVERABLE
+                        - INVALID_VIDEO_METADATA
                         - PlanV2IntegrityError
                         - PLAN_ARTIFACT_DIGEST_MISMATCH
                         - ChromeBinaryUnavailableError

packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts

@@ -76,6 +76,8 @@ const EXPECTED_NON_RETRYABLE_ERRORS = new Set([
   "PLAN_PROTOCOL_UNSUPPORTED",
   "PlanProtocolUnsupportedError",
   "PLAN_V2_INTEGRITY_UNRECOVERABLE",
+  "VIDEO_SOURCE_UNRENDERABLE",
+  "INVALID_VIDEO_METADATA",
   "PlanV2IntegrityError",
   "PLAN_ARTIFACT_DIGEST_MISMATCH",
   "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED",
@@ -209,6 +211,25 @@ describe("HyperframesRenderStack — snapshot", () => {
     }
   });
 
+  it("routes video failures consistently across SAM/CDK and both plan protocols", () => {
+    for (const definition of [SYNTHED.definition, readSamDefinition()]) {
+      const v1 = getV1TaskStates(definition);
+      const v2 = getV2TaskStates(definition);
+      for (const planState of [v1.Plan, v2.PlanV2]) {
+        const errors = new Set<string>();
+        collectNonRetryableErrors(planState, errors);
+        expect(errors.has("VIDEO_SOURCE_UNRENDERABLE")).toBe(true);
+        expect(errors.has("INVALID_VIDEO_METADATA")).toBe(true);
+        expect(errors.has("VIDEO_EXTRACTION_FAILED")).toBe(false);
+      }
+      for (const chunkState of [v1.RenderChunk, v2.RenderChunkV2]) {
+        const errors = new Set<string>();
+        collectNonRetryableErrors(chunkState, errors);
+        expect(errors.has("INVALID_VIDEO_METADATA")).toBe(true);
+      }
+    }
+  });
+
   it("keeps v1 and v2 locators disjoint across orchestration branches", () => {
     const { definition } = SYNTHED;
     const v1 = JSON.stringify({
@@ -271,6 +292,21 @@ function getV2TaskStates(definition: {
   };
 }
 
+function getV1TaskStates(definition: {
+  States: Record<string, unknown>;
+}): Record<"Plan" | "RenderChunk" | "Assemble", unknown> {
+  const renderChunks = requireRecord(definition.States.RenderChunks, "RenderChunks state");
+  const processor = isRecord(renderChunks.Iterator)
+    ? renderChunks.Iterator
+    : requireRecord(renderChunks.ItemProcessor, "RenderChunks processor");
+  const innerStates = requireRecord(processor.States, "RenderChunks processor states");
+  return {
+    Plan: definition.States.Plan,
+    RenderChunk: innerStates.RenderChunk,
+    Assemble: definition.States.Assemble,
+  };
+}
+
 function readSamDefinition(): { States: Record<string, unknown> } {
   const source = readFileSync(
     new URL("../../../../examples/aws-lambda/template.yaml", import.meta.url),

packages/aws-lambda/src/cdk/HyperframesRenderStack.ts

@@ -204,6 +204,8 @@ export class HyperframesRenderStack extends Construct {
       "PLAN_PROTOCOL_UNSUPPORTED",
       "PlanProtocolUnsupportedError",
       "PLAN_V2_INTEGRITY_UNRECOVERABLE",
+      "VIDEO_SOURCE_UNRENDERABLE",
+      "INVALID_VIDEO_METADATA",
       "PlanV2IntegrityError",
       "PLAN_ARTIFACT_DIGEST_MISMATCH",
       "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED",
@@ -219,6 +221,7 @@ export class HyperframesRenderStack extends Construct {
       "PLAN_PROTOCOL_UNSUPPORTED",
       "PlanProtocolUnsupportedError",
       "PLAN_V2_INTEGRITY_UNRECOVERABLE",
+      "INVALID_VIDEO_METADATA",
       "PlanV2IntegrityError",
       "PLAN_ARTIFACT_DIGEST_MISMATCH",
       "ChromeBinaryUnavailableError",

packages/aws-lambda/src/handler.test.ts

@@ -22,6 +22,7 @@ import { tmpdir } from "node:os";
 import { dirname, join } from "node:path";
 import {
   CURRENT_PLAN_PROTOCOL,
+  PlanVideosMetadataError,
   type AssembleResult,
   type ChunkResult,
   type PlanResult,
@@ -243,19 +244,25 @@ describe("handler dispatch", () => {
     ).toBe(true);
   });
 
-  it("normalizes producer terminal codes to Step Functions error names", async () => {
+  it("normalizes producer workflow codes to Step Functions error names", async () => {
     for (const code of [
       "PLAN_TOO_LARGE",
       "PLAN_PROTOCOL_UNSUPPORTED",
       "PLAN_V2_INTEGRITY_UNRECOVERABLE",
+      "VIDEO_SOURCE_UNRENDERABLE",
+      "VIDEO_EXTRACTION_FAILED",
+      "INVALID_VIDEO_METADATA",
     ] as const) {
       const tmpRoot = makeTmpRoot();
       const s3 = new FakeS3Client();
       s3.objects.set("s3://bucket/project.tar.gz", await makeMinimalProjectTar());
-      const terminal = Object.assign(new Error(`terminal: ${code}`), {
-        code,
-        name: "ProducerError",
-      });
+      const terminal =
+        code === "INVALID_VIDEO_METADATA"
+          ? new PlanVideosMetadataError("test invalid plan video metadata")
+          : Object.assign(new Error(`terminal: ${code}`), {
+              code,
+              name: "ProducerError",
+            });
 
       await expect(
         handler(

packages/aws-lambda/src/handler.ts

@@ -138,7 +138,7 @@ export async function handler(event: LambdaEvent, deps?: HandlerDeps): Promise<L
 
 /**
  * AWS Lambda reports `Error.name` to Step Functions, while producer errors
- * expose stable machine codes separately. Normalize the terminal codes
+ * expose stable machine codes separately. Normalize workflow-facing codes
  * whose historical class names differ from their orchestration contracts.
  */
 // The explicit error-name mapping is the public Step Functions failure contract.
@@ -149,7 +149,10 @@ function normalizeTerminalErrorName(error: unknown): void {
   if (
     candidate.code === "PLAN_PROTOCOL_UNSUPPORTED" ||
     candidate.code === "PLAN_TOO_LARGE" ||
-    candidate.code === "PLAN_V2_INTEGRITY_UNRECOVERABLE"
+    candidate.code === "PLAN_V2_INTEGRITY_UNRECOVERABLE" ||
+    candidate.code === "VIDEO_SOURCE_UNRENDERABLE" ||
+    candidate.code === "VIDEO_EXTRACTION_FAILED" ||
+    candidate.code === "INVALID_VIDEO_METADATA"
   ) {
     candidate.name = candidate.code;
   }

packages/gcp-cloud-run/src/server.test.ts

@@ -22,6 +22,7 @@ import { dirname, join } from "node:path";
 import {
   CURRENT_PLAN_PROTOCOL,
   PLAN_V2_INTEGRITY_UNRECOVERABLE,
+  PlanVideosMetadataError,
   PlanV2IntegrityError,
   PlanProtocolUnsupportedError,
   type AssembleResult,
@@ -574,6 +575,66 @@ describe("createApp HTTP mapping", () => {
     }
   });
 
+  it.each([
+    ["VIDEO_SOURCE_UNRENDERABLE", 400],
+    ["VIDEO_EXTRACTION_FAILED", 500],
+  ] as const)("routes producer video code %s with HTTP %s", async (code, expectedStatus) => {
+    const gcs = new FakeGcs();
+    await seedPlanTar(gcs, "gs://b/renders/r1/plan.tar.gz", PLAN_HASH);
+    const app = createApp(
+      depsWith(gcs, {
+        renderChunk: async () => {
+          throw Object.assign(new Error(`test ${code}`), {
+            name: "ProducerError",
+            code,
+          });
+        },
+      }),
+    );
+    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(expectedStatus);
+    const body = (await res.json()) as { error: string };
+    expect(body.error).toBe(code);
+  });
+
+  it("routes the real plan metadata error as non-retryable", async () => {
+    const gcs = new FakeGcs();
+    await seedProjectTar(gcs, "gs://b/sites/invalid-video-metadata/project.tar.gz");
+    const app = createApp(
+      depsWith(gcs, {
+        plan: async () => {
+          throw new PlanVideosMetadataError("test invalid plan video metadata");
+        },
+      }),
+    );
+    const res = await app.request("/", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({
+        Action: "plan",
+        ProjectGcsUri: "gs://b/sites/invalid-video-metadata/project.tar.gz",
+        PlanOutputGcsPrefix: "gs://b/renders/invalid-video-metadata/",
+        Config: { fps: 30, width: 640, height: 360, format: "mp4" },
+      }),
+    });
+
+    expect(res.status).toBe(400);
+    const body = (await res.json()) as { error: string };
+    expect(body.error).toBe("INVALID_VIDEO_METADATA");
+  });
+
   it("returns 500 for a retryable/unknown error", async () => {
     const gcs = new FakeGcs(); // plan tar NOT seeded → download fails (retryable)
     const app = createApp(depsWith(gcs));

packages/gcp-cloud-run/src/server.ts

@@ -182,7 +182,10 @@ function normalizeTerminalErrorName(error: unknown): void {
   if (
     candidate.code === "PLAN_PROTOCOL_UNSUPPORTED" ||
     candidate.code === "PLAN_TOO_LARGE" ||
-    candidate.code === "PLAN_V2_INTEGRITY_UNRECOVERABLE"
+    candidate.code === "PLAN_V2_INTEGRITY_UNRECOVERABLE" ||
+    candidate.code === "VIDEO_SOURCE_UNRENDERABLE" ||
+    candidate.code === "VIDEO_EXTRACTION_FAILED" ||
+    candidate.code === "INVALID_VIDEO_METADATA"
   ) {
     candidate.name = candidate.code;
   }
@@ -899,6 +902,8 @@ const NON_RETRYABLE_ERROR_NAMES = new Set([
   "PLAN_ARTIFACT_DIGEST_MISMATCH",
   "PLAN_PROTOCOL_UNSUPPORTED",
   "PLAN_V2_INTEGRITY_UNRECOVERABLE",
+  "VIDEO_SOURCE_UNRENDERABLE",
+  "INVALID_VIDEO_METADATA",
   // 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

packages/producer/src/distributed.ts

@@ -94,6 +94,7 @@ export {
   type EffectiveChunkResult,
   // Error codes + classes
   FFMPEG_VERSION_MISMATCH,
+  INVALID_VIDEO_METADATA,
   PLAN_HASH_MISMATCH,
   RenderChunkValidationError,
 } from "./services/distributed/renderChunk.js";
@@ -142,7 +143,7 @@ export {
 // ── Format union ────────────────────────────────────────────────────────────
 // Canonical output-format type. The aws-lambda package re-exports it so
 // CLI / adopter SDKs can derive runtime allowlists from one source.
-export type { DistributedFormat } from "./services/distributed/shared.js";
+export { PlanVideosMetadataError, type DistributedFormat } from "./services/distributed/shared.js";
 
 // ── Plan-time shared types from `freezePlan` ───────────────────────────────
 // Re-exported so adopters that deserialize a planDir's `meta/encoder.json`

packages/producer/src/server.errorCode.test.ts

@@ -19,6 +19,9 @@ describe("extractSafeRenderErrorCode", () => {
     expect(extractSafeRenderErrorCode({ code: "VIDEO_SOURCE_UNRENDERABLE" })).toBe(
       "VIDEO_SOURCE_UNRENDERABLE",
     );
+    expect(extractSafeRenderErrorCode({ code: "INVALID_VIDEO_METADATA" })).toBe(
+      "INVALID_VIDEO_METADATA",
+    );
   });
 
   it("does not forward arbitrary codes or parse message text", () => {

packages/producer/src/server.ts

@@ -119,6 +119,7 @@ interface PreparedRenderInput {
 
 const DEFAULT_SERVER_FPS = { num: 30, den: 1 } as const;
 const SAFE_RENDER_ERROR_CODES = new Set<string>([
+  "INVALID_VIDEO_METADATA",
   "VIDEO_SOURCE_UNRENDERABLE",
   "VIDEO_EXTRACTION_FAILED",
 ]);