fix(engine): encode with the same x264 and x265 thread count on any host size (#6127)

* fix(engine): encode with the same x264 and x265 thread count on any host size

libx264 and libx265 pick their thread count from the host core count when
ffmpeg is not told one, and the bitstream depends on it, so the same frames
encoded to different bytes on 8, 16 and 32 vCPU nodes and chunks of one
render could differ. Both encoders now pass the thread count from the CPU
quota (Node reads the container limit), which is fixed per pod.

* chore(engine): raise the packed size budget for the encoder thread module

packages/engine/src/index.ts

@@ -310,6 +310,7 @@ export {
 // ── Utilities ──────────────────────────────────────────────────────────────────
 export { quantizeTimeToFrame, MEDIA_VISUAL_STYLE_PROPERTIES } from "@hyperframes/core";
 export { frameFileExtension } from "./services/frameCapture.js";
+export { appendSoftwareEncoderThreadsArg } from "./services/softwareEncoderThreads.js";
 export { frameSequencePath } from "./services/extractedFrameIndex.js";
 export type { MotionBlurOptions, MotionBlurBlendSpace } from "./services/motionBlur.js";
 export type { PageMotionBlurOff } from "./types.js";

packages/engine/src/services/chunkEncoder.ts

@@ -47,6 +47,7 @@ import { extractAudioMetadata } from "../utils/ffprobe.js";
 import { type Fps, fpsToFfmpegArg, fpsToNumber } from "@hyperframes/core";
 import type { EncoderOptions, EncodeResult, MuxResult } from "./chunkEncoder.types.js";
 import { appendVp9CpuUsedArg } from "./vp9Options.js";
+import { appendSoftwareEncoderThreadsArg } from "./softwareEncoderThreads.js";
 import { appendRenderProvenanceArgs } from "../utils/renderProvenance.js";
 
 export type { EncoderOptions, EncodeResult, MuxResult } from "./chunkEncoder.types.js";
@@ -326,6 +327,7 @@ export function buildEncoderArgs(
     } else {
       const encoderName = codec === "h264" ? "libx264" : "libx265";
       args.push("-c:v", encoderName, "-preset", preset);
+      appendSoftwareEncoderThreadsArg(args);
       if (bitrate) args.push("-b:v", bitrate);
       else args.push("-crf", String(quality));
 

packages/engine/src/services/softwareEncoderThreads.test.ts

@@ -0,0 +1,98 @@
+import { spawnSync } from "node:child_process";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { getFfmpegBinary } from "../utils/ffmpegBinaries.js";
+
+const host = vi.hoisted(() => ({ quota: 6, cores: 8 }));
+vi.mock("node:os", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("node:os")>()),
+  availableParallelism: () => host.quota,
+  cpus: () => Array.from({ length: host.cores }, () => ({}) as never),
+}));
+
+import { buildEncoderArgs } from "./chunkEncoder.js";
+import type { EncoderOptions } from "./chunkEncoder.types.js";
+import { appendSoftwareEncoderThreadsArg } from "./softwareEncoderThreads.js";
+import { buildStreamingArgs } from "./streamingEncoder.js";
+
+const FFMPEG = getFfmpegBinary();
+const HAS_TASKSET = process.platform === "linux" && spawnSync("taskset", ["-V"]).status === 0;
+const HAS_FFMPEG = spawnSync(FFMPEG, ["-version"]).status === 0;
+const QUIET = ["-v", "error"];
+const SOURCE = [
+  "-f",
+  "lavfi",
+  "-i",
+  "testsrc2=size=320x240:rate=30:duration=2,noise=alls=12:allf=t",
+];
+
+const options = (codec: "h264" | "h265"): EncoderOptions => ({
+  fps: { num: 30, den: 1 },
+  width: 320,
+  height: 240,
+  codec,
+  preset: "medium",
+  quality: 23,
+});
+
+const threadsOf = (args: string[]) => args[args.indexOf("-threads") + 1];
+
+describe("software encoder thread count", () => {
+  it("takes the CPU quota and floors at one", () => {
+    const args: string[] = [];
+    appendSoftwareEncoderThreadsArg(args, 6.9);
+    appendSoftwareEncoderThreadsArg(args, 0);
+    expect(args).toEqual(["-threads", "6", "-threads", "1"]);
+  });
+
+  it.each(["h264", "h265"] as const)(
+    "pins %s threads in both encoders to the quota, not the host",
+    (codec) => {
+      const seen = new Set<string | undefined>();
+      for (const cores of [8, 16, 64]) {
+        host.cores = cores;
+        seen.add(threadsOf(buildEncoderArgs(options(codec), SOURCE, "out.mp4")));
+        seen.add(
+          threadsOf(
+            buildStreamingArgs({ ...options(codec), preset: "medium" } as never, "out.mp4"),
+          ),
+        );
+      }
+      expect([...seen]).toEqual(["6"]);
+    },
+  );
+});
+
+describe.skipIf(!HAS_TASKSET || !HAS_FFMPEG)("encoded bytes across host sizes", () => {
+  const dirs: string[] = [];
+  afterEach(() => {
+    for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true });
+  });
+
+  /** Encode with the production args as a host with `cpus` cores would run them. */
+  function videoMd5(codec: "h264" | "h265", cpus: string): string {
+    const dir = mkdtempSync(join(tmpdir(), "hf-threads-"));
+    dirs.push(dir);
+    const out = join(dir, "out.mp4");
+    const args = buildEncoderArgs(options(codec), SOURCE, out);
+    expect(spawnSync("taskset", ["-c", cpus, FFMPEG, ...QUIET, ...args]).status).toBe(0);
+    const md5 = spawnSync(
+      FFMPEG,
+      ["-v", "error", "-i", out, "-map", "0:v", "-c", "copy", "-f", "md5", "-"],
+      {
+        encoding: "utf-8",
+      },
+    );
+    return md5.stdout.trim();
+  }
+
+  it.each(["h264", "h265"] as const)(
+    "gives %s identical bytes on a 2-core and a 7-core host",
+    (codec) => {
+      expect(videoMd5(codec, "0-1")).toBe(videoMd5(codec, "0-6"));
+    },
+    120_000,
+  );
+});

