perf(producer): the audio task finds sound with a probe and reads URL clips in parallel

packages/engine/src/index.ts

@@ -214,6 +214,7 @@ export {
   isVideoPastTimelineEnd,
   resolveTimelineExtractionWindow,
   resolveVideoExtractionWindow,
+  rebaseVideoToWindow,
   resolveFinalFrameExtractionWindow,
   resolveVideoExtractionDuration,
   resolvePlayableVideoDuration,

packages/engine/src/services/audioMixer.ts

@@ -203,6 +203,8 @@ function legacyFilterScriptOptionIsUnsupported(stderr: string): boolean {
  */
 // Caps the ffmpeg runs that prepare clips; all of a long mix's clips at once starved the host (200 clips, 200 ffmpeg).
 const inPrepareSlot = createConcurrencyLimit(Math.max(2, Math.floor(cpus().length / 2)));
+// A clip read from a URL spends its time waiting on the network, so it needs no CPU-sized limit.
+const inRemotePrepareSlot = createConcurrencyLimit(16);
 
 const MAX_VOLUME_SEGMENTS = 32;
 
@@ -757,6 +759,23 @@ async function extractAudioSegment(
   };
 }
 
+function isRemoteElement(
+  element: { src: string },
+  baseDir: string,
+  compiledDir: string | undefined,
+  config?: { urlByAbsentLocalPath?: ReadonlyMap<string, string> },
+): boolean {
+  if (isHttpUrl(element.src)) return true;
+  if (!config?.urlByAbsentLocalPath?.size) return false;
+  try {
+    return config.urlByAbsentLocalPath.has(
+      resolveProjectRelativeSrc(element.src, baseDir, compiledDir),
+    );
+  } catch {
+    return false;
+  }
+}
+
 async function generateSilence(
   outputPath: string,
   duration: number,
@@ -1357,7 +1376,9 @@ export async function processCompositionAudio(
 
   await Promise.all(
     elements.map((element, index) =>
-      inPrepareSlot(async () => {
+      (isRemoteElement(element, baseDir, compiledDir, config)
+        ? inRemotePrepareSlot
+        : inPrepareSlot)(async () => {
         if (effectiveSignal.aborted) {
           failures.push({
             stage: "cancelled",

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

@@ -3,7 +3,7 @@ import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "nod
 import { createServer, type Server, type ServerResponse } from "node:http";
 import { once } from "node:events";
 import type { AddressInfo } from "node:net";
-import { tmpdir } from "node:os";
+import { cpus, tmpdir } from "node:os";
 import { join } from "node:path";
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
 import { getFfmpegBinary } from "../utils/ffmpegBinaries.js";
@@ -231,4 +231,50 @@ describe.skipIf(!HAS_FFMPEG)("audio cut from a cloud-held file's URL", { timeout
       await new Promise((resolve) => sidecarServer.close(resolve));
     }
   });
+
+  it("reads many clips of one URL at once, since the wait is on the network and not the CPU", async () => {
+    const body = makeVoice();
+    let open = 0;
+    let peak = 0;
+    server = createServer((req, res) => {
+      const start = Number(/bytes=(\d+)-/.exec(req.headers.range ?? "")?.[1] ?? 0);
+      open++;
+      peak = Math.max(peak, open);
+      res.on("close", () => open--);
+      setTimeout(() => {
+        res.writeHead(206, {
+          "Accept-Ranges": "bytes",
+          "Content-Length": body.length - start,
+          "Content-Range": `bytes ${start}-${body.length - 1}/${body.length}`,
+        });
+        res.end(body.subarray(start));
+      }, 300);
+    });
+    await new Promise<void>((resolve) => server!.listen(0, "127.0.0.1", resolve));
+    const { port } = server.address() as AddressInfo;
+    const baseDir = join(dir, "project");
+    const local = join(baseDir, "takes", "voice.wav");
+
+    const result = await processCompositionAudio(
+      Array.from({ length: 24 }, (_, i) => ({
+        id: `clip-${i}`,
+        src: "takes/voice.wav",
+        start: i,
+        end: i + 1,
+        mediaStart: i % 10,
+        layer: i,
+        volume: 1,
+        type: "audio" as const,
+      })),
+      baseDir,
+      join(dir, "work"),
+      join(dir, "out.m4a"),
+      24,
+      abort.signal,
+      { urlByAbsentLocalPath: new Map([[local, `http://127.0.0.1:${port}/voice.wav`]]) },
+    );
+
+    expect(result.success).toBe(true);
+    expect(peak).toBeGreaterThan(Math.max(2, Math.floor(cpus().length / 2)));
+  });
 });

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

@@ -1131,48 +1131,53 @@ describe.skipIf(spawnSync(getFfmpegBinary(), ["-version"]).status !== 0)(
       }
     }, 60_000);
 
-    it("gives buildPlanAudio the sound of a video whose src the scene script sets", async () => {
-      const project = join(runRoot, "project-script-src");
-      mkdirSync(project, { recursive: true });
-      const clip = spawnSync(getFfmpegBinary(), [
-        "-nostdin",
-        "-v",
-        "error",
-        "-f",
-        "lavfi",
-        "-i",
-        "testsrc=s=320x240:r=30:d=1",
-        "-f",
-        "lavfi",
-        "-i",
-        "sine=frequency=440:sample_rate=48000:d=1",
-        "-c:v",
-        "libx264",
-        "-pix_fmt",
-        "yuv420p",
-        "-c:a",
-        "aac",
-        "-shortest",
-        join(project, "clip.mp4"),
-      ]);
-      if (clip.status !== 0) throw new Error(`ffmpeg clip failed: ${clip.stderr}`);
-      writeFileSync(
-        join(project, "index.html"),
-        `<!doctype html><html><head><meta charset="utf-8"></head><body>
+    for (const slot of [
+      { name: "starts on the timeline", start: 0, duration: 1 },
+      { name: "starts before the timeline", start: -0.5, duration: 1.5 },
+    ]) {
+      it(`gives buildPlanAudio the sound of a video whose src the scene script sets and that ${slot.name}`, async () => {
+        const project = join(runRoot, `project-script-src-${slot.start}`);
+        mkdirSync(project, { recursive: true });
+        const clip = spawnSync(getFfmpegBinary(), [
+          "-nostdin",
+          "-v",
+          "error",
+          "-f",
+          "lavfi",
+          "-i",
+          "testsrc=s=320x240:r=30:d=1",
+          "-f",
+          "lavfi",
+          "-i",
+          "sine=frequency=440:sample_rate=48000:d=1",
+          "-c:v",
+          "libx264",
+          "-pix_fmt",
+          "yuv420p",
+          "-c:a",
+          "aac",
+          "-shortest",
+          join(project, "clip.mp4"),
+        ]);
+        if (clip.status !== 0) throw new Error(`ffmpeg clip failed: ${clip.stderr}`);
+        writeFileSync(
+          join(project, "index.html"),
+          `<!doctype html><html><head><meta charset="utf-8"></head><body>
   <div data-composition-id="root" data-width="320" data-height="240" data-duration="1" data-no-timeline>
-    <video id="v" data-has-audio="true" data-start="0" data-duration="1" playsinline></video>
+    <video id="v" data-has-audio="true" data-start="${slot.start}" data-duration="${slot.duration}" playsinline></video>
   </div>
   <script>document.getElementById("v").src = "clip.mp4";</script></body></html>`,
-      );
-      const config = { fps: 30, width: 320, height: 240, format: "mp4" } as const;
-      const carried = join(runRoot, "plan-script-src");
-      await buildLocalExecutionPlan(project, config, carried, { chunksCarryAudio: true });
-      const work = join(runRoot, "script-src-work");
-      mkdirSync(work, { recursive: true });
-      const audio = join(work, "audio.m4a");
-      expect(await buildPlanAudio(project, config, work, audio)).toEqual({ hasAudio: true });
-      expect(readFileSync(audio).equals(readFileSync(join(carried, "audio.m4a")))).toBe(true);
-    }, 60_000);
+        );
+        const config = { fps: 30, width: 320, height: 240, format: "mp4" } as const;
+        const carried = join(runRoot, `plan-script-src-${slot.start}`);
+        await buildLocalExecutionPlan(project, config, carried, { chunksCarryAudio: true });
+        const work = join(runRoot, `script-src-work-${slot.start}`);
+        mkdirSync(work, { recursive: true });
+        const audio = join(work, "audio.m4a");
+        expect(await buildPlanAudio(project, config, work, audio)).toEqual({ hasAudio: true });
+        expect(readFileSync(audio).equals(readFileSync(join(carried, "audio.m4a")))).toBe(true);
+      }, 60_000);
+    }
 
     it("keeps assembled loudness within 0.1 LUFS of the frame-fitted float delivery", async () => {
       const project = audioProject("project-loudness-delivery", 0.3, 3.01);

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

@@ -64,6 +64,7 @@ import { measurePlanSecondsPerFrame } from "./planFrameCost.js";
 import { runAudioStage } from "../render/stages/audioStage.js";
 import { runCompileStage } from "../render/stages/compileStage.js";
 import {
+  appendVideoAudioFromProbe,
   assertVideoExtractionSucceeded,
   runExtractVideosStage,
 } from "../render/stages/extractVideosStage.js";
@@ -1727,7 +1728,7 @@ export async function buildPlanAudio(
   });
   try {
     let stageStarted = Date.now();
-    await extractPlanVideos(c, true);
+    await appendVideoAudioFromProbe(c);
     logPlanStage(c.log, "extract", stageStarted);
     stageStarted = Date.now();
     const mixed = await mixPlanAudio(c, c.compiledDir);

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

@@ -33,7 +33,6 @@ import { join } from "node:path";
 import {
   type CaptureVideoMetadataHint,
   type EngineConfig,
-  type ExtractedFrames,
   type ExtractionResult,
   type FrameLookupTable,
   type HdrTransfer,
@@ -47,6 +46,10 @@ import {
   extractMediaMetadata,
   isVideoPastTimelineEnd,
   isHdrColorSpace,
+  rebaseVideoToWindow,
+  resolveFinalFrameExtractionWindow,
+  resolveVideoExtractionWindow,
+  isHttpUrl,
   resolveProjectRelativeSrc,
   runVideoExtractionWithRetry,
   safeVideoExtractionSourceIdentity,
@@ -568,7 +571,7 @@ export async function runExtractVideosStage(
  */
 export function appendAutoDetectedVideoAudio(
   composition: Pick<CompositionMetadata, "videos" | "audios">,
-  extracted: ExtractedFrames[],
+  extracted: ReadonlyArray<{ videoId: string; metadata: { hasAudio: boolean } }>,
 ): void {
   const existingAudioSrcs = new Set(composition.audios.map((a) => a.src));
   for (const ext of extracted) {
@@ -588,3 +591,42 @@ export function appendAutoDetectedVideoAudio(
     existingAudioSrcs.add(video.src);
   }
 }
+
+/**
+ * The audio-only form of the extract stage: which videos carry sound takes a probe per source, not their frames.
+ * Each video is moved to its extraction window first, as extraction does, so its sound lands where its picture does.
+ */
+export async function appendVideoAudioFromProbe(input: {
+  projectDir: string;
+  compiledDir: string;
+  composition: CompositionMetadata;
+  urlByAbsentLocalPath?: ReadonlyMap<string, string>;
+}): Promise<void> {
+  const { projectDir, compiledDir, composition, urlByAbsentLocalPath } = input;
+  const videos = composition.videos.filter(
+    (video) => !isVideoPastTimelineEnd(video, composition.duration),
+  );
+  const probed = await Promise.all(
+    videos.map(async (video) => {
+      const path = isHttpUrl(video.src)
+        ? video.src
+        : resolveProjectRelativeSrc(video.src, projectDir, compiledDir);
+      const readFrom = urlByAbsentLocalPath?.get(path) ?? path;
+      if (!isHttpUrl(readFrom) && !existsSync(readFrom)) return null;
+      const metadata = await extractMediaMetadata(readFrom);
+      const window = await resolveFinalFrameExtractionWindow(
+        readFrom,
+        video,
+        metadata,
+        resolveVideoExtractionWindow(video, metadata, composition.duration),
+      );
+      if (window.durationSeconds <= 0) return null;
+      rebaseVideoToWindow(video, window);
+      return { videoId: video.id, metadata };
+    }),
+  );
+  appendAutoDetectedVideoAudio(
+    composition,
+    probed.filter((entry) => entry !== null),
+  );
+}

packages/producer/src/services/render/stages/videoAudioProbe.test.ts

@@ -0,0 +1,56 @@
+import { afterEach, beforeEach, describe, expect, it } from "bun:test";
+import { spawnSync } from "node:child_process";
+import { mkdirSync, mkdtempSync, readdirSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { getFfmpegBinary } from "@hyperframes/engine";
+import { collectRenderMedia } from "../../renderMediaCollector.js";
+import { appendVideoAudioFromProbe } from "./extractVideosStage.js";
+
+const HAS_FFMPEG = spawnSync(getFfmpegBinary(), ["-version"]).status === 0;
+
+describe.skipIf(!HAS_FFMPEG)("appendVideoAudioFromProbe", () => {
+  let dir: string;
+  beforeEach(() => {
+    dir = mkdtempSync(join(tmpdir(), "hf-video-audio-probe-"));
+    mkdirSync(join(dir, "compiled"));
+  });
+  afterEach(() => rmSync(dir, { recursive: true, force: true }));
+
+  const make = (name: string, withAudio: boolean) => {
+    const args = [
+      "-v",
+      "error",
+      "-y",
+      "-f",
+      "lavfi",
+      "-i",
+      "testsrc2=size=64x64:rate=25:duration=4",
+    ];
+    if (withAudio)
+      args.push("-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000:duration=4");
+    args.push("-c:v", "mpeg4", ...(withAudio ? ["-c:a", "aac"] : []), join(dir, name));
+    expect(spawnSync(getFfmpegBinary(), args).status).toBe(0);
+  };
+
+  it("adds the sound of a video that has some, once per source, without extracting any frames", async () => {
+    make("talk.mp4", true);
+    make("silent.mp4", false);
+    const composition =
+      collectRenderMedia(`<div data-composition-id="root" data-start="0" data-duration="8">
+      <video id="a" src="talk.mp4" data-start="0" data-duration="2" data-has-audio="true"></video>
+      <video id="b" src="talk.mp4" data-start="2" data-duration="2" data-media-start="1" data-has-audio="true"></video>
+      <video id="c" src="silent.mp4" data-start="4" data-duration="2" data-has-audio="true"></video>
+    </div>`);
+    composition.audios.length = 0;
+
+    await appendVideoAudioFromProbe({
+      projectDir: dir,
+      compiledDir: join(dir, "compiled"),
+      composition,
+    });
+
+    expect(composition.audios.map((audio) => audio.id)).toEqual(["a-audio"]);
+    expect(readdirSync(join(dir, "compiled"))).toEqual([]);
+  });
+});