diff --git a/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts b/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts index 788bfb2c8d..b1d08f8534 100644 --- a/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts +++ b/packages/aws-lambda/src/cdk/HyperframesRenderStack.snapshot.test.ts @@ -64,6 +64,8 @@ const EXPECTED_NON_RETRYABLE_ERRORS = new Set([ "BROWSER_GPU_NOT_SOFTWARE", "FONT_FETCH_FAILED", "PLAN_TOO_LARGE", + "PLAN_PROTOCOL_UNSUPPORTED", + "PlanProtocolUnsupportedError", "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED", "ChromeBinaryUnavailableError", ]); diff --git a/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts b/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts index 1b0219b517..86d626d3f3 100644 --- a/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts +++ b/packages/aws-lambda/src/cdk/HyperframesRenderStack.ts @@ -200,6 +200,8 @@ export class HyperframesRenderStack extends Construct { "BROWSER_GPU_NOT_SOFTWARE", "FONT_FETCH_FAILED", "PLAN_TOO_LARGE", + "PLAN_PROTOCOL_UNSUPPORTED", + "PlanProtocolUnsupportedError", "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED", "ChromeBinaryUnavailableError", ]; @@ -208,12 +210,16 @@ export class HyperframesRenderStack extends Construct { "PLAN_HASH_MISMATCH", "S3_URI_NOT_ALLOWED", "BROWSER_GPU_NOT_SOFTWARE", + "PLAN_PROTOCOL_UNSUPPORTED", + "PlanProtocolUnsupportedError", "ChromeBinaryUnavailableError", ]; const NON_RETRYABLE_ASSEMBLE = [ "FFMPEG_VERSION_MISMATCH", "PLAN_HASH_MISMATCH", "S3_URI_NOT_ALLOWED", + "PLAN_PROTOCOL_UNSUPPORTED", + "PlanProtocolUnsupportedError", "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED", "ChromeBinaryUnavailableError", ]; diff --git a/packages/aws-lambda/src/handler.test.ts b/packages/aws-lambda/src/handler.test.ts index 5a67c2bfd4..c3d13ee73c 100644 --- a/packages/aws-lambda/src/handler.test.ts +++ b/packages/aws-lambda/src/handler.test.ts @@ -18,7 +18,12 @@ import { afterEach, beforeEach, describe, expect, it, mock } from "bun:test"; import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import type { AssembleResult, ChunkResult, PlanResult } from "@hyperframes/producer/distributed"; +import { + CURRENT_PLAN_PROTOCOL, + type AssembleResult, + type ChunkResult, + type PlanResult, +} from "@hyperframes/producer/distributed"; import type { AssembleEvent, LambdaEvent, PlanEvent, RenderChunkEvent } from "./events.js"; import { handler, unwrapEvent } from "./handler.js"; @@ -157,6 +162,7 @@ describe("handler dispatch", () => { writeFileSync(join(planDir, "meta", "chunks.json"), "[]"); return { planDir, + planProtocol: CURRENT_PLAN_PROTOCOL, planHash: "fakehash", chunkCount: 4, totalFrames: 720, @@ -224,6 +230,7 @@ describe("handler dispatch", () => { writeFileSync(join(planDir, "meta", "chunks.json"), "[]"); return { planDir, + planProtocol: CURRENT_PLAN_PROTOCOL, planHash: "fakehash", chunkCount: 1, totalFrames: 30, diff --git a/packages/gcp-cloud-run/src/server.test.ts b/packages/gcp-cloud-run/src/server.test.ts index 29888f470d..2e6f2e5e28 100644 --- a/packages/gcp-cloud-run/src/server.test.ts +++ b/packages/gcp-cloud-run/src/server.test.ts @@ -18,7 +18,13 @@ import { afterEach, describe, expect, it } from "bun:test"; import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import type { AssembleResult, ChunkResult, PlanResult } from "@hyperframes/producer/distributed"; +import { + CURRENT_PLAN_PROTOCOL, + PlanProtocolUnsupportedError, + type AssembleResult, + type ChunkResult, + type PlanResult, +} from "@hyperframes/producer/distributed"; import { asStorage, FakeGcs } from "./__fixtures__/fakeGcs.js"; import type { AssembleEvent, CloudRunEvent, PlanEvent, RenderChunkEvent } from "./events.js"; import { createApp, dispatch, type HandlerDeps, unwrapEvent } from "./server.js"; @@ -56,6 +62,7 @@ async function seedPlanTar(gcs: FakeGcs, uri: string, planHash: string): Promise const planResult: PlanResult = { planDir: "(set at call time)", + planProtocol: CURRENT_PLAN_PROTOCOL, planHash: PLAN_HASH, chunkCount: 3, totalFrames: 90, @@ -295,6 +302,34 @@ describe("createApp HTTP mapping", () => { expect(body.error).toBe("PLAN_HASH_MISMATCH"); }); + it("returns 400 for an unsupported plan protocol", async () => { + const gcs = new FakeGcs(); + await seedPlanTar(gcs, "gs://b/renders/r1/plan.tar.gz", PLAN_HASH); + const app = createApp( + depsWith(gcs, { + renderChunk: async () => { + throw new PlanProtocolUnsupportedError("unsupported test protocol"); + }, + }), + ); + 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("PlanProtocolUnsupportedError"); + }); + 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)); diff --git a/packages/gcp-cloud-run/src/server.ts b/packages/gcp-cloud-run/src/server.ts index 078d2c78f3..7aa7f84878 100644 --- a/packages/gcp-cloud-run/src/server.ts +++ b/packages/gcp-cloud-run/src/server.ts @@ -584,10 +584,12 @@ const NON_RETRYABLE_ERROR_NAMES = new Set([ // non-retryable list. "FormatNotSupportedInDistributedError", "PlanTooLargeError", + "PlanProtocolUnsupportedError", "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/producer/src/distributed.ts b/packages/producer/src/distributed.ts index 25db28ca88..8a6f7bfb82 100644 --- a/packages/producer/src/distributed.ts +++ b/packages/producer/src/distributed.ts @@ -82,6 +82,25 @@ export { } from "./services/distributed/renderConfigValidation.js"; export { hashProjectDir } from "./services/distributed/projectHash.js"; +// ── Plan protocol compatibility ──────────────────────────────────────────── +// Workers validate this descriptor before consuming layout-specific +// artifacts. Missing descriptors remain compatible with legacy v1 plans. +export { + CURRENT_PLAN_PROTOCOL, + DISTRIBUTED_RENDER_CAPABILITIES, + getDistributedRenderCapabilities, + PLAN_ARTIFACT_LAYOUT, + PLAN_HASH_SCHEMA, + PLAN_PROTOCOL_UNSUPPORTED, + PLAN_SCHEMA_VERSION, + PlanProtocolUnsupportedError, + readPlanProtocol, + type DistributedRenderCapabilities, + type PlanProtocolConsumerCapabilities, + type PlanProtocolDescriptor, + type PlanProtocolV1Descriptor, +} from "./services/distributed/planProtocol.js"; + // ── Format union ──────────────────────────────────────────────────────────── // Canonical output-format type. The aws-lambda package re-exports it so // CLI / adopter SDKs can derive runtime allowlists from one source. diff --git a/packages/producer/src/index.ts b/packages/producer/src/index.ts index 416633a58c..4bc24aaaa5 100644 --- a/packages/producer/src/index.ts +++ b/packages/producer/src/index.ts @@ -133,10 +133,23 @@ export { // separate subpath import. export { assemble, + CURRENT_PLAN_PROTOCOL, + DISTRIBUTED_RENDER_CAPABILITIES, + getDistributedRenderCapabilities, + PLAN_ARTIFACT_LAYOUT, + PLAN_HASH_SCHEMA, + PLAN_PROTOCOL_UNSUPPORTED, + PLAN_SCHEMA_VERSION, plan, + PlanProtocolUnsupportedError, + readPlanProtocol, renderChunk, type AssembleResult, type ChunkResult, + type DistributedRenderCapabilities, type DistributedRenderConfig, + type PlanProtocolConsumerCapabilities, + type PlanProtocolDescriptor, + type PlanProtocolV1Descriptor, type PlanResult, } from "./distributed.js"; diff --git a/packages/producer/src/services/distributed/assemble.ts b/packages/producer/src/services/distributed/assemble.ts index 3b59b91b3e..5aa745a1c4 100644 --- a/packages/producer/src/services/distributed/assemble.ts +++ b/packages/producer/src/services/distributed/assemble.ts @@ -40,6 +40,7 @@ import { defaultLogger, type ProducerLogger } from "../../logger.js"; import { formatExportFrameName } from "../../utils/paths.js"; import { padOrTrimAudioToVideoFrameCount } from "../render/audioPadTrim.js"; import type { ChunkSliceJson } from "../render/stages/freezePlan.js"; +import { DISTRIBUTED_RENDER_CAPABILITIES, readPlanProtocol } from "./planProtocol.js"; import type { DistributedFormat } from "./shared.js"; /** @@ -56,6 +57,7 @@ export interface AssembleResult { /** Shape of the planDir's top-level `plan.json` — only the fields `assemble` needs. */ interface PlanJsonForAssemble { + protocol?: unknown; planHash: string; totalFrames: number; hasAudio: boolean; @@ -118,10 +120,11 @@ export async function assemble( if (!existsSync(planJsonPath)) { throw new Error(`[assemble] planDir missing plan.json: ${planJsonPath}`); } + const plan = JSON.parse(readFileSync(planJsonPath, "utf-8")) as PlanJsonForAssemble; + readPlanProtocol(plan, DISTRIBUTED_RENDER_CAPABILITIES.roles.assembler); if (!existsSync(chunksJsonPath)) { throw new Error(`[assemble] planDir missing meta/chunks.json: ${chunksJsonPath}`); } - const plan = JSON.parse(readFileSync(planJsonPath, "utf-8")) as PlanJsonForAssemble; const chunks = JSON.parse(readFileSync(chunksJsonPath, "utf-8")) as ChunkSliceJson[]; if (chunkPaths.length !== chunks.length) { throw new Error( diff --git a/packages/producer/src/services/distributed/plan.test.ts b/packages/producer/src/services/distributed/plan.test.ts index d005e23e04..5a1afe4804 100644 --- a/packages/producer/src/services/distributed/plan.test.ts +++ b/packages/producer/src/services/distributed/plan.test.ts @@ -20,6 +20,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { recomputePlanHashFromPlanDir } from "../render/stages/freezePlan.js"; import { RenderQualityError } from "../renderOrchestrator.js"; +import { CURRENT_PLAN_PROTOCOL } from "./planProtocol.js"; import { applyDistributedAudioWarningPolicy, buildChunkSlices, @@ -345,6 +346,7 @@ describe("plan() — golden planDir + planHash determinism", () => { // ── PlanResult contract ───────────────────────────────────────────── expect(result.planDir).toBe(planDir); + expect(result.planProtocol).toEqual(CURRENT_PLAN_PROTOCOL); expect(result.planHash).toMatch(/^[0-9a-f]{64}$/); expect(result.chunkCount).toBe(1); expect(result.totalFrames).toBe(30); // 1s @ 30fps @@ -373,6 +375,7 @@ describe("plan() — golden planDir + planHash determinism", () => { unknown >; expect(planJson.planHash).toBe(result.planHash); + expect(planJson.protocol).toEqual(CURRENT_PLAN_PROTOCOL); expect(planJson.hasAudio).toBe(false); expect(planJson.totalFrames).toBe(result.totalFrames); }, @@ -460,8 +463,18 @@ describe("plan() — golden planDir + planHash determinism", () => { expect(recomputed).toBe(result.planHash); const planJson = JSON.parse(readFileSync(join(planDir, "plan.json"), "utf-8")) as { planHash: string; + protocol?: unknown; }; expect(planJson.planHash).toBe(result.planHash); + expect(planJson.protocol).toEqual(CURRENT_PLAN_PROTOCOL); + + delete planJson.protocol; + writeFileSync(join(planDir, "plan.json"), `${JSON.stringify(planJson, null, 2)}\n`, "utf-8"); + expect(recomputePlanHashFromPlanDir(planDir)).toBe(result.planHash); + + planJson.protocol = CURRENT_PLAN_PROTOCOL; + writeFileSync(join(planDir, "plan.json"), `${JSON.stringify(planJson, null, 2)}\n`, "utf-8"); + expect(recomputePlanHashFromPlanDir(planDir)).toBe(result.planHash); }, TIMEOUT_MS, ); diff --git a/packages/producer/src/services/distributed/plan.ts b/packages/producer/src/services/distributed/plan.ts index 8f11d7bd55..f1bf366551 100644 --- a/packages/producer/src/services/distributed/plan.ts +++ b/packages/producer/src/services/distributed/plan.ts @@ -79,6 +79,7 @@ import { readFfmpegVersion, readProducerVersion, } from "./shared.js"; +import { CURRENT_PLAN_PROTOCOL, type PlanProtocolV1Descriptor } from "./planProtocol.js"; /** * Caller-supplied configuration for a distributed render. `fps`, `width`, @@ -254,6 +255,7 @@ export interface DistributedRenderConfig { */ export interface PlanResult { planDir: string; + planProtocol: Readonly; planHash: string; chunkCount: number; totalFrames: number; @@ -1084,6 +1086,7 @@ export async function plan( return { planDir, + planProtocol: CURRENT_PLAN_PROTOCOL, planHash, chunkCount, totalFrames, diff --git a/packages/producer/src/services/distributed/planProtocol.test.ts b/packages/producer/src/services/distributed/planProtocol.test.ts new file mode 100644 index 0000000000..56e4003c36 --- /dev/null +++ b/packages/producer/src/services/distributed/planProtocol.test.ts @@ -0,0 +1,296 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { assemble } from "./assemble.js"; +import { + CURRENT_PLAN_PROTOCOL, + DISTRIBUTED_RENDER_CAPABILITIES, + getDistributedRenderCapabilities, + PLAN_ARTIFACT_LAYOUT, + PLAN_HASH_SCHEMA, + PLAN_PROTOCOL_UNSUPPORTED, + PLAN_SCHEMA_VERSION, + PlanProtocolUnsupportedError, + readPlanProtocol, + type DistributedRenderCapabilities, + type PlanProtocolConsumerCapabilities, + type PlanProtocolDescriptor, +} from "./planProtocol.js"; +import { + CHUNK_INDEX_OUT_OF_RANGE, + renderChunk, + RenderChunkValidationError, +} from "./renderChunk.js"; + +const tempDirs: string[] = []; + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); + +function expectUnsupported(run: () => unknown): PlanProtocolUnsupportedError { + let caught: unknown; + try { + run(); + } catch (error) { + caught = error; + } + expect(caught).toBeInstanceOf(PlanProtocolUnsupportedError); + expect((caught as PlanProtocolUnsupportedError).code).toBe(PLAN_PROTOCOL_UNSUPPORTED); + return caught as PlanProtocolUnsupportedError; +} + +function createReaderPlan(options: { + protocol?: unknown; + includeProtocol?: boolean; + malformedDownstreamArtifacts?: boolean; + omitDownstreamArtifacts?: boolean; +}): string { + const planDir = mkdtempSync(join(tmpdir(), "hf-plan-protocol-")); + tempDirs.push(planDir); + + const planJson: Record = { + planHash: "fake", + totalFrames: 1, + hasAudio: false, + dimensions: { + fpsNum: 30, + fpsDen: 1, + width: 16, + height: 16, + format: "png-sequence", + }, + }; + if (options.includeProtocol === true) { + planJson.protocol = options.protocol; + } + + writeFileSync(join(planDir, "plan.json"), JSON.stringify(planJson), "utf-8"); + if (options.omitDownstreamArtifacts === true) { + return planDir; + } + + mkdirSync(join(planDir, "meta"), { recursive: true }); + writeFileSync( + join(planDir, "meta", "encoder.json"), + options.malformedDownstreamArtifacts ? "{not-json" : "{}", + "utf-8", + ); + writeFileSync( + join(planDir, "meta", "chunks.json"), + options.malformedDownstreamArtifacts + ? "{not-json" + : JSON.stringify([{ index: 0, startFrame: 0, endFrame: 1 }]), + "utf-8", + ); + return planDir; +} + +describe("readPlanProtocol()", () => { + it("treats an absent descriptor as legacy v1", () => { + expect(readPlanProtocol({ planHash: "legacy" })).toBe(CURRENT_PLAN_PROTOCOL); + }); + + it("enforces whether a worker accepts descriptor-less legacy v1 plans", () => { + const capabilities: PlanProtocolConsumerCapabilities = { + accepts: [CURRENT_PLAN_PROTOCOL], + acceptsLegacyV1WithoutDescriptor: false, + }; + + expectUnsupported(() => readPlanProtocol({ planHash: "legacy" }, capabilities)); + }); + + it("enforces the worker's accepted protocol set", () => { + const capabilities: PlanProtocolConsumerCapabilities = { + accepts: [], + acceptsLegacyV1WithoutDescriptor: true, + }; + + expectUnsupported(() => readPlanProtocol({ protocol: CURRENT_PLAN_PROTOCOL }, capabilities)); + expectUnsupported(() => readPlanProtocol({ planHash: "legacy" }, capabilities)); + }); + + it("accepts the known v1 descriptor and ignores unknown optional fields", () => { + expect( + readPlanProtocol({ + protocol: { + schemaVersion: PLAN_SCHEMA_VERSION, + artifactLayout: PLAN_ARTIFACT_LAYOUT, + hashSchema: PLAN_HASH_SCHEMA, + producerBuildId: "optional-future-metadata", + }, + }), + ).toBe(CURRENT_PLAN_PROTOCOL); + }); + + it("rejects malformed and partial descriptors", () => { + for (const protocol of [ + null, + [], + "v1", + {}, + { schemaVersion: PLAN_SCHEMA_VERSION }, + { + schemaVersion: PLAN_SCHEMA_VERSION, + artifactLayout: PLAN_ARTIFACT_LAYOUT, + }, + ]) { + expectUnsupported(() => readPlanProtocol({ protocol })); + } + }); + + it("rejects unknown schema, layout, and hash-schema values", () => { + for (const protocol of [ + { ...CURRENT_PLAN_PROTOCOL, schemaVersion: 2 }, + { ...CURRENT_PLAN_PROTOCOL, artifactLayout: "plan-dir-v2" }, + { ...CURRENT_PLAN_PROTOCOL, hashSchema: "hyperframes-plan-hash-v2" }, + ]) { + expectUnsupported(() => readPlanProtocol({ protocol })); + } + }); + + it("keeps unsupported-protocol messages bounded and non-reflective", () => { + const untrustedValue = "secret-".repeat(1_000); + const error = expectUnsupported(() => + readPlanProtocol({ + protocol: { ...CURRENT_PLAN_PROTOCOL, hashSchema: untrustedValue }, + }), + ); + + expect(error.message).not.toContain("secret-"); + expect(error.message.length).toBeLessThan(200); + }); +}); + +describe("getDistributedRenderCapabilities()", () => { + it("reports explicit v1 support for every distributed role", () => { + expect(getDistributedRenderCapabilities()).toBe(DISTRIBUTED_RENDER_CAPABILITIES); + expect(DISTRIBUTED_RENDER_CAPABILITIES).toEqual({ + roles: { + planner: { + produces: [CURRENT_PLAN_PROTOCOL], + }, + chunk: { + accepts: [CURRENT_PLAN_PROTOCOL], + acceptsLegacyV1WithoutDescriptor: true, + }, + assembler: { + accepts: [CURRENT_PLAN_PROTOCOL], + acceptsLegacyV1WithoutDescriptor: true, + }, + }, + }); + }); + + it("can express a v2 planner with dual-version readers", () => { + const futureV2: PlanProtocolDescriptor = { + schemaVersion: 2, + artifactLayout: "plan-dir-v2", + hashSchema: "hyperframes-plan-hash-v2", + }; + const rolloutCapabilities: DistributedRenderCapabilities = { + roles: { + planner: { + produces: [futureV2], + }, + chunk: { + accepts: [CURRENT_PLAN_PROTOCOL, futureV2], + acceptsLegacyV1WithoutDescriptor: true, + }, + assembler: { + accepts: [CURRENT_PLAN_PROTOCOL, futureV2], + acceptsLegacyV1WithoutDescriptor: true, + }, + }, + }; + + expect(rolloutCapabilities.roles.planner.produces).toEqual([futureV2]); + expect(rolloutCapabilities.roles.chunk.accepts).toEqual([CURRENT_PLAN_PROTOCOL, futureV2]); + expect(rolloutCapabilities.roles.assembler.accepts).toEqual([CURRENT_PLAN_PROTOCOL, futureV2]); + }); +}); + +describe("distributed plan protocol readers", () => { + it("renderChunk accepts a legacy plan without a descriptor", async () => { + const planDir = createReaderPlan({}); + + let caught: unknown; + try { + await renderChunk(planDir, 999, join(planDir, "unused-output")); + } catch (error) { + caught = error; + } + + expect(caught).toBeInstanceOf(RenderChunkValidationError); + expect((caught as RenderChunkValidationError).code).toBe(CHUNK_INDEX_OUT_OF_RANGE); + }); + + it("assemble accepts a legacy plan without a descriptor", async () => { + const planDir = createReaderPlan({}); + const missingChunk = join(planDir, "missing-chunk"); + + await expect( + assemble(planDir, [missingChunk], null, join(planDir, "unused-output")), + ).rejects.toThrow("chunk path does not exist"); + }); + + it("renderChunk rejects an unknown v2 protocol before requiring v1 artifacts", async () => { + const planDir = createReaderPlan({ + includeProtocol: true, + protocol: { ...CURRENT_PLAN_PROTOCOL, schemaVersion: 2 }, + omitDownstreamArtifacts: true, + }); + + let caught: unknown; + try { + await renderChunk(planDir, 0, join(planDir, "unused-output")); + } catch (error) { + caught = error; + } + + expect(caught).toBeInstanceOf(PlanProtocolUnsupportedError); + expect((caught as PlanProtocolUnsupportedError).code).toBe(PLAN_PROTOCOL_UNSUPPORTED); + }); + + it("assemble rejects an unknown v2 protocol before requiring v1 artifacts", async () => { + const planDir = createReaderPlan({ + includeProtocol: true, + protocol: { ...CURRENT_PLAN_PROTOCOL, schemaVersion: 2 }, + omitDownstreamArtifacts: true, + }); + + let caught: unknown; + try { + await assemble(planDir, [], null, join(planDir, "unused-output")); + } catch (error) { + caught = error; + } + + expect(caught).toBeInstanceOf(PlanProtocolUnsupportedError); + expect((caught as PlanProtocolUnsupportedError).code).toBe(PLAN_PROTOCOL_UNSUPPORTED); + }); + + it("assemble rejects a partial protocol before parsing chunks", async () => { + const planDir = createReaderPlan({ + includeProtocol: true, + protocol: { + schemaVersion: PLAN_SCHEMA_VERSION, + artifactLayout: PLAN_ARTIFACT_LAYOUT, + }, + malformedDownstreamArtifacts: true, + }); + + let caught: unknown; + try { + await assemble(planDir, [], null, join(planDir, "unused-output")); + } catch (error) { + caught = error; + } + + expect(caught).toBeInstanceOf(PlanProtocolUnsupportedError); + expect((caught as PlanProtocolUnsupportedError).code).toBe(PLAN_PROTOCOL_UNSUPPORTED); + }); +}); diff --git a/packages/producer/src/services/distributed/planProtocol.ts b/packages/producer/src/services/distributed/planProtocol.ts new file mode 100644 index 0000000000..16163dd9d5 --- /dev/null +++ b/packages/producer/src/services/distributed/planProtocol.ts @@ -0,0 +1,155 @@ +/** + * Compatibility contract for the on-disk distributed render plan. + * + * A missing descriptor means the original v1 plan layout. This preserves + * replay compatibility with plan directories produced before the descriptor + * existed. Once a descriptor is present it must be complete and recognized: + * silently guessing across a partially-written or newer layout could render + * incorrect pixels or make assemble consume the wrong artifacts. + */ + +export const PLAN_SCHEMA_VERSION = 1 as const; +export const PLAN_ARTIFACT_LAYOUT = "plan-dir-v1" as const; +export const PLAN_HASH_SCHEMA = "hyperframes-plan-hash-v1" as const; +export const PLAN_PROTOCOL_UNSUPPORTED = "PLAN_PROTOCOL_UNSUPPORTED" as const; + +export interface PlanProtocolDescriptor { + readonly schemaVersion: number; + readonly artifactLayout: string; + readonly hashSchema: string; +} + +export interface PlanProtocolV1Descriptor extends PlanProtocolDescriptor { + readonly schemaVersion: typeof PLAN_SCHEMA_VERSION; + readonly artifactLayout: typeof PLAN_ARTIFACT_LAYOUT; + readonly hashSchema: typeof PLAN_HASH_SCHEMA; +} + +/** Descriptor written by the current producer and accepted by v1 workers. */ +export const CURRENT_PLAN_PROTOCOL: Readonly = Object.freeze({ + schemaVersion: PLAN_SCHEMA_VERSION, + artifactLayout: PLAN_ARTIFACT_LAYOUT, + hashSchema: PLAN_HASH_SCHEMA, +}); + +export interface PlanProtocolConsumerCapabilities { + readonly accepts: readonly Readonly[]; + readonly acceptsLegacyV1WithoutDescriptor: boolean; +} + +export interface DistributedRenderCapabilities { + readonly roles: Readonly<{ + planner: Readonly<{ + produces: readonly Readonly[]; + }>; + chunk: Readonly; + assembler: Readonly; + }>; +} + +/** Serializable capability payload for fleet rollout and worker handshakes. */ +export const DISTRIBUTED_RENDER_CAPABILITIES: Readonly = + Object.freeze({ + roles: Object.freeze({ + planner: Object.freeze({ + produces: Object.freeze([CURRENT_PLAN_PROTOCOL]), + }), + chunk: Object.freeze({ + accepts: Object.freeze([CURRENT_PLAN_PROTOCOL]), + acceptsLegacyV1WithoutDescriptor: true, + }), + assembler: Object.freeze({ + accepts: Object.freeze([CURRENT_PLAN_PROTOCOL]), + acceptsLegacyV1WithoutDescriptor: true, + }), + }), + }); + +export function getDistributedRenderCapabilities(): Readonly { + return DISTRIBUTED_RENDER_CAPABILITIES; +} + +/** Typed, deterministic compatibility failure. Retrying on the same worker cannot heal it. */ +export class PlanProtocolUnsupportedError extends Error { + // Public adapters inspect this typed code even though OSS producer does not. + // fallow-ignore-next-line unused-class-member + readonly code: typeof PLAN_PROTOCOL_UNSUPPORTED = PLAN_PROTOCOL_UNSUPPORTED; + + constructor(reason: string) { + super(`[planProtocol] ${reason} (${PLAN_PROTOCOL_UNSUPPORTED})`); + this.name = "PlanProtocolUnsupportedError"; + } +} + +function isRecord(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +function protocolMatches( + descriptor: Record, + expected: PlanProtocolDescriptor, +): boolean { + return ( + descriptor.schemaVersion === expected.schemaVersion && + descriptor.artifactLayout === expected.artifactLayout && + descriptor.hashSchema === expected.hashSchema + ); +} + +function capabilitiesAccept( + capabilities: Readonly, + protocol: Readonly, +): boolean { + return capabilities.accepts.some( + (accepted) => + accepted.schemaVersion === protocol.schemaVersion && + accepted.artifactLayout === protocol.artifactLayout && + accepted.hashSchema === protocol.hashSchema, + ); +} + +/** + * Read and validate the plan protocol before a worker consumes other plan + * artifacts. Unknown fields on a recognized v1 descriptor are intentionally + * ignored so optional metadata can be added without a lockstep fleet deploy. + */ +export function readPlanProtocol( + planJson: unknown, + capabilities: Readonly = DISTRIBUTED_RENDER_CAPABILITIES.roles + .chunk, +): Readonly { + if (!isRecord(planJson)) { + throw new PlanProtocolUnsupportedError("plan.json must contain a JSON object"); + } + + if (!Object.prototype.hasOwnProperty.call(planJson, "protocol")) { + if ( + !capabilities.acceptsLegacyV1WithoutDescriptor || + !capabilitiesAccept(capabilities, CURRENT_PLAN_PROTOCOL) + ) { + throw new PlanProtocolUnsupportedError( + "legacy v1 plan without a protocol descriptor is not accepted by this worker", + ); + } + return CURRENT_PLAN_PROTOCOL; + } + + const descriptor = planJson.protocol; + if (!isRecord(descriptor)) { + throw new PlanProtocolUnsupportedError("plan.json protocol descriptor must be an object"); + } + + for (const field of ["schemaVersion", "artifactLayout", "hashSchema"] as const) { + if (!Object.prototype.hasOwnProperty.call(descriptor, field)) { + throw new PlanProtocolUnsupportedError(`plan.json protocol descriptor is missing ${field}`); + } + } + + if ( + !protocolMatches(descriptor, CURRENT_PLAN_PROTOCOL) || + !capabilitiesAccept(capabilities, CURRENT_PLAN_PROTOCOL) + ) { + throw new PlanProtocolUnsupportedError("unsupported plan.json protocol descriptor"); + } + return CURRENT_PLAN_PROTOCOL; +} diff --git a/packages/producer/src/services/distributed/publicExports.test.ts b/packages/producer/src/services/distributed/publicExports.test.ts index 004ba4fce6..a72fd19c00 100644 --- a/packages/producer/src/services/distributed/publicExports.test.ts +++ b/packages/producer/src/services/distributed/publicExports.test.ts @@ -64,6 +64,34 @@ describe("@hyperframes/producer/distributed (subpath)", () => { expect(typeof distributedSubpath.applyRuntimeEnvSnapshot).toBe("function"); expect(typeof distributedSubpath.readWebGlVendorInfoFromCanvas).toBe("function"); }); + + it("exports the plan protocol contract", () => { + expect(distributedSubpath.PLAN_SCHEMA_VERSION).toBe(1); + expect(distributedSubpath.PLAN_ARTIFACT_LAYOUT).toBe("plan-dir-v1"); + expect(distributedSubpath.PLAN_HASH_SCHEMA).toBe("hyperframes-plan-hash-v1"); + expect(distributedSubpath.PLAN_PROTOCOL_UNSUPPORTED).toBe("PLAN_PROTOCOL_UNSUPPORTED"); + expect(distributedSubpath.CURRENT_PLAN_PROTOCOL).toEqual({ + schemaVersion: 1, + artifactLayout: "plan-dir-v1", + hashSchema: "hyperframes-plan-hash-v1", + }); + expect(distributedSubpath.DISTRIBUTED_RENDER_CAPABILITIES.roles).toEqual({ + planner: { + produces: [distributedSubpath.CURRENT_PLAN_PROTOCOL], + }, + chunk: { + accepts: [distributedSubpath.CURRENT_PLAN_PROTOCOL], + acceptsLegacyV1WithoutDescriptor: true, + }, + assembler: { + accepts: [distributedSubpath.CURRENT_PLAN_PROTOCOL], + acceptsLegacyV1WithoutDescriptor: true, + }, + }); + expect(typeof distributedSubpath.getDistributedRenderCapabilities).toBe("function"); + expect(typeof distributedSubpath.readPlanProtocol).toBe("function"); + expect(typeof distributedSubpath.PlanProtocolUnsupportedError).toBe("function"); + }); }); describe("@hyperframes/producer (main entry)", () => { @@ -73,6 +101,17 @@ describe("@hyperframes/producer (main entry)", () => { expect(typeof producerIndex.assemble).toBe("function"); }); + it("re-exports the plan protocol contract", () => { + expect(producerIndex.CURRENT_PLAN_PROTOCOL).toBe(distributedSubpath.CURRENT_PLAN_PROTOCOL); + expect(producerIndex.DISTRIBUTED_RENDER_CAPABILITIES).toBe( + distributedSubpath.DISTRIBUTED_RENDER_CAPABILITIES, + ); + expect(typeof producerIndex.getDistributedRenderCapabilities).toBe("function"); + expect(producerIndex.PLAN_PROTOCOL_UNSUPPORTED).toBe("PLAN_PROTOCOL_UNSUPPORTED"); + expect(typeof producerIndex.readPlanProtocol).toBe("function"); + expect(typeof producerIndex.PlanProtocolUnsupportedError).toBe("function"); + }); + it("preserves the existing in-process exports (executeRenderJob unchanged)", () => { // The distributed primitives must NOT break the in-process surface; // spot-check the load-bearing exports the in-process callers rely on. diff --git a/packages/producer/src/services/distributed/renderChunk.ts b/packages/producer/src/services/distributed/renderChunk.ts index 4f5009f15b..79ac08d2f1 100644 --- a/packages/producer/src/services/distributed/renderChunk.ts +++ b/packages/producer/src/services/distributed/renderChunk.ts @@ -80,6 +80,7 @@ import { type PlanVideosJson, readFfmpegVersion, } from "./shared.js"; +import { DISTRIBUTED_RENDER_CAPABILITIES, readPlanProtocol } from "./planProtocol.js"; /** * Non-retryable error codes raised when the planDir is structurally @@ -229,6 +230,7 @@ export function rebuildExtractedFramesFromPlanDir( /** Plan-time JSON manifest written by `freezePlan`. */ interface PlanJson { + protocol?: unknown; planHash: string; producerVersion: string; ffmpegVersion: string; @@ -333,7 +335,15 @@ export async function renderChunk( const planJsonPath = join(planDir, "plan.json"); const encoderJsonPath = join(planDir, "meta", "encoder.json"); const chunksJsonPath = join(planDir, "meta", "chunks.json"); - for (const required of [planJsonPath, encoderJsonPath, chunksJsonPath]) { + if (!existsSync(planJsonPath)) { + throw new RenderChunkValidationError( + MISSING_PLAN_ARTIFACT, + `[renderChunk] planDir is missing required artifact: ${planJsonPath}`, + ); + } + const plan = JSON.parse(readFileSync(planJsonPath, "utf-8")) as PlanJson; + readPlanProtocol(plan, DISTRIBUTED_RENDER_CAPABILITIES.roles.chunk); + for (const required of [encoderJsonPath, chunksJsonPath]) { if (!existsSync(required)) { throw new RenderChunkValidationError( MISSING_PLAN_ARTIFACT, @@ -341,7 +351,6 @@ export async function renderChunk( ); } } - const plan = JSON.parse(readFileSync(planJsonPath, "utf-8")) as PlanJson; const encoder = JSON.parse(readFileSync(encoderJsonPath, "utf-8")) as LockedRenderConfig; const chunks = JSON.parse(readFileSync(chunksJsonPath, "utf-8")) as ChunkSliceJson[]; diff --git a/packages/producer/src/services/render/stages/freezePlan.ts b/packages/producer/src/services/render/stages/freezePlan.ts index 58bccc0cdb..5bf8a738e3 100644 --- a/packages/producer/src/services/render/stages/freezePlan.ts +++ b/packages/producer/src/services/render/stages/freezePlan.ts @@ -14,6 +14,7 @@ import { existsSync, mkdirSync, readFileSync, readdirSync, writeFileSync } from import { join, relative, resolve } from "node:path"; import type { Fps } from "@hyperframes/core"; +import { CURRENT_PLAN_PROTOCOL } from "../../distributed/planProtocol.js"; import { canonicalJsonStringify, computePlanHash, @@ -355,6 +356,7 @@ export async function freezePlan(input: FreezePlanInput): Promise