packages/engine/src/services/softwareEncoderThreads.ts

@@ -0,0 +1,9 @@
+import { availableParallelism } from "node:os";
+
+/** x264/x265 bitstreams depend on the thread count, so it comes from the CPU quota, never ffmpeg's host-core guess. */
+export function appendSoftwareEncoderThreadsArg(
+  args: string[],
+  parallelism: number = availableParallelism(),
+): void {
+  args.push("-threads", String(Math.max(1, Math.floor(parallelism))));
+}

packages/engine/src/services/streamingEncoder.ts

@@ -46,6 +46,7 @@ import {
 import { DEFAULT_CONFIG, type EngineConfig } from "../config.js";
 import { fpsToFfmpegArg, fpsToNumber, type Fps } from "@hyperframes/core";
 import { appendVp9CpuUsedArg } from "./vp9Options.js";
+import { appendSoftwareEncoderThreadsArg } from "./softwareEncoderThreads.js";
 import { appendRenderProvenanceArgs } from "../utils/renderProvenance.js";
 
 import { appendLockedGopArgs, lockedGopCodecParams, resolveLockedGopSize } from "./chunkEncoder.js";
@@ -338,6 +339,7 @@ export function buildStreamingArgs(
     } else {
       const encoderName = codec === "h264" ? "libx264" : "libx265";
       args.push("-c:v", encoderName, "-preset", preset);
+      appendSoftwareEncoderThreadsArg(args);
       if (bitrate) args.push("-b:v", bitrate);
       else args.push("-crf", String(quality));
 

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

@@ -26,7 +26,7 @@ import {
   statSync,
   writeFileSync,
 } from "node:fs";
-import { tmpdir } from "node:os";
+import { availableParallelism, tmpdir } from "node:os";
 import { join } from "node:path";
 import {
   applyFaststart,
@@ -38,7 +38,7 @@ import {
 } from "@hyperframes/engine";
 import { padOrTrimAudioToVideoFrameCount } from "../render/audioPadTrim.js";
 import type { ChunkSliceJson } from "../render/stages/freezePlan.js";
-import { assemble, cfrRateControlArgs } from "./assemble.js";
+import { assemble, cfrEncodeArgs, cfrRateControlArgs } from "./assemble.js";
 import type { DistributedFormat } from "./shared.js";
 
 let runRoot: string;
@@ -663,6 +663,11 @@ describe("assemble()", () => {
     TIMEOUT_MS,
   );
 
+  it("cfr re-encode pins x264 threads to the CPU quota", () => {
+    const args = cfrEncodeArgs(["-i", "chunk.mp4"], {}, "30");
+    expect(args[args.indexOf("-threads") + 1]).toBe(String(availableParallelism()));
+  });
+
   it("cfr re-encode keeps the chunks' rate control", () => {
     expect(cfrRateControlArgs({})).toEqual(["-preset", "medium", "-crf", "18"]);
     expect(cfrRateControlArgs({ quality: "high" })).toEqual(["-preset", "slow", "-crf", "15"]);

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

@@ -40,6 +40,7 @@ import {
   muxVideoWithAudio,
   resolveConfig,
   runFfmpeg,
+  appendSoftwareEncoderThreadsArg,
 } from "@hyperframes/engine";
 import { fpsToFfmpegArg } from "@hyperframes/core";
 import { defaultLogger, type ProducerLogger } from "../../logger.js";
@@ -267,6 +268,27 @@ function videoInputArgs(ctx: JoinContext, videoSources: readonly string[]): stri
     : concatDemuxerInput(videoSources, ctx.workDir, ctx.fpsArg);
 }
 
+export function cfrEncodeArgs(
+  inputArgs: string[],
+  encoder: CfrRateControlSource,
+  fpsArg: string,
+): string[] {
+  const args = [
+    ...inputArgs,
+    "-c:v",
+    "libx264",
+    ...cfrRateControlArgs(encoder),
+    "-pix_fmt",
+    "yuv420p",
+    "-fps_mode",
+    "cfr",
+    "-r",
+    fpsArg,
+  ];
+  appendSoftwareEncoderThreadsArg(args);
+  return args;
+}
+
 // ── 2c. Optional exact-CFR re-encode ──────────────────────────────────
 // Lands the stream-level `avg_frame_rate` (PTS-derived under stream copy)
 // on the requested rational, for consumers that strict-check it.
@@ -276,18 +298,7 @@ async function reencodeCfr(
   encoder: CfrRateControlSource,
 ): Promise<string> {
   const cfrOutputPath = join(ctx.workDir, `cfr.${ctx.format}`);
-  const cfrArgs = [
-    ...videoInputArgs(ctx, chunkPaths),
-    "-c:v",
-    "libx264",
-    ...cfrRateControlArgs(encoder),
-    "-pix_fmt",
-    "yuv420p",
-    "-fps_mode",
-    "cfr",
-    "-r",
-    ctx.fpsArg,
-  ];
+  const cfrArgs = cfrEncodeArgs(videoInputArgs(ctx, chunkPaths), encoder, ctx.fpsArg);
   appendRenderProvenanceArgs(cfrArgs, cfrOutputPath);
   cfrArgs.push("-y", cfrOutputPath);
   await ctx.timed("cfrMs", () =>

scripts/package-size-budgets.json

@@ -12,7 +12,7 @@
     "unpacked": 3950000
   },
   "@hyperframes/engine": {
-    "packed": 428347,
+    "packed": 428562,
     "unpacked": 1436293
   },
   "@hyperframes/gcp-cloud-run": {