perf(producer): cut distributed chunks by estimated capture cost (#6295)

* perf(producer): cut distributed chunks by estimated capture cost, same chunk count

Auto-sized plans place chunk boundaries so each chunk carries about the same estimated capture
cost (page base plus active video megapixels); chunk count, explicit chunkSize, uniform-cost
compositions and the targetChunkFrames ceiling are unchanged. Frame-cost calibration samples by
cost interval plus the heaviest frame, so a heavy window is always measured.

* fix(producer): hold cost-weighted chunks to the worker time limit's frame cap

The weighted cut could grow a chunk up to MAX_CHUNK_SIZE frames, past the per-chunk frame count resolveMaxParallelChunks sized against the worker time limit. The cut now takes the same cap, from one shared helper, and keeps the equal-frame size as its floor.

* fix(producer): keep the chunk-cut doc comment under 120 characters

* perf(producer): drop the calibration sampler change from the cost-weighted cut

The cut no longer touches how capture cost is measured; that keeps the producer bundle within its size budget and the change to the chunk boundaries alone.

* perf(producer): compute the chunk frame cap inline to keep the bundle small

packages/engine/src/index.ts

@@ -218,6 +218,7 @@ export {
   resolveVideoExtractionWindow,
   rebaseVideoToWindow,
   resolveFinalFrameExtractionWindow,
+  resolveFrameFormat,
   resolveVideoExtractionDuration,
   resolvePlayableVideoDuration,
   extractionFrameCountForDuration,

packages/producer/src/services/distributed/chunkWeights.test.ts

@@ -0,0 +1,208 @@
+import { describe, expect, it } from "bun:test";
+import { buildWeightedChunkSlices, planFrameWeights } from "./chunkWeights.js";
+import {
+  buildChunkSlices,
+  buildPlanChunkSlices,
+  MAX_CHUNK_SIZE,
+  MIN_CHUNK_SIZE,
+  resolveChunkPlan,
+} from "./plan.js";
+import type { PlanVideosJson } from "./shared.js";
+
+type Slices = ReturnType<typeof buildChunkSlices>;
+
+function chunkCosts(weights: Float64Array, slices: Slices) {
+  return slices.map((slice) =>
+    weights.slice(slice.startFrame, slice.endFrame).reduce((sum, weight) => sum + weight, 0),
+  );
+}
+
+function expectExactCover(slices: Slices, totalFrames: number) {
+  expect(slices[0]?.startFrame).toBe(0);
+  expect(slices.at(-1)?.endFrame).toBe(totalFrames);
+  slices.forEach((slice, index) => {
+    expect(slice.index).toBe(index);
+    expect(slice.endFrame).toBeGreaterThan(slice.startFrame);
+    if (index > 0) expect(slice.startFrame).toBe(slices[index - 1]?.endFrame ?? -1);
+  });
+}
+
+function weightsWithHeavyWindow(totalFrames: number, from: number, to: number, heavy: number) {
+  const weights = new Float64Array(totalFrames).fill(1);
+  weights.fill(heavy, from, to);
+  return weights;
+}
+
+function video(id: string, start: number, end: number, hidden?: boolean) {
+  return { id, start, end, hidden } as PlanVideosJson["videos"][number];
+}
+
+function extractedVideo(videoId: string, width: number, height: number, hasAlpha = false) {
+  return {
+    videoId,
+    metadata: { width, height, hasAlpha, videoCodec: "h264" },
+  } as PlanVideosJson["extracted"][number];
+}
+
+describe("buildPlanChunkSlices", () => {
+  it.each([
+    [1500, 18],
+    [660, 16],
+    [121, 12],
+    [50, 16],
+    [1, 16],
+    [54000, 256],
+  ])("uniform cost keeps the equal-frame cut (%d frames, max %d chunks)", (frames, max) => {
+    const { chunkCount, effectiveChunkSize } = resolveChunkPlan(frames, undefined, max);
+    const fullLengthVideo = planFrameWeights({
+      videos: [video("v", 0, frames)],
+      extracted: [extractedVideo("v", 1920, 1080)],
+      totalFrames: frames,
+      fps: 1,
+      width: 1920,
+      height: 1080,
+    });
+    for (const weights of [new Float64Array(frames).fill(0.7), fullLengthVideo]) {
+      expect(buildPlanChunkSlices(weights, {}, chunkCount, effectiveChunkSize)).toEqual(
+        buildChunkSlices(frames, chunkCount, effectiveChunkSize),
+      );
+    }
+  });
+
+  it("balances a heavy window across the same number of chunks", () => {
+    const frames = 1500;
+    const { chunkCount, effectiveChunkSize } = resolveChunkPlan(frames, undefined, 18);
+    const weights = weightsWithHeavyWindow(frames, 700, 783, 4);
+    const slices = buildPlanChunkSlices(weights, {}, chunkCount, effectiveChunkSize);
+
+    expect(slices).toHaveLength(chunkCount);
+    expectExactCover(slices, frames);
+    for (const slice of slices) {
+      expect(slice.endFrame - slice.startFrame).toBeGreaterThanOrEqual(MIN_CHUNK_SIZE);
+    }
+    const costs = chunkCosts(weights, slices);
+    const mean = costs.reduce((sum, cost) => sum + cost, 0) / costs.length;
+    expect(Math.max(...costs)).toBeLessThanOrEqual(mean * 1.15);
+
+    const uniformCosts = chunkCosts(
+      weights,
+      buildChunkSlices(frames, chunkCount, effectiveChunkSize),
+    );
+    expect(Math.max(...uniformCosts)).toBeGreaterThan(mean * 1.5);
+    const heavyChunk = slices.find((slice) => slice.startFrame <= 740 && slice.endFrame > 740);
+    expect((heavyChunk?.endFrame ?? 0) - (heavyChunk?.startFrame ?? 0)).toBeLessThan(
+      effectiveChunkSize,
+    );
+  });
+
+  it("keeps the equal-frame cut when the caller passed a chunkSize", () => {
+    const weights = weightsWithHeavyWindow(1500, 700, 783, 4);
+    expect(buildPlanChunkSlices(weights, { chunkSize: 84 }, 18, 84)).toEqual(
+      buildChunkSlices(1500, 18, 84),
+    );
+  });
+
+  it.each([
+    [25, 16],
+    [11, 16],
+    [12, 1],
+  ])("stays a valid cover on tiny compositions (%d frames, max %d chunks)", (frames, max) => {
+    const { chunkCount, effectiveChunkSize } = resolveChunkPlan(frames, undefined, max);
+    const weights = weightsWithHeavyWindow(frames, 0, 3, 6);
+    const slices = buildPlanChunkSlices(weights, {}, chunkCount, effectiveChunkSize);
+    expect(slices).toHaveLength(chunkCount);
+    expectExactCover(slices, frames);
+    for (const slice of slices) {
+      expect(slice.endFrame - slice.startFrame).toBeGreaterThanOrEqual(
+        Math.floor(frames / chunkCount),
+      );
+    }
+  });
+
+  it("keeps every chunk under MAX_CHUNK_SIZE or the equal cut, whichever is larger", () => {
+    const frames = 9000;
+    const weights = weightsWithHeavyWindow(frames, 0, 100, 50);
+    const slices = buildPlanChunkSlices(weights, {}, 2, 4500);
+    expectExactCover(slices, frames);
+    for (const slice of slices) expect(slice.endFrame - slice.startFrame).toBeLessThanOrEqual(4500);
+
+    const fourChunks = buildPlanChunkSlices(
+      weightsWithHeavyWindow(frames, 0, 100, 200),
+      {},
+      4,
+      2250,
+    );
+    expectExactCover(fourChunks, frames);
+    for (const slice of fourChunks) {
+      expect(slice.endFrame - slice.startFrame).toBeLessThanOrEqual(MAX_CHUNK_SIZE);
+    }
+  });
+
+  it("keeps targetChunkFrames as a per-chunk frame ceiling", () => {
+    const frames = 1500;
+    const { chunkCount, effectiveChunkSize } = resolveChunkPlan(frames, undefined, 18, 120);
+    const weights = weightsWithHeavyWindow(frames, 700, 783, 4);
+    const slices = buildPlanChunkSlices(
+      weights,
+      { targetChunkFrames: 120 },
+      chunkCount,
+      effectiveChunkSize,
+    );
+    expect(slices).toHaveLength(chunkCount);
+    expectExactCover(slices, frames);
+    for (const slice of slices) expect(slice.endFrame - slice.startFrame).toBeLessThanOrEqual(120);
+  });
+});
+
+describe("buildWeightedChunkSlices", () => {
+  it("returns null for uniform cost so the caller keeps the equal-frame cut", () => {
+    expect(buildWeightedChunkSlices(new Float64Array(100).fill(2), 4, 10, 3600)).toBeNull();
+  });
+});
+
+describe("planFrameWeights", () => {
+  const base = { totalFrames: 100, fps: 10, width: 3840, height: 2160 };
+  const uhd = (3840 * 2160) / 1e6;
+
+  it("weights active frames by video megapixels and frame format", () => {
+    const weights = planFrameWeights({
+      ...base,
+      videos: [video("jpg", 1, 3), video("png", 2, 4), video("hidden", 0, 10, true)],
+      extracted: [
+        extractedVideo("jpg", 3840, 2160),
+        extractedVideo("png", 3840, 2160, true),
+        extractedVideo("hidden", 3840, 2160),
+      ],
+    });
+    expect(weights[5]).toBeCloseTo(0.7);
+    expect(weights[15]).toBeCloseTo(0.7 + uhd);
+    expect(weights[25]).toBeCloseTo(0.7 + uhd + 1.5 * uhd);
+    expect(weights[35]).toBeCloseTo(0.7 + 1.5 * uhd);
+    expect(weights[40]).toBeCloseTo(0.7);
+  });
+
+  it("caps megapixels at the composition size and clamps windows to the timeline", () => {
+    const weights = planFrameWeights({
+      ...base,
+      width: 1920,
+      height: 1080,
+      videos: [video("big", -5, 500), video("unextracted", 0, 5)],
+      extracted: [extractedVideo("big", 3840, 2160)],
+    });
+    expect(weights).toHaveLength(100);
+    expect(weights[0]).toBeCloseTo(0.7 + (1920 * 1080) / 1e6);
+    expect(weights[99]).toBeCloseTo(0.7 + (1920 * 1080) / 1e6);
+  });
+
+  it("holds every chunk to the worker time limit's frame cap", () => {
+    const frames = 1500;
+    const { chunkCount, effectiveChunkSize } = resolveChunkPlan(frames, undefined, 18);
+    const weights = weightsWithHeavyWindow(frames, 700, 783, 4);
+    const capped = buildPlanChunkSlices(weights, {}, chunkCount, effectiveChunkSize, 90);
+    expectExactCover(capped, frames);
+    for (const slice of capped) expect(slice.endFrame - slice.startFrame).toBeLessThanOrEqual(90);
+
+    const uncapped = buildPlanChunkSlices(weights, {}, chunkCount, effectiveChunkSize);
+    expect(Math.max(...uncapped.map((s) => s.endFrame - s.startFrame))).toBeGreaterThan(90);
+  });
+});

packages/producer/src/services/distributed/chunkWeights.ts

@@ -0,0 +1,89 @@
+import { resolveFrameFormat } from "@hyperframes/engine";
+import type { ChunkSliceJson } from "../render/stages/freezePlan.js";
+import type { PlanVideosJson } from "./shared.js";
+
+/** Per-frame page cost, in megapixel-frames of jpg capture (measured: 0.03 s base / 0.045 s per MP). */
+const PAGE_BASE_WEIGHT = 0.7;
+/** png frames cost ~1.5x jpg per megapixel. */
+const PNG_WEIGHT_FACTOR = 1.5;
+
+/** Costs whose max/min stay within 5% count as uniform. */
+function isUniformFrameCost(weights: Float64Array): boolean {
+  let lightest = Infinity;
+  let heaviest = 0;
+  for (const weight of weights) {
+    lightest = Math.min(lightest, weight);
+    heaviest = Math.max(heaviest, weight);
+  }
+  return heaviest <= lightest * 1.05;
+}
+
+/** Estimated capture cost of every frame: page base plus each active video's pixels. */
+export function planFrameWeights(input: {
+  videos: PlanVideosJson["videos"];
+  extracted: PlanVideosJson["extracted"];
+  totalFrames: number;
+  fps: number;
+  width: number;
+  height: number;
+}): Float64Array {
+  const { totalFrames, fps } = input;
+  const delta = new Float64Array(totalFrames + 1);
+  const outputPixels = input.width * input.height;
+  const extractedById = new Map(input.extracted.map((entry) => [entry.videoId, entry]));
+  for (const video of input.videos) {
+    const entry = extractedById.get(video.id);
+    if (!entry || video.hidden) continue;
+    const start = Math.max(0, Math.round(video.start * fps));
+    const end = Math.min(totalFrames, Math.round(video.end * fps));
+    if (!(end > start)) continue;
+    const { width, height } = entry.metadata;
+    const factor = resolveFrameFormat(entry.metadata) === "png" ? PNG_WEIGHT_FACTOR : 1;
+    const weight = (Math.min(width * height, outputPixels) / 1e6) * factor;
+    delta[start] = delta[start]! + weight;
+    delta[end] = delta[end]! - weight;
+  }
+  const weights = new Float64Array(totalFrames);
+  let active = 0;
+  for (let frame = 0; frame < totalFrames; frame++) {
+    active += delta[frame]!;
+    weights[frame] = PAGE_BASE_WEIGHT + active;
+  }
+  return weights;
+}
+
+/**
+ * Cut `[0, weights.length)` into `chunkCount` ranges of ~equal summed weight, each within
+ * `[minFrames, maxFrames]` where the frame count allows. `null` when the cost is uniform.
+ */
+export function buildWeightedChunkSlices(
+  weights: Float64Array,
+  chunkCount: number,
+  minFrames: number,
+  maxFrames: number,
+): ChunkSliceJson[] | null {
+  if (isUniformFrameCost(weights)) return null;
+  const totalFrames = weights.length;
+  const total = weights.reduce((sum, weight) => sum + weight, 0);
+  const minSize = Math.max(1, Math.min(minFrames, Math.floor(totalFrames / chunkCount)));
+  const maxSize = Math.max(maxFrames, Math.ceil(totalFrames / chunkCount));
+  const slices: ChunkSliceJson[] = [];
+  let startFrame = 0;
+  let frame = 0;
+  let costBefore = 0;
+  for (let index = 0; index < chunkCount - 1; index++) {
+    const target = ((index + 1) * total) / chunkCount;
+    // Advance while the boundary after `frame` lands nearer the target than the one before it.
+    while (frame < totalFrames && costBefore + weights[frame]! / 2 < target) {
+      costBefore += weights[frame++]!;
+    }
+    const remaining = chunkCount - index - 1;
+    const low = Math.max(startFrame + minSize, totalFrames - remaining * maxSize);
+    const high = Math.min(startFrame + maxSize, totalFrames - remaining * minSize);
+    const endFrame = Math.min(high, Math.max(low, frame));
+    slices.push({ index, startFrame, endFrame });
+    startFrame = endFrame;
+  }
+  slices.push({ index: chunkCount - 1, startFrame, endFrame: totalFrames });
+  return slices;
+}

packages/producer/src/services/distributed/plan.ts

@@ -62,6 +62,7 @@ import {
 } from "../renderOrchestrator.js";
 import { closeFileServerSafely } from "../fileServer.js";
 import { measurePlanSecondsPerFrame } from "./planFrameCost.js";
+import { buildWeightedChunkSlices, planFrameWeights } from "./chunkWeights.js";
 import { runAudioStage } from "../render/stages/audioStage.js";
 import { runCompileStage } from "../render/stages/compileStage.js";
 import {
@@ -431,6 +432,8 @@ export const DEFAULT_MAX_PARALLEL_CHUNKS = 16;
  * outweighs the parallelism gain on tiny renders.
  */
 export const MIN_CHUNK_SIZE = 10;
+/** Largest `chunkSize`; `validateDistributedRenderConfig` enforces it, and cost-weighted chunks stay under it. */
+export const MAX_CHUNK_SIZE = 3600;
 /** 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
@@ -499,6 +502,14 @@ export class ChunkExceedsWorkerTimeoutError extends Error {
   }
 }
 
+/** Frames one chunk can capture inside the worker time limit, after the per-chunk overhead. */
+function framesWithinWorkerTimeout(workerTimeoutSeconds: number, secondsPerFrame: number): number {
+  return Math.floor(
+    (workerTimeoutSeconds * WORKER_TIMEOUT_MARGIN - ESTIMATED_CHUNK_OVERHEAD_SECONDS) /
+      secondsPerFrame,
+  );
+}
+
 /**
  * The `maxParallelChunks` a plan uses: an explicit value, else one wave across `workerConcurrency`
  * (16 if unknown), widened to fit `workerTimeoutSeconds` unless chunkSize/targetChunkFrames pin it.
@@ -531,10 +542,7 @@ export function resolveMaxParallelChunks(
   const wide = Math.max(1, Math.min(wave, Math.floor(totalFrames / minFramesPerChunk)));
   if (workerTimeoutSeconds === undefined) return wide;
 
-  const maxFramesPerChunk = Math.floor(
+  const maxFramesPerChunk = framesWithinWorkerTimeout(workerTimeoutSeconds, secondsPerFrame);
-    (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);
@@ -837,6 +845,29 @@ export function buildChunkSlices(
   return slices;
 }
 
+/**
+ * The plan's chunk cut: auto-sized plans re-cut the same chunk count by estimated capture cost;
+ * explicit `chunkSize` or uniform cost keeps {@link buildChunkSlices}; `targetChunkFrames` and the time limit cap it.
+ */
+export function buildPlanChunkSlices(
+  frameWeights: Float64Array,
+  config: Pick<DistributedRenderConfig, "chunkSize" | "targetChunkFrames">,
+  chunkCount: number,
+  effectiveChunkSize: number,
+  maxFramesPerChunk = Infinity,
+): ChunkSliceJson[] {
+  const weighted =
+    config.chunkSize === undefined
+      ? buildWeightedChunkSlices(
+          frameWeights,
+          chunkCount,
+          MIN_CHUNK_SIZE,
+          Math.min(config.targetChunkFrames ?? MAX_CHUNK_SIZE, MAX_CHUNK_SIZE, maxFramesPerChunk),
+        )
+      : null;
+  return weighted ?? buildChunkSlices(frameWeights.length, chunkCount, effectiveChunkSize);
+}
+
 /**
  * Hash the deterministic-font bundle that ships inside `@hyperframes/producer`.
  * The compiled HTML already inlines per-family `@font-face` data URIs, so the
@@ -1526,6 +1557,14 @@ export async function buildLocalExecutionPlan(
   const digests: PlanFileDigests = options.digests ?? new Map();
   const separateAudio = plansSeparateAudio(config, options);
   const chunksCarryAudio = !separateAudio && carriesChunkAudio(config, options);
+  const frameWeights = planFrameWeights({
+    videos: planVideosJson.videos,
+    extracted: planVideosJson.extracted,
+    totalFrames,
+    fps: config.fps,
+    width,
+    height,
+  });
   stageStarted = Date.now();
   const [audioOutcome, overlapOutcome, frameCostOutcome] = await Promise.allSettled([
     separateAudio
@@ -1580,7 +1619,18 @@ export async function buildLocalExecutionPlan(
     maxParallel,
     config.targetChunkFrames,
   );
-  const chunks = buildChunkSlices(totalFrames, chunkCount, effectiveChunkSize);
+  const chunks = buildPlanChunkSlices(
+    frameWeights,
+    config,
+    chunkCount,
+    effectiveChunkSize,
+    chunkSizingUsesFrameCost(config) && config.workerTimeoutSeconds !== undefined
+      ? framesWithinWorkerTimeout(
+          config.workerTimeoutSeconds,
+          estimateFrameSeconds(megapixels, frameCostOutcome.value),
+        )
+      : undefined,
+  );
   const audioResult = audioOutcome.value;
   if (audioResult?.audioError) {
     applyDistributedAudioWarningPolicy(job, audioResult.audioError, audioResult.audioFailures, log);

packages/producer/src/services/distributed/renderConfigValidation.ts

@@ -24,7 +24,11 @@ import {
 } from "@hyperframes/engine";
 import { DISTRIBUTED_FPS, isDistributedFps } from "@hyperframes/core";
 import { type DistributedFormat } from "./shared.js";
-import { type DistributedRenderConfig, MAX_PARALLEL_CHUNKS_CEILING } from "./plan.js";
+import {
+  type DistributedRenderConfig,
+  MAX_CHUNK_SIZE,
+  MAX_PARALLEL_CHUNKS_CEILING,
+} from "./plan.js";
 
 /**
  * `DistributedRenderConfig` minus the runtime-only fields (`logger`,
@@ -62,7 +66,6 @@ const ALLOWED_HDR_MODES = ["auto", "force-sdr"] as const;
 
 const MAX_DIMENSION = 7680;
 const MIN_DIMENSION = 16;
-const MAX_CHUNK_SIZE = 3600;
 
 /**
  * Throw an `InvalidConfigError` if `config` is not a valid