ext/hyperframes
2
0
Fork 0
mirror of https://github.com/heygen-com/hyperframes.git synced 2026-10-30 17:27:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(producer): size distributed chunks to the worker limit and concurrency (#5765)

Browse Source 
* perf(producer): size distributed chunks to the worker limit and concurrency

* chore(cli): drop a comment the field name already says
This commit is contained in:
Miguel Ángel
2026-10-17 08:45:47 +00:00
committed by GitHub
parent d5162beb50
commit c69fd2c6ec
18 changed files with 378 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/deploy/migrating-to-hyperframes-lambda.mdx
+2 -2
View File
@@ -57,8 +57,8 @@ Most adopters' render config maps directly:
| `codec: 'h264' / 'h265'` | `--codec=h264` or `--codec=h265` (mp4 only) | h265 uses libx265 with closed-GOP keyint params so chunked concat-copy round-trips losslessly. |
| Output format | `--format=mp4 / mov / webm / png-sequence` | webm uses libvpx-vp9 + closed-GOP concat-copy. Distributed mode still refuses HDR mp4 at plan time. |
| Quality preset | `--quality=draft / standard / high` | Maps onto ffmpeg encoder presets. |
| Chunk size in frames | `--chunk-size=240` (default 240) | ~8s at 30 fps; sized to fit Lambda's 15-min cap with headroom. |
| Max parallel chunks | `--max-parallel-chunks=16` (default 16) | Caps the Map state's fan-out. |
| Chunk size in frames | `--chunk-size=240` (default: auto) | By default each chunk is sized to finish inside the Lambda timeout with headroom; a render that cannot fails at plan time with `CHUNK_EXCEEDS_WORKER_TIMEOUT`. |
| Max parallel chunks | `--max-parallel-chunks=16` (default: auto) | By default one wave across the deploy's `--concurrency` (16 if unknown), widened when a chunk would outrun the Lambda timeout. |
| Per-chunk frame ceiling | `--target-chunk-frames=N` (optional) | Caps frames per chunk so one chunk can't run past Lambda's 15-min cap on a long video: the planner adds chunks (up to `--max-parallel-chunks`) to keep each at or below `N`, and short videos still collapse to fewer chunks. A ceiling, not a fixed size; ignored when `--chunk-size` is set. |
| Bitrate / CRF | `config.bitrate` or `config.crf` in the SDK | Mutually exclusive; the current Lambda CLI does not expose these two fields. |
docs/packages/cli.mdx
+2 -2
View File
@@ -1942,8 +1942,8 @@ And these drive `render` and `render-batch`:
| `--codec` | `h264` | `h264` or `h265`, MP4 only |
| `--quality` | `standard` | `draft`, `standard`, or `high` |
| `--output-resolution` | — | Supersample via Chrome `deviceScaleFactor` to a preset (`1080p`, `4k`, `uhd`, `hd`, `landscape-4k`, …) without changing layout |
| `--chunk-size` | `240` | Frames per chunk |
| `--max-parallel-chunks` | `16` | Concurrent chunks per render |
| `--chunk-size` | auto | Frames per chunk. By default each chunk is sized to finish inside the Lambda timeout |
| `--max-parallel-chunks` | auto | Chunks per render. By default one wave across the deploy's `--concurrency` (16 if unknown), widened when a chunk would outrun the Lambda timeout |
| `--target-chunk-frames` | — | Ceiling on frames per chunk. The planner adds chunks — up to `--max-parallel-chunks` — to stay under it, and short videos still collapse to fewer. Ignored when `--chunk-size` is set. |
| `--variables` | — | JSON object reaching every chunk worker as `window.__hfVariables` |
| `--variables-file` | — | Read those from a JSON file |
packages/aws-lambda/src/handler.test.ts
+39
View File
@@ -247,6 +247,43 @@ describe("handler dispatch", () => {
).toBe(true);
});
it("plans against the function's own timeout unless the config pins one", async () => {
const seen: Array<number | undefined> = [];
for (const pinned of [undefined, 300]) {
const tmpRoot = makeTmpRoot();
const s3 = new FakeS3Client();
s3.objects.set("s3://bucket/project.tar.gz", await makeMinimalProjectTar());
await handler(
{
Action: "plan",
PlanProtocol: "v1",
ProjectS3Uri: "s3://bucket/project.tar.gz",
PlanOutputS3Prefix: "s3://bucket/renders/timeout/",
Config: { fps: 30, width: 640, height: 360, format: "mp4", workerTimeoutSeconds: pinned },
},
{
s3: s3 as unknown as import("@aws-sdk/client-s3").S3Client,
primitives: {
plan: mock(async (_dir: string, config: { workerTimeoutSeconds?: number }) => {
seen.push(config.workerTimeoutSeconds);
throw new Error("stop after config");
}) as unknown as typeof import("@hyperframes/producer/distributed").plan,
renderChunk: mock(async () => {
throw new Error("unused");
}) as unknown as typeof import("@hyperframes/producer/distributed").renderChunk,
assemble: mock(async () => {
throw new Error("unused");
}) as unknown as typeof import("@hyperframes/producer/distributed").assemble,
},
tmpRoot,
skipChromeResolution: true,
getRemainingTimeInMillis: () => 899_987,
},
).catch(() => undefined);
}
expect(seen).toEqual([900, 300]);
});
it("normalizes producer workflow codes to Step Functions error names", async () => {
for (const code of [
"PLAN_TOO_LARGE",
@@ -257,6 +294,8 @@ describe("handler dispatch", () => {
"VIDEO_SOURCE_UNRENDERABLE",
"VIDEO_EXTRACTION_FAILED",
"INVALID_VIDEO_METADATA",
"FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED",
"CHUNK_EXCEEDS_WORKER_TIMEOUT",
] as const) {
const tmpRoot = makeTmpRoot();
const s3 = new FakeS3Client();
packages/aws-lambda/src/handler.ts
+19 -5
View File
@@ -91,6 +91,11 @@ export interface HandlerDeps {
tmpRoot?: string;
/** Skip Chrome resolution (used by handler dispatch tests that mock renderChunk). */
skipChromeResolution?: boolean;
/**
* The Lambda runtime's `Context` arrives as this second argument and supplies it. Plan
* and chunk share one function, so its remaining time at plan start is the chunk limit.
*/
getRemainingTimeInMillis?: () => number;
}
/**
@@ -162,7 +167,9 @@ function normalizeTerminalErrorName(error: unknown): void {
candidate.code === "NOT_MEDIA_PAYLOAD" ||
candidate.code === "VIDEO_SOURCE_UNRENDERABLE" ||
candidate.code === "VIDEO_EXTRACTION_FAILED" ||
candidate.code === "INVALID_VIDEO_METADATA"
candidate.code === "INVALID_VIDEO_METADATA" ||
candidate.code === "FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED" ||
candidate.code === "CHUNK_EXCEEDS_WORKER_TIMEOUT"
) {
candidate.name = candidate.code;
}
@@ -316,6 +323,14 @@ function primeRuntimeEnv(): void {
// ── Plan ────────────────────────────────────────────────────────────────────
function planConfig(event: PlanEvent, deps?: HandlerDeps): DistributedRenderConfig {
const remainingMs = deps?.getRemainingTimeInMillis?.();
if (event.Config.workerTimeoutSeconds !== undefined || remainingMs === undefined) {
return { ...event.Config };
}
return { ...event.Config, workerTimeoutSeconds: Math.ceil(remainingMs / 1000) };
}
// The v1 handler owns one transactional download, plan, archive, upload, and cleanup lifecycle.
// fallow-ignore-next-line complexity
async function handlePlan(event: PlanEvent, deps?: HandlerDeps): Promise<PlanLambdaResult> {
@@ -323,6 +338,7 @@ async function handlePlan(event: PlanEvent, deps?: HandlerDeps): Promise<PlanLam
return handlePlanV2(event, deps);
}
const started = Date.now();
const config = planConfig(event, deps);
const s3 = deps?.s3 ?? getS3Client();
const primitive = deps?.primitives?.plan ?? plan;
@@ -349,9 +365,6 @@ async function handlePlan(event: PlanEvent, deps?: HandlerDeps): Promise<PlanLam
await downloadS3ObjectToFile(s3, event.ProjectS3Uri, projectArchive);
await untarDirectory(projectArchive, projectDir);
const config: DistributedRenderConfig = {
...event.Config,
};
const result: PlanResult = await primitive(projectDir, config, planDir);
// Upload the planDir as a single tarball. Step Functions cannot pass
@@ -403,6 +416,7 @@ async function handlePlanV2(
deps?: HandlerDeps,
): Promise<Extract<PlanLambdaResult, { PlanProtocol: "v2" }>> {
const started = Date.now();
const config = planConfig(event, deps);
const s3 = deps?.s3 ?? getS3Client();
const primitive = deps?.primitives?.planV2WithPublisher ?? planV2WithPublisher;
if (!deps?.skipChromeResolution && !process.env.PRODUCER_HEADLESS_SHELL_PATH) {
@@ -420,7 +434,7 @@ async function handlePlanV2(
planOutputS3Prefix: event.PlanOutputS3Prefix,
temporaryRoot: work,
});
const manifest: PlanV2Manifest = await primitive(projectDir, { ...event.Config }, publisher, {
const manifest: PlanV2Manifest = await primitive(projectDir, config, publisher, {
stagingParentDir: work,
});
packages/aws-lambda/src/sdk/validateConfig.test.ts
+10
View File
@@ -121,6 +121,16 @@ describe("validateDistributedRenderConfig", () => {
{ ...VALID, maxParallelChunks: 0 } satisfies SerializableDistributedRenderConfig,
"config.maxParallelChunks",
],
[
"workerTimeoutSeconds 0",
{ ...VALID, workerTimeoutSeconds: 0 } satisfies SerializableDistributedRenderConfig,
"config.workerTimeoutSeconds",
],
[
"fractional workerConcurrency",
{ ...VALID, workerConcurrency: 1.5 } satisfies SerializableDistributedRenderConfig,
"config.workerConcurrency",
],
[
"unknown runtimeCap",
{
packages/cli/src/commands/lambda.ts
+10 -2
View File
@@ -130,8 +130,16 @@ export default defineCommand({
format: { type: "string", description: "mp4 | mov | png-sequence | webm (default: mp4)" },
codec: { type: "string", description: "h264 | h265 (mp4 only)" },
quality: { type: "string", description: "draft | standard | high" },
"chunk-size": { type: "string", description: "Frames per chunk (default: 240)" },
"max-parallel-chunks": { type: "string", description: "Max concurrent chunks (default: 16)" },
"chunk-size": {
type: "string",
description:
"Frames per chunk (default: sized so each chunk fits the Lambda timeout, one wave across --concurrency)",
},
"max-parallel-chunks": {
type: "string",
description:
"Max chunks per render (default: one wave across the deploy's --concurrency, widened to fit the Lambda timeout)",
},
"target-chunk-frames": {
type: "string",
description:
packages/cli/src/commands/lambda/deploy.ts
+1
View File
@@ -86,6 +86,7 @@ export async function runDeploy(args: Partial<DeployArgs> = {}): Promise<void> {
stateMachineArn: outputs.stateMachineArn,
functionName: outputs.functionName,
lambdaMemoryMb: resolved.lambdaMemoryMb,
reservedConcurrency: resolved.reservedConcurrency,
deployedAt: new Date().toISOString(),
});
packages/cli/src/commands/lambda/render-batch.test.ts
+4
View File
@@ -167,4 +167,8 @@ describe("buildLambdaBatchRenderConfig — aspect-agnostic wire threading", () =
expect(config.outputResolution).toBe("portrait-4k");
expect(config.outputResolutionAspectAgnostic).toBeUndefined();
});
it("hands the deploy's reserved concurrency to the planner", () => {
expect(buildLambdaBatchRenderConfig(baseArgs, 64).workerConcurrency).toBe(64);
});
});
packages/cli/src/commands/lambda/render-batch.ts
+6 -1
View File
@@ -217,7 +217,10 @@ export async function runRenderBatch(args: RenderBatchArgs): Promise<void> {
failCommand();
}
const config: SerializableDistributedRenderConfig = buildLambdaBatchRenderConfig(args);
const config: SerializableDistributedRenderConfig = buildLambdaBatchRenderConfig(
args,
stack.reservedConcurrency,
);
// Deploy the site once and reuse it across every entry. --site-id and
// --dry-run both skip the deploy via a synthesised handle.
@@ -354,6 +357,7 @@ function makePlaceholderSiteHandle(siteId: string, bucketName: string): SiteHand
*/
export function buildLambdaBatchRenderConfig(
args: RenderBatchArgs,
workerConcurrency?: number,
): SerializableDistributedRenderConfig {
const config: SerializableDistributedRenderConfig = {
fps: args.fps,
@@ -366,6 +370,7 @@ export function buildLambdaBatchRenderConfig(
chunkSize: args.chunkSize,
maxParallelChunks: args.maxParallelChunks,
targetChunkFrames: args.targetChunkFrames,
workerConcurrency,
runtimeCap: "lambda",
};
if (args.outputResolutionAspectAgnostic) {
packages/cli/src/commands/lambda/render.test.ts
+16
View File
@@ -10,6 +10,7 @@
*/
import { describe, expect, it } from "vitest";
import lambdaCommand from "../lambda.js";
import { buildLambdaRenderConfig, type RenderArgs } from "./render.js";
const BASE_ARGS: RenderArgs = {
@@ -55,3 +56,18 @@ describe("buildLambdaRenderConfig — aspect-agnostic wire threading", () => {
expect(config.variables).toEqual({ alice: "hello" });
});
});
describe("lambda render chunk sizing", () => {
it("hands the deploy's reserved concurrency to the planner", () => {
expect(buildLambdaRenderConfig(BASE_ARGS, undefined, 64).workerConcurrency).toBe(64);
expect(buildLambdaRenderConfig(BASE_ARGS, undefined).workerConcurrency).toBeUndefined();
});
it("help no longer claims --chunk-size defaults to 240", () => {
const chunkSizeArg = lambdaCommand.args
? Reflect.get(lambdaCommand.args, "chunk-size")
: undefined;
expect(chunkSizeArg?.description).not.toContain("240");
expect(chunkSizeArg?.description).toContain("timeout");
});
});
packages/cli/src/commands/lambda/render.ts
+7 -1
View File
@@ -124,7 +124,11 @@ export async function runRender(args: RenderArgs): Promise<void> {
}
}
const config: SerializableDistributedRenderConfig = buildLambdaRenderConfig(args, variables);
const config: SerializableDistributedRenderConfig = buildLambdaRenderConfig(
args,
variables,
stack.reservedConcurrency,
);
// When the caller passes only --site-id, synthesise the minimum-shape
// SiteHandle pointing at the deterministic content-addressed key. The
@@ -241,6 +245,7 @@ function sleep(ms: number): Promise<void> {
export function buildLambdaRenderConfig(
args: RenderArgs,
variables: Record<string, unknown> | undefined,
workerConcurrency?: number,
): SerializableDistributedRenderConfig {
const config: SerializableDistributedRenderConfig = {
fps: args.fps,
@@ -253,6 +258,7 @@ export function buildLambdaRenderConfig(
chunkSize: args.chunkSize,
maxParallelChunks: args.maxParallelChunks,
targetChunkFrames: args.targetChunkFrames,
workerConcurrency,
runtimeCap: "lambda",
variables,
};
packages/cli/src/commands/lambda/state.ts
+1
View File
@@ -23,6 +23,7 @@ export interface StackOutputs {
functionName: string;
/** Lambda memory MB used during deploy; carried so cost math doesn't have to re-derive it. */
lambdaMemoryMb: number;
reservedConcurrency?: number;
deployedAt: string;
}
packages/gcp-cloud-run/src/server.test.ts
+1
View File
@@ -586,6 +586,7 @@ describe("createApp HTTP mapping", () => {
["VIDEO_EXTRACTION_FAILED", 500],
["FONT_FETCH_FAILED", 400],
["FONT_FETCH_UNAVAILABLE", 500],
["CHUNK_EXCEEDS_WORKER_TIMEOUT", 400],
] as const)("routes producer code %s with HTTP %s", async (code, expectedStatus) => {
const gcs = new FakeGcs();
await seedPlanTar(gcs, "gs://b/renders/r1/plan.tar.gz", PLAN_HASH);
packages/gcp-cloud-run/src/server.ts
+3 -1
View File
@@ -194,7 +194,8 @@ function normalizeTerminalErrorName(error: unknown): void {
candidate.code === "FONT_FETCH_UNAVAILABLE" ||
candidate.code === "VIDEO_SOURCE_UNRENDERABLE" ||
candidate.code === "VIDEO_EXTRACTION_FAILED" ||
candidate.code === "INVALID_VIDEO_METADATA"
candidate.code === "INVALID_VIDEO_METADATA" ||
candidate.code === "CHUNK_EXCEEDS_WORKER_TIMEOUT"
) {
candidate.name = candidate.code;
}
@@ -913,6 +914,7 @@ const NON_RETRYABLE_ERROR_NAMES = new Set([
"PLAN_V2_INTEGRITY_UNRECOVERABLE",
"VIDEO_SOURCE_UNRENDERABLE",
"INVALID_VIDEO_METADATA",
"CHUNK_EXCEEDS_WORKER_TIMEOUT",
"NOT_MEDIA_PAYLOAD",
// 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
packages/producer/src/distributed.ts
+2
View File
@@ -48,6 +48,8 @@ export {
PLAN_DIR_SIZE_LIMIT_BYTES,
PLAN_PROJECT_DIR_SKIP_SEGMENTS,
// Error codes + classes
CHUNK_EXCEEDS_WORKER_TIMEOUT,
ChunkExceedsWorkerTimeoutError,
FORMAT_NOT_SUPPORTED_IN_DISTRIBUTED,
FormatNotSupportedInDistributedError,
PLAN_TOO_LARGE,
packages/producer/src/services/distributed/plan.test.ts
+129
View File
@@ -21,18 +21,23 @@ import { tmpdir } from "node:os";
import { join } from "node:path";
import { applyConcreteGpuScreenshotClamp, buildChromeArgs } from "@hyperframes/engine";
import { recomputePlanHashFromPlanDir } from "../render/stages/freezePlan.js";
import { defaultLogger } from "../../logger.js";
import { RenderQualityError } from "../renderOrchestrator.js";
import { CURRENT_PLAN_PROTOCOL } from "./planProtocol.js";
import {
applyDistributedAudioWarningPolicy,
buildLocalExecutionPlan,
buildChunkSlices,
CHUNK_EXCEEDS_WORKER_TIMEOUT,
ChunkExceedsWorkerTimeoutError,
DEFAULT_CHUNK_SIZE,
DEFAULT_MAX_PARALLEL_CHUNKS,
estimateChunkSeconds,
MIN_CHUNK_SIZE,
plan,
resolveDistributedEngineConfig,
resolveChunkPlan,
resolveMaxParallelChunks,
} from "./plan.js";
import { buildSyntheticRenderJob } from "./shared.js";
@@ -180,6 +185,108 @@ describe("distributed synthetic render job", () => {
});
});
describe("resolveMaxParallelChunks", () => {
const MP_1080P = (1920 * 1080) / 1e6;
const THIRTY_MIN_30FPS = 54000;
const LAMBDA_LIMIT = 900;
const recordingLog = () => {
const warnings: string[] = [];
return { log: { ...defaultLogger, warn: (m: string) => warnings.push(m) }, warnings };
};
it("sizes 30 min of 1080p30 so every chunk fits the Lambda limit with margin", () => {
const { log } = recordingLog();
const maxParallel = resolveMaxParallelChunks(
{ workerTimeoutSeconds: LAMBDA_LIMIT, workerConcurrency: 8 },
THIRTY_MIN_30FPS,
MP_1080P,
log,
);
const { chunkCount, effectiveChunkSize } = resolveChunkPlan(
THIRTY_MIN_30FPS,
undefined,
maxParallel,
);
expect(chunkCount).toBeGreaterThan(DEFAULT_MAX_PARALLEL_CHUNKS);
expect(estimateChunkSeconds(effectiveChunkSize, MP_1080P)).toBeLessThanOrEqual(
LAMBDA_LIMIT * 0.75,
);
// The old fixed 16 planned chunks with no headroom left under the limit.
const old = resolveChunkPlan(THIRTY_MIN_30FPS, undefined, DEFAULT_MAX_PARALLEL_CHUNKS);
expect(estimateChunkSeconds(old.effectiveChunkSize, MP_1080P)).toBeGreaterThan(
LAMBDA_LIMIT * 0.75,
);
});
it("tells the user to raise --concurrency when the fitting chunk count exceeds it", () => {
const { log, warnings } = recordingLog();
const maxParallel = resolveMaxParallelChunks(
{ workerTimeoutSeconds: LAMBDA_LIMIT, workerConcurrency: 8 },
THIRTY_MIN_30FPS,
MP_1080P,
log,
);
expect(warnings).toHaveLength(1);
expect(warnings[0]).toContain(`--concurrency ${maxParallel}`);
});
it("goes one wave wide when concurrency allows, up to the 256 ceiling", () => {
const { log, warnings } = recordingLog();
const args = [3 * THIRTY_MIN_30FPS, MP_1080P, log] as const;
expect(
resolveMaxParallelChunks(
{ workerTimeoutSeconds: LAMBDA_LIMIT, workerConcurrency: 200 },
...args,
),
).toBe(200);
expect(
resolveMaxParallelChunks(
{ workerTimeoutSeconds: LAMBDA_LIMIT, workerConcurrency: 1000 },
...args,
),
).toBe(256);
expect(warnings).toEqual([]);
});
it("keeps short renders to few chunks so per-chunk overhead does not dominate", () => {
const { log } = recordingLog();
const worker = { workerTimeoutSeconds: LAMBDA_LIMIT, workerConcurrency: 256 };
expect(resolveMaxParallelChunks(worker, 60, MP_1080P, log)).toBe(1);
expect(resolveMaxParallelChunks(worker, 300, MP_1080P, log)).toBeLessThanOrEqual(3);
expect(resolveMaxParallelChunks(worker, 3600, MP_1080P, log)).toBeLessThan(64);
});
it("explicit chunking flags win and no worker inputs keeps the old default", () => {
const worker = { workerTimeoutSeconds: LAMBDA_LIMIT, workerConcurrency: 200 };
const args = [THIRTY_MIN_30FPS, MP_1080P] as const;
expect(resolveMaxParallelChunks({ ...worker, maxParallelChunks: 4 }, ...args)).toBe(4);
expect(resolveMaxParallelChunks({ ...worker, chunkSize: 240 }, ...args)).toBe(
DEFAULT_MAX_PARALLEL_CHUNKS,
);
expect(resolveMaxParallelChunks({ ...worker, targetChunkFrames: 300 }, ...args)).toBe(
DEFAULT_MAX_PARALLEL_CHUNKS,
);
expect(resolveMaxParallelChunks({}, ...args)).toBe(DEFAULT_MAX_PARALLEL_CHUNKS);
});
it("fails fast with a typed error when even 256 chunks cannot fit the limit", () => {
const eightHours = 8 * 60 * 60 * 30;
let caught: unknown;
try {
resolveMaxParallelChunks({ workerTimeoutSeconds: LAMBDA_LIMIT }, eightHours, MP_1080P);
} catch (err) {
caught = err;
}
expect(caught).toBeInstanceOf(ChunkExceedsWorkerTimeoutError);
expect((caught as ChunkExceedsWorkerTimeoutError).code).toBe(CHUNK_EXCEEDS_WORKER_TIMEOUT);
expect((caught as Error).message).toContain("900 s worker time limit");
// A limit shorter than one chunk's fixed overhead can never fit.
expect(() => resolveMaxParallelChunks({ workerTimeoutSeconds: 30 }, 300, MP_1080P)).toThrow(
ChunkExceedsWorkerTimeoutError,
);
});
});
describe("resolveChunkPlan", () => {
it("returns 1 chunk when totalFrames fits in configChunkSize", () => {
const result = resolveChunkPlan(60, 240, 16);
@@ -388,6 +495,28 @@ describe("plan() defaults", () => {
expect(DEFAULT_MAX_PARALLEL_CHUNKS).toBe(16);
expect(MIN_CHUNK_SIZE).toBe(10);
});
it("sizes chunks from the worker inputs the adapter passes", async () => {
const config = { fps: 30, width: 320, height: 240, format: "mp4" } as const;
const fixed = await plan(projectDir, config, join(runRoot, "plan-fixed-16"));
const sized = await plan(
projectDir,
{ ...config, workerTimeoutSeconds: 900, workerConcurrency: 8 },
join(runRoot, "plan-worker-sized"),
);
expect(fixed.chunkCount).toBe(3);
expect(sized.chunkCount).toBe(1);
}, 30_000);
it("rejects at plan time when no chunk can finish inside the worker limit", async () => {
await expect(
plan(
projectDir,
{ fps: 30, width: 320, height: 240, format: "mp4", workerTimeoutSeconds: 30 },
join(runRoot, "plan-too-slow"),
),
).rejects.toBeInstanceOf(ChunkExceedsWorkerTimeoutError);
}, 30_000);
});
describe("plan() — golden planDir + planHash determinism", () => {
packages/producer/src/services/distributed/plan.ts
+118 -14
View File
@@ -170,8 +170,15 @@ export interface DistributedRenderConfig {
* old 8-second-GOP behavior.
*/
chunkSize?: number;
/** Default `16`. Caps long renders to fewer-but-longer chunks for operational fairness. */
/**
* Caps the chunk count. When omitted, chunks are sized from `workerConcurrency` and
* `workerTimeoutSeconds` (see {@link resolveMaxParallelChunks}); without those, `16`.
*/
maxParallelChunks?: number;
/** Seconds a chunk worker may run before the platform kills it (Lambda: the function timeout). */
workerTimeoutSeconds?: number;
/** How many chunk workers can run at once (Lambda: reserved concurrency). */
workerConcurrency?: number;
/**
* Upper bound on frames-per-chunk, in frames. Optional; when omitted (the
* default) chunk sizing is unchanged. When set, chunking targets the fewest
@@ -371,9 +378,8 @@ export const PLAN_PROJECT_DIR_SKIP_SEGMENTS: ReadonlySet<string> = new Set([
]);
/**
* Default chunk size in frames (~8s @ 30fps; fits Lambda's 15-min cap).
* Used when the caller explicitly passes this value. When `chunkSize` is
* `undefined`, `plan()` auto-sizes from `maxParallelChunks` instead.
* Chunk size in frames (~8s @ 30fps) for callers that want a fixed GOP.
* `plan()` never applies it on its own; an undefined `chunkSize` auto-sizes.
*/
export const DEFAULT_CHUNK_SIZE = 240;
/** Default cap on parallel chunks for operational fairness across renders. */
@@ -385,6 +391,98 @@ export const DEFAULT_MAX_PARALLEL_CHUNKS = 16;
* outweighs the parallelism gain on tiny renders.
*/
export const MIN_CHUNK_SIZE = 10;
/** Most chunks one render may plan; `validateDistributedRenderConfig` enforces it on `maxParallelChunks`. */
export const MAX_PARALLEL_CHUNKS_CEILING = 256;
// 0.11 s per megapixel-frame (~0.23 s per 1080p frame) was measured on one simple 1080p fixture on a
// 10 GB Lambda, fixed cost included; heavier compositions cost more, hence the margin and overhead.
const ESTIMATED_SECONDS_PER_MEGAPIXEL_FRAME = 0.11;
const ESTIMATED_CHUNK_OVERHEAD_SECONDS = 60;
/** Share of the worker time limit a chunk's predicted time may use. */
const WORKER_TIMEOUT_MARGIN = 0.75;
/** Predicted seconds for one chunk of `frames` frames at `megapixels` output pixels per frame. */
export function estimateChunkSeconds(frames: number, megapixels: number): number {
return (
ESTIMATED_CHUNK_OVERHEAD_SECONDS + frames * ESTIMATED_SECONDS_PER_MEGAPIXEL_FRAME * megapixels
);
}
/** Non-retryable: even {@link MAX_PARALLEL_CHUNKS_CEILING} chunks cannot fit the worker time limit. */
export const CHUNK_EXCEEDS_WORKER_TIMEOUT = "CHUNK_EXCEEDS_WORKER_TIMEOUT";
export class ChunkExceedsWorkerTimeoutError extends Error {
// fallow-ignore-next-line unused-class-member
readonly code: typeof CHUNK_EXCEEDS_WORKER_TIMEOUT = CHUNK_EXCEEDS_WORKER_TIMEOUT;
constructor(predictedChunkSeconds: number, workerTimeoutSeconds: number) {
super(
`[plan] each chunk is predicted to take about ${predictedChunkSeconds} s even split ` +
`${MAX_PARALLEL_CHUNKS_CEILING} ways, over the ${workerTimeoutSeconds * WORKER_TIMEOUT_MARGIN} s ` +
`budget (${WORKER_TIMEOUT_MARGIN * 100}% of the ${workerTimeoutSeconds} s worker time limit) ` +
`(${CHUNK_EXCEEDS_WORKER_TIMEOUT}). Shorten the video or lower its ` +
`resolution, raise the worker time limit, or pass an explicit chunkSize or ` +
`maxParallelChunks to override the estimate.`,
);
this.name = "ChunkExceedsWorkerTimeoutError";
}
}
/**
* The `maxParallelChunks` a plan uses: explicit chunking flags win; otherwise one wave across
* `workerConcurrency`, widened so each chunk fits `workerTimeoutSeconds`; with neither, 16.
*/
export function resolveMaxParallelChunks(
config: Pick<
DistributedRenderConfig,
| "maxParallelChunks"
| "chunkSize"
| "targetChunkFrames"
| "workerTimeoutSeconds"
| "workerConcurrency"
>,
totalFrames: number,
/** Output pixels per frame, after supersampling. */
megapixels: number,
log: ProducerLogger = defaultLogger,
): number {
if (config.maxParallelChunks !== undefined) return config.maxParallelChunks;
const { workerTimeoutSeconds, workerConcurrency } = config;
const pinsChunking = config.chunkSize !== undefined || config.targetChunkFrames !== undefined;
if (pinsChunking || (workerTimeoutSeconds === undefined && workerConcurrency === undefined)) {
return DEFAULT_MAX_PARALLEL_CHUNKS;
}
const secondsPerFrame = ESTIMATED_SECONDS_PER_MEGAPIXEL_FRAME * megapixels;
const wave = Math.min(
workerConcurrency ?? DEFAULT_MAX_PARALLEL_CHUNKS,
MAX_PARALLEL_CHUNKS_CEILING,
);
const minFramesPerChunk = Math.ceil(ESTIMATED_CHUNK_OVERHEAD_SECONDS / secondsPerFrame);
const wide = Math.max(1, Math.min(wave, Math.floor(totalFrames / minFramesPerChunk)));
if (workerTimeoutSeconds === undefined) return wide;
const maxFramesPerChunk = Math.floor(
(workerTimeoutSeconds * WORKER_TIMEOUT_MARGIN - ESTIMATED_CHUNK_OVERHEAD_SECONDS) /
secondsPerFrame,
);
const needed = maxFramesPerChunk < 1 ? Infinity : Math.ceil(totalFrames / maxFramesPerChunk);
if (needed > MAX_PARALLEL_CHUNKS_CEILING) {
const framesAtCeiling = Math.ceil(totalFrames / MAX_PARALLEL_CHUNKS_CEILING);
throw new ChunkExceedsWorkerTimeoutError(
Math.ceil(estimateChunkSeconds(framesAtCeiling, megapixels)),
workerTimeoutSeconds,
);
}
if (needed > wave) {
log.warn(
`[plan] ${needed} chunks are needed to keep each under the ${workerTimeoutSeconds} s ` +
`worker time limit, but the plan assumes ${wave} workers at once, so the render runs ` +
`in ${Math.ceil(needed / wave)} waves. Raise --concurrency to ${needed} to render in one ` +
`wave (Lambda: hyperframes lambda deploy --concurrency ${needed}).`,
{ needed, workerConcurrency: wave, workerTimeoutSeconds },
);
}
return Math.max(wide, needed);
}
/**
* Default hard ceiling on `<planDir>/` size in bytes. 2 GB fits inside
* AWS Lambda's 10 GB `/tmp` alongside the chunk worker's captured frames
@@ -1085,6 +1183,21 @@ export async function buildLocalExecutionPlan(
}
}
// ── Chunking decisions ── (before extraction so an impossible plan fails fast)
const maxParallel = resolveMaxParallelChunks(
config,
totalFrames,
(compileResult.outputWidth * compileResult.outputHeight) / 1e6,
log,
);
const { chunkCount, effectiveChunkSize } = resolveChunkPlan(
totalFrames,
config.chunkSize,
maxParallel,
config.targetChunkFrames,
);
const chunks = buildChunkSlices(totalFrames, chunkCount, effectiveChunkSize);
// The compiled tree is now stable. If it already exceeds the final plan
// budget, extraction/audio/freeze can only add retained bytes, so fail before
// generating a multi-GiB frame tree.
@@ -1198,16 +1311,7 @@ export async function buildLocalExecutionPlan(
renameSync(audioResult.audioOutputPath, planAudioPath);
}
// ── Chunking decisions + locked config ──
const maxParallel = config.maxParallelChunks ?? DEFAULT_MAX_PARALLEL_CHUNKS;
const { chunkCount, effectiveChunkSize } = resolveChunkPlan(
totalFrames,
config.chunkSize,
maxParallel,
config.targetChunkFrames,
);
const chunks = buildChunkSlices(totalFrames, chunkCount, effectiveChunkSize);
// ── Locked config ──
const ffmpegVersion = await readFfmpegVersion();
const producerVersion = readProducerVersion();
const fontSnapshotSha = await readFontSnapshotSha();
packages/producer/src/services/distributed/renderConfigValidation.ts
+8 -2
View File
@@ -24,7 +24,7 @@ import {
} from "@hyperframes/engine";
import { DISTRIBUTED_FPS, isDistributedFps } from "@hyperframes/core";
import { type DistributedFormat } from "./shared.js";
import { type DistributedRenderConfig } from "./plan.js";
import { type DistributedRenderConfig, MAX_PARALLEL_CHUNKS_CEILING } from "./plan.js";
/**
* `DistributedRenderConfig` minus the runtime-only fields (`logger`,
@@ -63,7 +63,6 @@ const ALLOWED_HDR_MODES = ["auto", "force-sdr"] as const;
const MAX_DIMENSION = 7680;
const MIN_DIMENSION = 16;
const MAX_CHUNK_SIZE = 3600;
const MAX_PARALLEL_CHUNKS_CEILING = 256;
/**
* Throw an `InvalidConfigError` if `config` is not a valid
@@ -187,6 +186,13 @@ export function validateDistributedRenderConfig(
}
}
for (const field of ["workerTimeoutSeconds", "workerConcurrency"] as const) {
const value = config[field];
if (value !== undefined && (!Number.isInteger(value) || value < 1)) {
throw new InvalidConfigError(`config.${field}`, `must be a positive integer; got ${value}`);
}
}
if (config.runtimeCap !== undefined && !ALLOWED_RUNTIME_CAPS.includes(config.runtimeCap)) {
throw new InvalidConfigError(
"config.runtimeCap",