perf(producer): a cloud plan corrects its audio peak from the mix itself, in parallel (#5981)

* perf(producer): the true-peak correction of long audio encodes in parallel segments

* perf(producer): a cloud plan corrects its audio peak from the mix itself, one lossy pass

* fix(producer): only the PCM-sourced peak correction splits; AAC input stays one encoder

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

@@ -84,6 +84,7 @@ vi.mock("../utils/ffprobe.js", async (importOriginal) => {
 import { RATE_RANGE } from "@hyperframes/core/audio-automation";
 import { AudioFxRenderError } from "./audioFxRender.js";
 import { parseAudioElements, processCompositionAudio } from "./audioMixer.js";
+import { encodeAac } from "./aacEncode.js";
 
 describe("parseAudioElements strict literal timing", () => {
   it.each(["", "   ", "0s", "0abc", "0px", "-1s", "Infinity", "NaN", "0x10"])(
@@ -1485,6 +1486,43 @@ describe("processCompositionAudio", () => {
     expect(prepareArgs).toContain(join(baseDir, "assets", filename));
   });
 
+  it("keeps the float mix the AAC was encoded from when asked to", async () => {
+    const baseDir = mkdtempSync(join(tmpdir(), "hf-audio-base-"));
+    const workDir = mkdtempSync(join(tmpdir(), "hf-audio-work-"));
+    tempDirs.push(baseDir, workDir);
+    writeFileSync(join(baseDir, "a.wav"), "stub");
+    writeFileSync(join(baseDir, "b.wav"), "stub");
+    runFfmpegMock.mockImplementation(async (args: string[]) => {
+      if (args.at(-1)?.endsWith("master-sum.wav")) writeFileSync(args.at(-1)!, "float mix");
+      return { success: true, durationMs: 1, stderr: "", exitCode: 0 };
+    });
+    const track = (id: string, layer: number) => ({
+      id,
+      src: `${id}.wav`,
+      start: 0,
+      end: 2,
+      mediaStart: 0,
+      layer,
+      volume: 1,
+      type: "audio" as const,
+    });
+    const kept = join(baseDir, "kept.wav");
+
+    const result = await processCompositionAudio(
+      [track("a", 0), track("b", 1)],
+      baseDir,
+      workDir,
+      join(baseDir, "out.m4a"),
+      2,
+      undefined,
+      { keepEncoderInputAt: kept },
+    );
+
+    expect(result.success).toBe(true);
+    expect(vi.mocked(encodeAac).mock.calls.at(-1)?.[0]).toBe(join(workDir, "master-sum.wav"));
+    expect(readFileSync(kept, "utf8")).toBe("float mix");
+  });
+
   it("prepares browser root-absolute audio srcs from the project root", async () => {
     const baseDir = mkdtempSync(join(tmpdir(), "hf-audio-base-"));
     const workDir = mkdtempSync(join(tmpdir(), "hf-audio-work-"));

packages/engine/src/services/audioMixer.ts

@@ -1,6 +1,15 @@
 // fallow-ignore-file complexity code-duplication
 import { isSelfOrAncestorHidden, memberGroupKey, isMemberGroupHidden } from "./mediaHidden.js";
-import { closeSync, existsSync, mkdirSync, mkdtempSync, openSync, rmSync, writeFileSync } from "fs";
+import {
+  closeSync,
+  existsSync,
+  mkdirSync,
+  mkdtempSync,
+  openSync,
+  renameSync,
+  rmSync,
+  writeFileSync,
+} from "fs";
 import { join, dirname, isAbsolute, relative } from "path";
 import { parseHTML } from "linkedom";
 import { extractAudioMetadata, type AudioMetadata } from "../utils/ffprobe.js";
@@ -1082,7 +1091,9 @@ async function mixThroughFloatSum(
   workDir: string,
   totalDuration: number,
   signal?: AbortSignal,
-  config?: Partial<Pick<EngineConfig, "ffmpegProcessTimeout" | "audioGain">>,
+  config?: Partial<Pick<EngineConfig, "ffmpegProcessTimeout" | "audioGain">> & {
+    keepEncoderInputAt?: string;
+  },
 ): Promise<MixResult> {
   const sumPath = join(workDir, "master-sum.wav");
   const sum = await mixAudioTracks(tracks, sumPath, totalDuration, signal, config);
@@ -1111,6 +1122,7 @@ async function mixThroughFloatSum(
       failures: [failure],
     };
   }
+  if (config?.keepEncoderInputAt) renameSync(encoderInput, config.keepEncoderInputAt);
   return { ...sum, outputPath, durationMs: sum.durationMs + encoded.durationMs };
 }
 
@@ -1125,6 +1137,7 @@ export async function processCompositionAudio(
     /** Serialised master bus chain, from `parseMasterFxChain`. */
     masterFxChain?: string;
     urlByAbsentLocalPath?: ReadonlyMap<string, string>;
+    keepEncoderInputAt?: string;
   },
   compiledDir?: string,
 ): Promise<MixResult> {

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

@@ -1419,6 +1419,7 @@ export async function buildLocalExecutionPlan(
   // Chunks cut their audio from this track, so it is mixed to the frames' exact length in its one
   // encode; the assembler pads or trims the track of any other plan.
   const audioSeconds = chunksCarryAudio ? totalFrames / config.fps : probeResult.duration;
+  const planMixPcmPath = join(workDir, "plan-mix.wav");
   stageStarted = Date.now();
   const [audioOutcome, overlapOutcome, frameCostOutcome] = await Promise.allSettled([
     runAudioStage({
@@ -1431,6 +1432,7 @@ export async function buildLocalExecutionPlan(
       audioGain: cfg.audioGain,
       audios: composition.audios,
       masterFxChain: composition.masterFxChain,
+      ...(chunksCarryAudio ? { keepEncoderInputAt: planMixPcmPath } : {}),
       abortSignal,
       assertNotAborted,
     }).finally(() => logPlanStage(log, "audio", stageStarted)),
@@ -1497,10 +1499,11 @@ export async function buildLocalExecutionPlan(
     stageStarted = Date.now();
     const peak = await constrainAacTruePeak({
       audioPath: planAudioPath,
+      ...(existsSync(planMixPcmPath) ? { sourcePath: planMixPcmPath } : {}),
       targetDurationSeconds: audioSeconds,
       signal: abortSignal,
       timeoutMs: cfg.ffmpegProcessTimeout,
-    });
+    }).finally(() => rmSync(planMixPcmPath, { force: true }));
     if (!peak.success) throw encoderFailureError("[plan] audio true-peak check failed", peak);
     logPlanStage(log, "audio-peak", stageStarted);
     if (options.sliceChunkAudio) {

packages/producer/src/services/render/audioPadTrim.ts

@@ -373,6 +373,8 @@ export async function padOrTrimAudioToVideoFrameCount(
  */
 export async function constrainAacTruePeak(input: {
   audioPath: string;
+  /** The PCM `audioPath` was encoded from: a correction re-encodes it instead of decoding the AAC. */
+  sourcePath?: string;
   targetDurationSeconds: number;
   signal?: AbortSignal;
   timeoutMs?: number;
@@ -388,7 +390,7 @@ export async function constrainAacTruePeak(input: {
     targetDurationSeconds: input.targetDurationSeconds,
     signal: input.signal,
     probeTruePeak: (audioPath, signal) => measureTruePeakDbfs(audioPath, { timeoutMs, signal }),
-    encode: defaultAacEncoder(timeoutMs, input.signal),
+    encode: defaultAacEncoder(timeoutMs, input.signal, input.sourcePath),
   });
 }
 
@@ -581,12 +583,18 @@ async function defaultRunFfmpeg(
   return toStepResult(await runFfmpeg(args, { signal, timeout: timeoutMs }), timeoutMs);
 }
 
-function defaultAacEncoder(timeoutMs: number, signal?: AbortSignal): AacEncoder {
+function defaultAacEncoder(
+  timeoutMs: number,
+  signal?: AbortSignal,
+  pcmSourcePath?: string,
+): AacEncoder {
   return async (inputPath, outputPath, durationSeconds, filter) => {
-    // One encoder, as before this went through encodeAac: splitting it is a separate change.
-    const options = { signal, timeout: timeoutMs, filter, segments: 1 };
+    // Each segment decodes its input from the start: cheap from PCM, a second lossy pass from AAC.
+    const options = pcmSourcePath
+      ? { signal, timeout: timeoutMs, filter }
+      : { signal, timeout: timeoutMs, filter, segments: 1 };
     return toStepResult(
-      await encodeAac(inputPath, outputPath, durationSeconds, options),
+      await encodeAac(pcmSourcePath ?? inputPath, outputPath, durationSeconds, options),
       timeoutMs,
     );
   };

packages/producer/src/services/render/audioTruePeak.fakeFfmpeg.test.ts

@@ -1,13 +1,19 @@
-import { mkdtempSync, rmSync } from "node:fs";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import { extractAudioMetadata, runFfmpeg, type RunFfmpegResult } from "@hyperframes/engine";
-import { padOrTrimAudioToVideoFrameCount } from "./audioPadTrim.js";
+import {
+  encodeAac,
+  extractAudioMetadata,
+  runFfmpeg,
+  type RunFfmpegResult,
+} from "@hyperframes/engine";
+import { constrainAacTruePeak, padOrTrimAudioToVideoFrameCount } from "./audioPadTrim.js";
 import { measureTruePeakDbfs } from "./audioTruePeak.js";
 
 vi.mock("@hyperframes/engine", async (importOriginal) => ({
   ...(await importOriginal<typeof import("@hyperframes/engine")>()),
+  encodeAac: vi.fn(),
   extractAudioMetadata: vi.fn(),
   runFfmpeg: vi.fn(),
 }));
@@ -36,7 +42,10 @@ const scanFilters = () =>
     .filter(Boolean);
 
 beforeEach(() => probed(180, 48_000));
-afterEach(() => vi.mocked(runFfmpeg).mockReset());
+afterEach(() => {
+  vi.mocked(runFfmpeg).mockReset();
+  vi.mocked(encodeAac).mockReset();
+});
 
 describe("measureTruePeakDbfs against ffmpeg's stderr", () => {
   it("throws when ffmpeg prints no true-peak summary", async () => {
@@ -100,3 +109,50 @@ describe("a local render's audio pad/trim", () => {
     ]);
   });
 });
+
+describe("the true-peak correction", () => {
+  const dir = mkdtempSync(join(tmpdir(), "hf-peak-correct-"));
+  afterAll(() => rmSync(dir, { recursive: true, force: true }));
+
+  async function correctHotAudio(sourcePath?: string) {
+    const scan = (dbfs: string) =>
+      ran(true, `${FRAME.replaceAll("-3.0", dbfs)}\n${SUMMARY.replace("-3.0", dbfs)}`);
+    vi.mocked(runFfmpeg).mockImplementation(async (args, opts) => {
+      const result = scan(args.some((arg) => arg.includes(".true-peak-")) ? "-1.6" : "-0.3");
+      opts?.onStderr?.(result.stderr);
+      return result;
+    });
+    vi.mocked(encodeAac).mockImplementation(async (_input, output) => {
+      writeFileSync(output, "corrected");
+      return ran(true, "");
+    });
+    const audioPath = join(dir, "audio.m4a");
+    writeFileSync(audioPath, "mixed");
+    const result = await constrainAacTruePeak({
+      audioPath,
+      ...(sourcePath ? { sourcePath } : {}),
+      targetDurationSeconds: 180,
+      timeoutMs: 1000,
+    });
+    expect(result).toEqual({ success: true, loweredDb: 1.2 });
+    expect(vi.mocked(encodeAac)).toHaveBeenCalledTimes(1);
+    return { audioPath, call: vi.mocked(encodeAac).mock.calls[0]! };
+  }
+
+  it("re-encodes the plan's PCM mix, not the AAC, in the encoder's parallel segments", async () => {
+    const sourcePath = join(dir, "mix.wav");
+    const { call } = await correctHotAudio(sourcePath);
+    expect(call[0]).toBe(sourcePath);
+    expect(call[3]).toEqual({
+      signal: undefined,
+      timeout: 1000,
+      filter: expect.stringMatching(/^volume=-1\.200dB,/),
+    });
+  });
+
+  it("re-encodes an AAC without its PCM in one encoder, not decoded again per segment", async () => {
+    const { audioPath, call } = await correctHotAudio();
+    expect(call[0]).toBe(audioPath);
+    expect(call[3]).toMatchObject({ segments: 1 });
+  });
+});

packages/producer/src/services/render/stages/audioStage.ts

@@ -43,6 +43,8 @@ export interface AudioStageInput {
   audios: CompositionMetadata["audios"];
   masterFxChain?: string;
   urlByAbsentLocalPath?: ReadonlyMap<string, string>;
+  /** Keep the float mix the AAC was encoded from here, for a later re-encode from it. */
+  keepEncoderInputAt?: string;
   abortSignal: AbortSignal | undefined;
   assertNotAborted: () => void;
   /** Where a per-track failure's detail goes. Optional so tests need not pass one. */
@@ -84,6 +86,7 @@ export async function runAudioStage(input: AudioStageInput): Promise<AudioStageR
     audios,
     masterFxChain,
     urlByAbsentLocalPath,
+    keepEncoderInputAt,
     abortSignal,
     assertNotAborted,
     log,
@@ -111,7 +114,13 @@ export async function runAudioStage(input: AudioStageInput): Promise<AudioStageR
         audioOutputPath,
         duration,
         abortSignal,
-        { ffmpegProcessTimeout, audioGain, masterFxChain, urlByAbsentLocalPath },
+        {
+          ffmpegProcessTimeout,
+          audioGain,
+          masterFxChain,
+          urlByAbsentLocalPath,
+          keepEncoderInputAt,
+        },
         compiledDir,
       );
     } catch (err) {