From cffc3aa4e1f888b4867e4b69f602d1c12dc58e32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miguel=20=C3=81ngel?= Date: Wed, 7 Oct 2026 04:29:35 +0000 Subject: [PATCH] feat(render): report encoding and assembling progress (#5137) * feat(render): report encoding and assembling progress * fix(render): close every progress stage and fail a render whose progress callback throws * fix(render): send a repeated progress update once * fix(render): keep Docker terminal output as before and drop the unneeded callback ending * fix(render): keep one failure stage code per stage whatever its frame counts --- packages/cli/src/commands/render.test.ts | 10 +++ packages/cli/src/commands/render.ts | 20 +++-- packages/cli/src/ui/progress.test.ts | 34 +++++++- packages/cli/src/ui/progress.ts | 6 ++ packages/cli/src/utils/dockerRunArgs.test.ts | 8 ++ packages/cli/src/utils/dockerRunArgs.ts | 2 + packages/engine/src/services/chunkEncoder.ts | 42 +++++++-- .../src/services/streamingEncoder.test.ts | 21 +++++ .../engine/src/services/streamingEncoder.ts | 22 ++++- packages/engine/src/utils/runFfmpeg.test.ts | 34 +++++++- packages/engine/src/utils/runFfmpeg.ts | 27 ++++++ .../src/services/render/shared.test.ts | 69 ++++++++++++++- .../producer/src/services/render/shared.ts | 85 +++++++++++++++++-- .../render/stages/assembleStage.test.ts | 31 +++++++ .../services/render/stages/assembleStage.ts | 30 ++++++- .../render/stages/captureHdrHybridLoop.ts | 3 +- .../render/stages/captureHdrSequentialLoop.ts | 3 +- .../services/render/stages/captureHdrStage.ts | 8 +- .../render/stages/captureSegmentedStage.ts | 3 +- .../services/render/stages/captureStage.ts | 6 +- .../stages/captureStreamingStage.test.ts | 3 +- .../render/stages/captureStreamingStage.ts | 20 +++-- .../render/stages/encodeStage.test.ts | 52 ++++++++++++ .../src/services/render/stages/encodeStage.ts | 26 +++++- .../src/services/renderOrchestrator.ts | 41 +++++++-- 25 files changed, 553 insertions(+), 53 deletions(-) diff --git a/packages/cli/src/commands/render.test.ts b/packages/cli/src/commands/render.test.ts index c4e82c217c..ff2ca137b2 100644 --- a/packages/cli/src/commands/render.test.ts +++ b/packages/cli/src/commands/render.test.ts @@ -2047,6 +2047,16 @@ describe("normalizeStageCode", () => { expect(normalizeStageCode("Starting browsers (5/6 ready)")).toBe("starting_browsers"); }); + it("keeps one code per stage whatever its live frame counts", () => { + expect(normalizeStageCode("Encoding frame 600/600")).toBe("encoding_video"); + expect(normalizeStageCode("Encoding frame 12/90")).toBe("encoding_video"); + expect(normalizeStageCode("Capturing frame 120/600 (6 workers)")).toBe("capturing_frame"); + expect(normalizeStageCode("Streaming frame 3/40 (segment 1/2, 2 workers)")).toBe( + "streaming_frame", + ); + expect(normalizeStageCode("Assembling final video")).toBe("assembling_final_video"); + }); + it("slugifies an unrecognized stage string instead of bucketing it as unknown", () => { expect(normalizeStageCode("Some New Stage!")).toBe("some_new_stage"); }); diff --git a/packages/cli/src/commands/render.ts b/packages/cli/src/commands/render.ts index 46dab62817..904df5a818 100644 --- a/packages/cli/src/commands/render.ts +++ b/packages/cli/src/commands/render.ts @@ -72,7 +72,7 @@ import { errorBox, } from "../ui/format.js"; import { warnIfWebmAlphaDropped } from "../utils/webmAlphaCheck.js"; -import { renderProgress } from "../ui/progress.js"; +import { renderProgress, renderMachineProgress } from "../ui/progress.js"; import { trackRenderComplete, trackRenderError, @@ -801,6 +801,7 @@ async function renderDocker( outputDir: resolve(outputDir), outputFilename, platform, + hostStdoutIsTty: process.stdout.isTTY === true, options: { fps: options.fps, quality: options.quality, @@ -1095,8 +1096,9 @@ async function executeLocalRender( const onProgress = options.quiet ? undefined - : (progressJob: { progress: number }, message: string) => { + : (progressJob: Pick, message: string) => { renderProgress(progressJob.progress, message); + renderMachineProgress(progressJob.progress, progressJob.stageProgress); }; try { @@ -1654,15 +1656,19 @@ const KNOWN_STAGE_CODES: Readonly> = { "Render complete": "render_complete", "Render cancelled": "render_cancelled", pipeline: "pipeline", + "Starting browsers": "starting_browsers", + // Was "Encoding video" before encode reported frames; keeps the same bucket. + "Encoding frame": "encoding_video", }; +// Live counts in a progress label ("Capturing frame 120/600 (6 workers)") would make a code per render. +const STAGE_COUNTS = /\([^)]*\)|\d+\/\d+/g; + export function normalizeStageCode(stage: string): string { - const known = KNOWN_STAGE_CODES[stage]; + const base = stage.replace(STAGE_COUNTS, "").trim(); + const known = KNOWN_STAGE_CODES[base]; if (known) return known; - // The producer's "Starting browsers (k/n ready)" carries live counts; keep one code for it. - if (stage.startsWith("Starting browsers")) return "starting_browsers"; - const slug = stage - .trim() + const slug = base .toLowerCase() .replace(/[^a-z0-9]+/g, "_") .replace(/^_+|_+$/g, ""); diff --git a/packages/cli/src/ui/progress.test.ts b/packages/cli/src/ui/progress.test.ts index 8500779bf0..220efac25d 100644 --- a/packages/cli/src/ui/progress.test.ts +++ b/packages/cli/src/ui/progress.test.ts @@ -1,6 +1,6 @@ import { afterEach, describe, expect, it } from "vitest"; -import { renderProgress } from "./progress.js"; +import { renderMachineProgress, renderProgress } from "./progress.js"; const originalWrite = process.stdout.write.bind(process.stdout); const originalIsTTY = process.stdout.isTTY; @@ -31,3 +31,35 @@ describe("renderProgress", () => { expect(output).not.toContain("\r"); }); }); + +describe("renderMachineProgress", () => { + const capture = (isTTY: boolean) => { + let output = ""; + Object.defineProperty(process.stdout, "isTTY", { value: isTTY, configurable: true }); + process.stdout.write = ((chunk: string | Uint8Array) => { + output += String(chunk); + return true; + }) as typeof process.stdout.write; + renderMachineProgress(83.4, { code: "encode", done: 812, total: 1800 }); + return output; + }; + + it("writes one @hf-progress line for a parent reading piped stdout", () => { + expect(capture(false)).toBe( + '@hf-progress {"code":"encode","done":812,"total":1800,"pct":83}\n', + ); + }); + + it("stays silent in a terminal", () => { + expect(capture(true)).toBe(""); + }); + + it("stays silent in a container whose host prints to a terminal", () => { + process.env.HYPERFRAMES_STDOUT_IS_TTY = "1"; + try { + expect(capture(false)).toBe(""); + } finally { + delete process.env.HYPERFRAMES_STDOUT_IS_TTY; + } + }); +}); diff --git a/packages/cli/src/ui/progress.ts b/packages/cli/src/ui/progress.ts index a109dfe6d8..9a2791f5c2 100644 --- a/packages/cli/src/ui/progress.ts +++ b/packages/cli/src/ui/progress.ts @@ -1,4 +1,5 @@ import { c } from "./colors.js"; +import type { RenderJob } from "@hyperframes/producer"; const { stdout } = process; @@ -18,3 +19,8 @@ export function renderProgress(percent: number, stage: string, row?: number): vo stdout.write(`\r\x1b[2K${line}`); } } + +export function renderMachineProgress(percent: number, stage: RenderJob["stageProgress"]): void { + if (stdout.isTTY || process.env.HYPERFRAMES_STDOUT_IS_TTY === "1" || !stage) return; + stdout.write(`@hf-progress ${JSON.stringify({ ...stage, pct: Math.round(percent) })}\n`); +} diff --git a/packages/cli/src/utils/dockerRunArgs.test.ts b/packages/cli/src/utils/dockerRunArgs.test.ts index c349763ecd..497bad2ecb 100644 --- a/packages/cli/src/utils/dockerRunArgs.test.ts +++ b/packages/cli/src/utils/dockerRunArgs.test.ts @@ -56,6 +56,14 @@ describe("buildDockerRunArgs", () => { `); }); + it("tells the container its output reaches a terminal only when the host's stdout is one", () => { + const flags = (hostStdoutIsTty?: boolean) => + buildDockerRunArgs({ ...FIXED_INPUT, hostStdoutIsTty, options: BASE }).slice(0, 4); + expect(flags(true)).toEqual(["run", "--rm", "-e", "HYPERFRAMES_STDOUT_IS_TTY=1"]); + expect(flags(false)).toEqual(["run", "--rm", "--platform", "linux/amd64"]); + expect(flags(undefined)).toEqual(["run", "--rm", "--platform", "linux/amd64"]); + }); + it("omits --workers when auto sizing should happen inside the container", () => { const args = buildDockerRunArgs({ ...FIXED_INPUT, options: BASE }); expect(args).not.toContain("--workers"); diff --git a/packages/cli/src/utils/dockerRunArgs.ts b/packages/cli/src/utils/dockerRunArgs.ts index ba5d5d310a..a1770189ae 100644 --- a/packages/cli/src/utils/dockerRunArgs.ts +++ b/packages/cli/src/utils/dockerRunArgs.ts @@ -17,6 +17,7 @@ export interface DockerRunArgsInput { outputDir: string; /** Filename within `outputDir` (joined to /output inside the container). */ outputFilename: string; + hostStdoutIsTty?: boolean; /** * Docker `--platform` value (`linux/amd64` or `linux/arm64`). When omitted, * resolves to the host architecture via `resolveDockerPlatform()`. Pinning @@ -109,6 +110,7 @@ export function buildDockerRunArgs(input: DockerRunArgsInput): string[] { return [ "run", "--rm", + ...(input.hostStdoutIsTty ? ["-e", "HYPERFRAMES_STDOUT_IS_TTY=1"] : []), "--platform", platform, "--shm-size=2g", diff --git a/packages/engine/src/services/chunkEncoder.ts b/packages/engine/src/services/chunkEncoder.ts index 9bef8f6575..28243068a3 100644 --- a/packages/engine/src/services/chunkEncoder.ts +++ b/packages/engine/src/services/chunkEncoder.ts @@ -32,7 +32,12 @@ import { import { type HdrTransfer, getHdrEncoderColorParams } from "../utils/hdr.js"; import { withEvenDimensionPad } from "../utils/evenDimensions.js"; import { SDR_CAPTURE_TO_BT709_FILTER } from "../utils/sdrCaptureColor.js"; -import { formatFfmpegError, isExternalFfmpegInterruption, runFfmpeg } from "../utils/runFfmpeg.js"; +import { + ffmpegStatsReader, + formatFfmpegError, + isExternalFfmpegInterruption, + runFfmpeg, +} from "../utils/runFfmpeg.js"; import { extractAudioMetadata } from "../utils/ffprobe.js"; import { type Fps, fpsToFfmpegArg, fpsToNumber } from "@hyperframes/core"; import type { EncoderOptions, EncodeResult, MuxResult } from "./chunkEncoder.types.js"; @@ -491,6 +496,11 @@ export function buildEncoderArgs( return args; } +const framesReader = (onFrames?: (frames: number) => void) => + onFrames && ffmpegStatsReader(({ frames }) => frames !== undefined && onFrames(frames)); +const secondsReader = (onSeconds?: (seconds: number) => void) => + onSeconds && ffmpegStatsReader(({ seconds }) => seconds !== undefined && onSeconds(seconds)); + export async function encodeFramesFromDir( framesDir: string, framePattern: string, @@ -498,6 +508,7 @@ export async function encodeFramesFromDir( options: EncoderOptions, signal?: AbortSignal, config?: Partial>, + onFramesEncoded?: (frames: number) => void, ): Promise { const startTime = Date.now(); @@ -527,7 +538,11 @@ export async function encodeFramesFromDir( const inputArgs = ["-framerate", fpsToFfmpegArg(options.fps), "-i", inputPath]; const args = buildEncoderArgs(options, inputArgs, outputPath, gpuEncoder); const encodeTimeout = config?.ffmpegEncodeTimeout ?? DEFAULT_CONFIG.ffmpegEncodeTimeout; - const result = await runFfmpeg(args, { signal, timeout: encodeTimeout }); + const result = await runFfmpeg(args, { + signal, + timeout: encodeTimeout, + onStderr: framesReader(onFramesEncoded), + }); if (result.terminationReason === "abort") { return { success: false, @@ -636,6 +651,7 @@ export async function encodeFramesChunkedConcat( chunkSizeFrames: number, signal?: AbortSignal, config?: Partial>, + onFramesEncoded?: (frames: number) => void, ): Promise { const start = Date.now(); const files = readdirSync(framesDir) @@ -693,7 +709,13 @@ export async function encodeFramesChunkedConcat( if (options.useGpu) gpuEncoder = await getCachedGpuEncoder(); const args = buildEncoderArgs(options, inputArgs, chunkPath, gpuEncoder); const encodeTimeout = config?.ffmpegEncodeTimeout ?? DEFAULT_CONFIG.ffmpegEncodeTimeout; - const processResult = await runFfmpeg(args, { signal, timeout: encodeTimeout }); + const processResult = await runFfmpeg(args, { + signal, + timeout: encodeTimeout, + onStderr: framesReader( + onFramesEncoded && ((frames) => onFramesEncoded(startNumber + frames)), + ), + }); const chunkResult = { success: processResult.success, error: processResult.success @@ -750,6 +772,7 @@ export async function muxVideoWithAudio( signal?: AbortSignal, config?: MuxVideoWithAudioOptions, fps?: Fps, + onSecondsWritten?: (seconds: number) => void, ): Promise { const outputDir = dirname(outputPath); if (!existsSync(outputDir)) mkdirSync(outputDir, { recursive: true }); @@ -804,7 +827,11 @@ export async function muxVideoWithAudio( args.push("-y", outputPath); const processTimeout = config?.ffmpegProcessTimeout ?? DEFAULT_CONFIG.ffmpegProcessTimeout; - const result = await runFfmpeg(args, { signal, timeout: processTimeout }); + const result = await runFfmpeg(args, { + signal, + timeout: processTimeout, + onStderr: secondsReader(onSecondsWritten), + }); if (signal?.aborted) { return { @@ -977,6 +1004,7 @@ export async function applyFaststart( signal?: AbortSignal, config?: Partial>, fps?: Fps, + onSecondsWritten?: (seconds: number) => void, ): Promise { // faststart is MP4-only (moves moov atom to file start for streaming). // WebM and MOV don't need it — skip the re-mux. @@ -995,7 +1023,11 @@ export async function applyFaststart( args.push("-y", outputPath); const processTimeout = config?.ffmpegProcessTimeout ?? DEFAULT_CONFIG.ffmpegProcessTimeout; - const result = await runFfmpeg(args, { signal, timeout: processTimeout }); + const result = await runFfmpeg(args, { + signal, + timeout: processTimeout, + onStderr: secondsReader(onSecondsWritten), + }); if (signal?.aborted) { return { diff --git a/packages/engine/src/services/streamingEncoder.test.ts b/packages/engine/src/services/streamingEncoder.test.ts index af54d2b52a..abfd969982 100644 --- a/packages/engine/src/services/streamingEncoder.test.ts +++ b/packages/engine/src/services/streamingEncoder.test.ts @@ -604,6 +604,27 @@ describe("spawnStreamingEncoder lifecycle and cleanup", () => { expect(result.fileSize).toBe(0); // No real ffmpeg, no file written }); + it("reports ffmpeg's encoded-frame count while close() waits, and stops at exit", async () => { + const { spawn, calls } = createSpawnSpy(); + vi.resetModules(); + vi.doMock("child_process", () => ({ spawn })); + + const { spawnStreamingEncoder } = await import("./streamingEncoder.js"); + const dir = mkdtempSync(join(tmpdir(), "se-encoded-")); + const encoder = await spawnStreamingEncoder(join(dir, "out.mp4"), baseOptions); + const proc = calls[0]!.proc; + proc.stderr.emit("data", Buffer.from("frame= 10 fps=20 q=28.0 size=1kB time=00:00:00.33\r")); + + const heard: number[] = []; + const closePromise = encoder.close((frames) => heard.push(frames)); + proc.stderr.emit("data", Buffer.from("frame= 25 fps=20 q=28.0 size=2kB time=00:00:00.83\r")); + process.nextTick(() => proc.emit("close", 0)); + await closePromise; + proc.stderr.emit("data", Buffer.from("frame= 30 fps=20 q=28.0 size=2kB time=00:00:01.00\r")); + + expect(heard).toEqual([10, 25]); + }); + it("returns a failure result (does NOT throw) when ffmpeg exits non-zero before close()", async () => { const { spawn, calls } = createSpawnSpy(); vi.resetModules(); diff --git a/packages/engine/src/services/streamingEncoder.ts b/packages/engine/src/services/streamingEncoder.ts index 97949ab624..f3513785d4 100644 --- a/packages/engine/src/services/streamingEncoder.ts +++ b/packages/engine/src/services/streamingEncoder.ts @@ -31,7 +31,11 @@ import { buildVideoToolboxRateControlArgs, mapPresetForGpuEncoder, } from "../utils/gpuEncoder.js"; -import { formatFfmpegError, isExternalFfmpegInterruption } from "../utils/runFfmpeg.js"; +import { + ffmpegStatsReader, + formatFfmpegError, + isExternalFfmpegInterruption, +} from "../utils/runFfmpeg.js"; import { getFfmpegBinary } from "../utils/ffmpegBinaries.js"; import { getHdrEncoderColorParams } from "../utils/hdr.js"; import { withEvenDimensionPad } from "../utils/evenDimensions.js"; @@ -187,7 +191,8 @@ export interface StreamingEncoder { * calls would interleave frame bytes on the pipe and race the drain wait. */ writeFrame: (buffer: Buffer) => Promise; - close: () => Promise; + /** `onFramesEncoded` hears ffmpeg's encoded-frame count while close waits for it to finish. */ + close: (onFramesEncoded?: (frames: number) => void) => Promise; getExitStatus: () => "running" | "success" | "error"; /** * The FFmpeg failure reason (exit code + tail of stderr), or `undefined` @@ -497,9 +502,16 @@ export async function spawnStreamingEncoder( // libx264 printed its summary and exited 255, observable as // "Streaming encode failed: FFmpeg exited with code 255" with audio:0kB). const streamingTimeout = config?.ffmpegStreamingTimeout ?? DEFAULT_CONFIG.ffmpegStreamingTimeout; + let framesEncoded = 0; + let onFramesEncoded: ((frames: number) => void) | undefined; const managed = new ManagedChildProcess(ffmpeg, { signal, inactivityTimeoutMs: streamingTimeout, + onStderr: ffmpegStatsReader(({ frames }) => { + if (frames === undefined) return; + framesEncoded = frames; + onFramesEncoded?.(frames); + }), }); const exitPromise = managed.wait().then((outcome) => { exitCode = outcome.exitCode; @@ -604,7 +616,7 @@ export async function spawnStreamingEncoder( return true; }, - close: async (): Promise => { + close: async (onEncoded?: (frames: number) => void): Promise => { // INVARIANT: close() is idempotent. The renderOrchestrator HDR cleanup // path tracks an `encoderClosed` flag and may still re-call close() in // the outer finally if the inner cleanup raised before the flag flipped. @@ -616,6 +628,10 @@ export async function spawnStreamingEncoder( // repeated calls. If you change this method, preserve idempotency or // a regression here will silently double-close ffmpeg and produce // harder-to-trace errors at the orchestrator layer. + if (onEncoded && exitStatus === "running") { + onFramesEncoded = onEncoded; + onEncoded(framesEncoded); + } const stdin = ffmpeg.stdin; if (stdin && !stdin.destroyed) { await new Promise((resolve) => { diff --git a/packages/engine/src/utils/runFfmpeg.test.ts b/packages/engine/src/utils/runFfmpeg.test.ts index adcffb41b3..2bd153fad8 100644 --- a/packages/engine/src/utils/runFfmpeg.test.ts +++ b/packages/engine/src/utils/runFfmpeg.test.ts @@ -5,7 +5,12 @@ import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; -import { formatFfmpegError, isExternalFfmpegInterruption, runFfmpegPipeline } from "./runFfmpeg.js"; +import { + formatFfmpegError, + isExternalFfmpegInterruption, + runFfmpegPipeline, + ffmpegStatsReader, +} from "./runFfmpeg.js"; const HAS_FFMPEG = spawnSync("ffmpeg", ["-version"]).status === 0; @@ -288,3 +293,30 @@ describe("runFfmpegPipeline start failure", () => { expect(Date.now() - startedAt).toBeLessThan(2_000); }); }); + +describe("ffmpegStatsReader", () => { + it("reads frames and output seconds from stats lines split anywhere across chunks", () => { + const stats: unknown[] = []; + const read = ffmpegStatsReader((s) => stats.push(s)); + read("Input #0, image2pipe\nframe= 12 fps=24 q=2"); + read("8.0 size= 512kB time=00:00:00.40 bitrate=N/A speed=0.8x\rframe= "); + read("30 fps=25 q=28.0 size=1024kB time=00:01:01.50 bitrate=1Mbits/s\r"); + read("[out#0/mp4] video:1024kB\n"); + expect(stats).toEqual([ + { frames: 12, seconds: 0.4 }, + { frames: 30, seconds: 61.5 }, + ]); + }); + + it("reports a copy pass's seconds when it prints no frame count", () => { + const stats: unknown[] = []; + ffmpegStatsReader((s) => stats.push(s))("size= 9000kB time=00:00:12.00 bitrate=6000kbits/s\r"); + expect(stats).toEqual([{ frames: undefined, seconds: 12 }]); + }); + + it("keeps ffmpeg's centiseconds exact", () => { + const stats: unknown[] = []; + ffmpegStatsReader((s) => stats.push(s))("size= 9000kB time=00:01:08.04 bitrate=6000kbits/s\r"); + expect(stats).toEqual([{ frames: undefined, seconds: 68.04 }]); + }); +}); diff --git a/packages/engine/src/utils/runFfmpeg.ts b/packages/engine/src/utils/runFfmpeg.ts index fa2ddbcc48..92cceffefe 100644 --- a/packages/engine/src/utils/runFfmpeg.ts +++ b/packages/engine/src/utils/runFfmpeg.ts @@ -50,6 +50,33 @@ export function isExternalFfmpegInterruption( return result.exitCode === 255 && FFMPEG_SIGTERM_EXIT_LINE.test(result.stderr); } +/** ffmpeg's stats line as it writes: frames encoded so far and seconds of output written. */ +export interface FfmpegStats { + frames?: number; + seconds?: number; +} + +/** Reads stderr chunks, which split stats lines anywhere, and reports each complete stats line. */ +export function ffmpegStatsReader(onStats: (stats: FfmpegStats) => void): (chunk: string) => void { + let rest = ""; + return (chunk) => { + const lines = (rest + chunk).split(/[\r\n]/); + rest = lines.pop() ?? ""; + for (const line of lines) { + const frame = /frame=\s*(\d+)/.exec(line)?.[1]; + const time = /time=\s*(\d+):(\d+):(\d+(?:\.\d+)?)/.exec(line); + if (frame === undefined && !time) continue; + onStats({ + frames: frame === undefined ? undefined : Number(frame), + seconds: time + ? Math.round((Number(time[1]) * 3600 + Number(time[2]) * 60 + Number(time[3])) * 100) / + 100 + : undefined, + }); + } + }; +} + const DEFAULT_TIMEOUT = 300_000; const DEFAULT_STDERR_TAIL_LINES = 15; diff --git a/packages/producer/src/services/render/shared.test.ts b/packages/producer/src/services/render/shared.test.ts index 95df6f98e7..56c5d167b6 100644 --- a/packages/producer/src/services/render/shared.test.ts +++ b/packages/producer/src/services/render/shared.test.ts @@ -1,7 +1,13 @@ import { describe, expect, it, setSystemTime } from "bun:test"; import type { ParallelProgress } from "@hyperframes/engine"; import type { RenderJob } from "../renderOrchestrator.js"; -import { reportWorkerStartup, resolveBrowserMediaEnd } from "./shared.js"; +import { + reportAssembleProgress, + reportEncodeProgress, + reportFrameProgress, + reportWorkerStartup, + resolveBrowserMediaEnd, +} from "./shared.js"; describe("resolveBrowserMediaEnd", () => { it("prefers a runtime duration over a stale compiler-clamped end", () => { @@ -53,3 +59,64 @@ describe("reportWorkerStartup", () => { ]); }); }); + +describe("stage progress for machines", () => { + const job = () => ({ progress: 0, status: "rendering" }) as unknown as RenderJob; + const heard = (j: RenderJob) => { + const seen: unknown[] = []; + return { + seen, + onProgress: (_job: RenderJob, stage: string) => + void seen.push({ stage, progress: j.progress, ...j.stageProgress }), + }; + }; + + it("names capture frames done out of total", () => { + const j = job(); + const { seen, onProgress } = heard(j); + reportFrameProgress(j, "Capturing frame 3/3", 70, onProgress, 3, 3); + expect(seen).toEqual([ + { stage: "Capturing frame 3/3", progress: 70, code: "capture", done: 3, total: 3 }, + ]); + }); + + it("moves encode from where capture left the bar to 90, never back", () => { + const j = job(); + j.progress = 80; + const { seen, onProgress } = heard(j); + reportEncodeProgress(j, 0, 600, onProgress, 80); + reportEncodeProgress(j, 600, 600, onProgress, 80); + expect(seen).toEqual([ + { stage: "Encoding frame 0/600", progress: 80, code: "encode", done: 0, total: 600 }, + { stage: "Encoding frame 600/600", progress: 90, code: "encode", done: 600, total: 600 }, + ]); + }); + + it("sends a stage's closing update once when ffmpeg already reported the last frame", () => { + const j = job(); + const { seen, onProgress } = heard(j); + reportEncodeProgress(j, 600, 600, onProgress, 75); + reportEncodeProgress(j, 600, 600, onProgress, 75); + expect(seen).toHaveLength(1); + }); + + it("reports nothing for a stage with no frames or seconds, so the bar never turns NaN", () => { + const j = job(); + j.progress = 80; + const { seen, onProgress } = heard(j); + reportEncodeProgress(j, 0, 0, onProgress, 80); + reportAssembleProgress(j, 0, 0, onProgress); + expect(seen).toEqual([]); + expect(j.progress).toBe(80); + }); + + it("keeps assemble under 100 until the render completes, even past the end", () => { + const j = job(); + j.progress = 90; + const { seen, onProgress } = heard(j); + reportAssembleProgress(j, 25, 20, onProgress); + expect(seen).toEqual([ + { stage: "Assembling final video", progress: 99, code: "assemble", done: 20, total: 20 }, + ]); + }); +}); diff --git a/packages/producer/src/services/render/shared.ts b/packages/producer/src/services/render/shared.ts index e1a52b92d3..7beb08a99f 100644 --- a/packages/producer/src/services/render/shared.ts +++ b/packages/producer/src/services/render/shared.ts @@ -35,7 +35,12 @@ import type { import type { CompiledComposition } from "../htmlCompiler.js"; import { defaultLogger, type ProducerLogger } from "../../logger.js"; import { isPathInside } from "../../utils/paths.js"; -import type { ProgressCallback, RenderJob, RenderStatus } from "../renderOrchestrator.js"; +import type { + ProgressCallback, + RenderJob, + RenderStageProgress, + RenderStatus, +} from "../renderOrchestrator.js"; export interface CompositionMetadata { duration: number; @@ -246,10 +251,12 @@ export function updateJobStatus( stage: string, progress: number, onProgress?: ProgressCallback, + stageProgress?: RenderStageProgress, ): void { job.warnings ??= []; job.status = status; job.currentStage = stage; + job.stageProgress = stageProgress; const boundedProgress = Math.max(0, Math.min(100, Math.round(progress))); job.progress = Math.max(job.progress, boundedProgress); if (status === "failed" || status === "complete" || status === "cancelled") { @@ -269,9 +276,10 @@ export function updateJobStatus( const PROGRESS_REPORT_INTERVAL_MS = 250; const lastFrameReportAt = new WeakMap(); const lastStartupReportAt = new WeakMap(); +const lastReportSent = new WeakMap(); -// The job updates on every call; the callback fires on the first call per job, when forced, -// and at most once per interval in between. +// The job updates on every call; the callback fires on the first call per job, when forced, and at +// most once per interval in between, never twice in a row with the same stage and percent. function reportThrottled( lastReportAt: WeakMap, job: RenderJob, @@ -279,12 +287,20 @@ function reportThrottled( progress: number, onProgress: ProgressCallback | undefined, force: boolean, + stageProgress: RenderStageProgress, + status: RenderStatus = "rendering", ): void { const now = Date.now(); const last = lastReportAt.get(job); - const due = force || last === undefined || now - last >= PROGRESS_REPORT_INTERVAL_MS; - if (due) lastReportAt.set(job, now); - updateJobStatus(job, "rendering", stage, progress, due ? onProgress : undefined); + const sent = `${stage}|${progress}`; + const due = + lastReportSent.get(job) !== sent && + (force || last === undefined || now - last >= PROGRESS_REPORT_INTERVAL_MS); + if (due) { + lastReportAt.set(job, now); + lastReportSent.set(job, sent); + } + updateJobStatus(job, status, stage, progress, due ? onProgress : undefined, stageProgress); } /** Capture-loop progress: the first frame, the last frame, and at most four reports a second between. */ @@ -293,9 +309,61 @@ export function reportFrameProgress( stage: string, progress: number, onProgress: ProgressCallback | undefined, - isLastFrame: boolean, + done: number, + total: number, ): void { - reportThrottled(lastFrameReportAt, job, stage, progress, onProgress, isLastFrame); + reportThrottled(lastFrameReportAt, job, stage, progress, onProgress, done === total, { + code: "capture", + done, + total, + }); +} + +const lastEncodeReportAt = new WeakMap(); + +/** Frames ffmpeg has encoded, from where capture left the bar up to 90%; same cadence as capture. */ +export function reportEncodeProgress( + job: RenderJob, + done: number, + total: number, + onProgress: ProgressCallback | undefined, + from: number, +): void { + if (total <= 0) return; + const frames = Math.min(done, total); + reportThrottled( + lastEncodeReportAt, + job, + `Encoding frame ${frames}/${total}`, + from + ((90 - from) * frames) / total, + onProgress, + frames === total, + { code: "encode", done: frames, total }, + "encoding", + ); +} + +const lastAssembleReportAt = new WeakMap(); + +/** Seconds of the final file written, 90% to 99%: only a complete file reaches 100. */ +export function reportAssembleProgress( + job: RenderJob, + done: number, + total: number, + onProgress: ProgressCallback | undefined, +): void { + if (total <= 0) return; + const seconds = Math.min(done, total); + reportThrottled( + lastAssembleReportAt, + job, + "Assembling final video", + 90 + (9 * seconds) / total, + onProgress, + seconds >= total, + { code: "assemble", done: seconds, total }, + "assembling", + ); } const workerPhasesByJob = new WeakMap>(); @@ -322,6 +390,7 @@ export function reportWorkerStartup( job.progress, onProgress, false, + { code: "start_browsers", done: ready, total: progress.activeWorkers }, ); } diff --git a/packages/producer/src/services/render/stages/assembleStage.test.ts b/packages/producer/src/services/render/stages/assembleStage.test.ts index afd7f15329..da1ef78c77 100644 --- a/packages/producer/src/services/render/stages/assembleStage.test.ts +++ b/packages/producer/src/services/render/stages/assembleStage.test.ts @@ -20,7 +20,9 @@ vi.mock("../audioPadTrim.js", () => ({ padOrTrimAudioToVideoFrameCount: padOrTrimAudioMock, })); +const { reportAssembleProgressMock } = vi.hoisted(() => ({ reportAssembleProgressMock: vi.fn() })); vi.mock("../shared.js", () => ({ + reportAssembleProgress: reportAssembleProgressMock, updateJobStatus: vi.fn(), })); @@ -84,6 +86,7 @@ describe("runAssembleStage audio duration parity", () => { undefined, { audioCodec: "aac" }, { num: 30, den: 1 }, + expect.any(Function), ); }); @@ -156,6 +159,34 @@ describe("runAssembleStage audio duration parity", () => { }); }); +describe("runAssembleStage progress", () => { + beforeEach(() => { + resetMocks(); + reportAssembleProgressMock.mockReset(); + }); + + it.each([ + ["the audio mux", true], + ["MP4 faststart", false], + ])("reports the seconds %s has written against the video's length", async (_pass, hasAudio) => { + const write = async (...args: unknown[]) => { + (args.at(-1) as (seconds: number) => void)(0.5); + return { success: true }; + }; + muxVideoWithAudioMock.mockImplementation(write); + applyFaststartMock.mockImplementation(write); + const input = makeInput({ hasAudio }); + await runAssembleStage(input); + expect(reportAssembleProgressMock).toHaveBeenCalledWith(input.job, 0.5, 1, undefined); + }); + + it("closes at the video's full length when no pass reports seconds", async () => { + const input = makeInput({ hasAudio: false }); + await runAssembleStage(input); + expect(reportAssembleProgressMock.mock.calls).toEqual([[input.job, 1, 1, undefined]]); + }); +}); + describe("runAssembleStage HLS packaging", () => { beforeEach(resetMocks); diff --git a/packages/producer/src/services/render/stages/assembleStage.ts b/packages/producer/src/services/render/stages/assembleStage.ts index adabe3b675..46b560f677 100644 --- a/packages/producer/src/services/render/stages/assembleStage.ts +++ b/packages/producer/src/services/render/stages/assembleStage.ts @@ -27,7 +27,7 @@ import { AAC_DELIVERY_TRUE_PEAK_DBFS, padOrTrimAudioToVideoFrameCount } from ".. import { encoderFailureError } from "../encoderInterruption.js"; import { DEFAULT_HLS_SEGMENT_SECONDS } from "../hlsConfig.js"; import type { RenderOutputFormat } from "../renderFormat.js"; -import { updateJobStatus } from "../shared.js"; +import { reportAssembleProgress, updateJobStatus } from "../shared.js"; import { defaultLogger } from "../../../logger.js"; export interface AssembleStageInput { @@ -66,7 +66,31 @@ function recordLimiterAttenuation(job: RenderJob, audioLoweredDb: number | undef ); } +function startAssembleProgress( + job: AssembleStageInput["job"], + seconds: number, + onProgress: AssembleStageInput["onProgress"], +): ((secondsWritten: number) => void) | undefined { + updateJobStatus(job, "assembling", "Assembling final video", 90, onProgress, { + code: "assemble", + ...(seconds > 0 && { done: 0, total: seconds }), + }); + if (seconds <= 0) return undefined; + return (done) => reportAssembleProgress(job, done, seconds, onProgress); +} + export async function runAssembleStage(input: AssembleStageInput): Promise { + const { job, onProgress } = input; + const seconds = job.duration ?? 0; + const result = await assembleOutput(input, startAssembleProgress(job, seconds, onProgress)); + reportAssembleProgress(job, seconds, seconds, onProgress); + return result; +} + +async function assembleOutput( + input: AssembleStageInput, + onSecondsWritten: ((secondsWritten: number) => void) | undefined, +): Promise { const { job, videoOnlyPath, @@ -76,12 +100,10 @@ export async function runAssembleStage(input: AssembleStageInput): Promise + reportEncodeProgress(job, frames, totalFrames, onProgress, encodeFrom), + ); hdrEncoderClosed = true; assertNotAborted(); if (!hdrEncodeResult.success) { throw encoderFailureError("HDR encode failed", hdrEncodeResult); } + reportEncodeProgress(job, totalFrames, totalFrames, onProgress, encodeFrom); captureDurationMs = Date.now() - stageStart; encodeMs = hdrEncodeResult.durationMs; } catch (error) { diff --git a/packages/producer/src/services/render/stages/captureSegmentedStage.ts b/packages/producer/src/services/render/stages/captureSegmentedStage.ts index 1f819e4b1e..1b60ba4c7d 100644 --- a/packages/producer/src/services/render/stages/captureSegmentedStage.ts +++ b/packages/producer/src/services/render/stages/captureSegmentedStage.ts @@ -221,7 +221,8 @@ async function captureSegmentFrames( ")", Math.round(25 + ((i + 1) / ctx.totalFrames) * 55), ctx.onProgress, - i + 1 === ctx.totalFrames, + i + 1, + ctx.totalFrames, ); } } diff --git a/packages/producer/src/services/render/stages/captureStage.ts b/packages/producer/src/services/render/stages/captureStage.ts index ae436f8b35..7b43ff53f8 100644 --- a/packages/producer/src/services/render/stages/captureStage.ts +++ b/packages/producer/src/services/render/stages/captureStage.ts @@ -376,7 +376,8 @@ export async function runCaptureStage(input: CaptureStageInput): Promise { "Streaming frame 1/40 (2 workers)", "Streaming frame 2/40 (2 workers)", ]); - expect(stages).toHaveLength(41); + expect(stages).toHaveLength(42); + expect(stages.at(-1)).toBe("Encoding frame 40/40"); }); it("does not flood the callback when frames arrive faster than the report interval", async () => { diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.ts b/packages/producer/src/services/render/stages/captureStreamingStage.ts index fd07fcc450..99d99f706a 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.ts @@ -75,7 +75,7 @@ import type { ProgressCallback, RenderJob } from "../../renderOrchestrator.js"; import { wrapCaptureStageError } from "../captureStageError.js"; import { pushWorkerDedupPerfs } from "../perfSummary.js"; import { ensureFrameWritten } from "./captureHdrFrameShared.js"; -import { reportFrameProgress, reportWorkerStartup } from "../shared.js"; +import { reportEncodeProgress, reportFrameProgress, reportWorkerStartup } from "../shared.js"; import { encoderFailureError } from "../encoderInterruption.js"; import type { SdrStreamingCapturePlan } from "../capturePlan.js"; @@ -494,7 +494,8 @@ async function runWorkerEncodePipelineLoop( `Streaming frame ${prev.idx + 1}/${totalFrames}`, Math.round(25 + ((prev.idx + 1) / totalFrames) * 55), onProgress, - prev.idx + 1 === totalFrames, + prev.idx + 1, + totalFrames, ); }; @@ -521,7 +522,8 @@ async function runWorkerEncodePipelineLoop( `Streaming frame ${item.idx + 1}/${totalFrames}`, Math.round(25 + ((item.idx + 1) / totalFrames) * 55), onProgress, - item.idx + 1 === totalFrames, + item.idx + 1, + totalFrames, ); } }; @@ -827,7 +829,8 @@ export async function runCaptureStreamingStage( `Streaming frame ${progress.capturedFrames}/${progress.totalFrames} (${workerCount} workers)`, Math.round(25 + frameProgress * 55), onProgress, - progress.capturedFrames === progress.totalFrames, + progress.capturedFrames, + progress.totalFrames, ); }, onFrameBuffer, @@ -971,7 +974,8 @@ export async function runCaptureStreamingStage( `Streaming frame ${i + 1}/${totalFrames}`, Math.round(progress), onProgress, - i + 1 === totalFrames, + i + 1, + totalFrames, ); } } @@ -992,13 +996,17 @@ export async function runCaptureStreamingStage( } // Close encoder and get result - const encodeResult = await currentEncoder.close(); + const encodeFrom = job.progress; + const encodeResult = await currentEncoder.close((frames) => + reportEncodeProgress(job, frames, totalFrames, onProgress, encodeFrom), + ); streamingEncoderClosed = true; assertNotAborted(); if (!encodeResult.success) { throw encoderFailureError("Streaming encode failed", encodeResult); } + reportEncodeProgress(job, totalFrames, totalFrames, onProgress, encodeFrom); return { success: true, diff --git a/packages/producer/src/services/render/stages/encodeStage.test.ts b/packages/producer/src/services/render/stages/encodeStage.test.ts index 200f364099..816e55985d 100644 --- a/packages/producer/src/services/render/stages/encodeStage.test.ts +++ b/packages/producer/src/services/render/stages/encodeStage.test.ts @@ -208,6 +208,58 @@ describe("frame pattern follows the capture format, not the output's alpha need" }); }); +describe("runEncodeStage progress", () => { + async function stagesSeen(input: EncodeStageInput, totalFrames: number): Promise { + const { runEncodeStage } = await import("./encodeStage.js"); + input.job.totalFrames = totalFrames; + const seen: unknown[] = []; + await runEncodeStage({ + ...input, + onProgress: (job, stage) => + void seen.push({ stage, progress: job.progress, ...job.stageProgress }), + }); + return seen; + } + + it("reports the frames ffmpeg has encoded, for the bar and for machines", async () => { + encodeFramesFromDirMock.mockImplementationOnce(async (...args: unknown[]) => { + (args[6] as (frames: number) => void)(2); + return { success: true, outputPath: "", durationMs: 1, framesEncoded: 4, fileSize: 1 }; + }); + expect(await stagesSeen(makeInput(), 4)).toEqual([ + { stage: "Encoding video", progress: 75, code: "encode", done: 0, total: 4 }, + { stage: "Encoding frame 2/4", progress: 83, code: "encode", done: 2, total: 4 }, + { stage: "Encoding frame 4/4", progress: 90, code: "encode", done: 4, total: 4 }, + ]); + }); + + it("closes a PNG sequence at its last frame", async () => { + const paths = createFramesDir("png"); + const pngs = makeInput({ + framesDir: paths.framesDir, + outputPath: join(paths.root, "frames-out"), + isPngSequence: true, + }); + expect(await stagesSeen(pngs, 4)).toEqual([ + { stage: "Writing PNG sequence", progress: 75, code: "encode", done: 0, total: 4 }, + { stage: "Encoding frame 4/4", progress: 90, code: "encode", done: 4, total: 4 }, + ]); + }); + + it("closes a GIF encode at its last frame, though ffmpeg reports none in between", async () => { + const paths = createFramesDir("jpg"); + const gif = makeInput({ + framesDir: paths.framesDir, + outputPath: join(paths.root, "out.gif"), + isGif: true, + }); + expect(await stagesSeen(gif, 4)).toEqual([ + { stage: "Encoding GIF", progress: 75, code: "encode", done: 0, total: 4 }, + { stage: "Encoding frame 4/4", progress: 90, code: "encode", done: 4, total: 4 }, + ]); + }); +}); + describe("runEncodeStage config plumbing", () => { it("throws a typed retryable error when the GIF encoder is externally interrupted", async () => { const { EncoderInterruptedError } = await import("../encoderInterruption.js"); diff --git a/packages/producer/src/services/render/stages/encodeStage.ts b/packages/producer/src/services/render/stages/encodeStage.ts index ca6a4e2a3d..3c9465f33b 100644 --- a/packages/producer/src/services/render/stages/encodeStage.ts +++ b/packages/producer/src/services/render/stages/encodeStage.ts @@ -63,7 +63,7 @@ import { buildGifPaletteuseArgs, type GifEncodeArgsInput, } from "./gifEncodeArgs.js"; -import { updateJobStatus } from "../shared.js"; +import { reportEncodeProgress, updateJobStatus } from "../shared.js"; import { encoderFailureError } from "../encoderInterruption.js"; import { frameFileExtension } from "@hyperframes/engine"; @@ -241,6 +241,19 @@ async function encodeGifFromDir( } } +function startEncodeProgress( + job: EncodeStageInput["job"], + stage: string, + onProgress: EncodeStageInput["onProgress"], +): (frames?: number) => void { + const total = job.totalFrames ?? 0; + updateJobStatus(job, "encoding", stage, 75, onProgress, { + code: "encode", + ...(total > 0 && { done: 0, total }), + }); + return (frames = total) => reportEncodeProgress(job, frames, total, onProgress, 75); +} + export async function runEncodeStage(input: EncodeStageInput): Promise { const { job, @@ -273,7 +286,7 @@ export async function runEncodeStage(input: EncodeStageInput): Promise name.endsWith(".png")) @@ -296,6 +309,7 @@ export async function runEncodeStage(input: EncodeStageInput): Promise runCompileStage({ @@ -3313,7 +3332,9 @@ async function executeRenderPipeline(input: { }); // ── Stage 2: Video frame extraction ───────────────────────────────── - updateJobStatus(job, "preprocessing", "Extracting video frames", 10, onProgress); + updateJobStatus(job, "preprocessing", "Extracting video frames", 10, onProgress, { + code: "extract_video", + }); const compiledDir = join(workDir, "compiled"); const extractResult = await observeRenderStage( @@ -3412,7 +3433,9 @@ async function executeRenderPipeline(input: { }); // ── Stage 3: Audio processing ─────────────────────────────────────── - updateJobStatus(job, "preprocessing", "Processing audio tracks", 20, onProgress); + updateJobStatus(job, "preprocessing", "Processing audio tracks", 20, onProgress, { + code: "process_audio", + }); const audioResult = await observeRenderStage( observability, @@ -3467,7 +3490,9 @@ async function executeRenderPipeline(input: { // ── Stage 4: Frame capture ────────────────────────────────────────── const stage4Start = Date.now(); - updateJobStatus(job, "rendering", "Starting frame capture", 25, onProgress); + updateJobStatus(job, "rendering", "Starting frame capture", 25, onProgress, { + code: "start_browsers", + }); // Start file server (may already be running from duration discovery). // The page-side compositing stub is injected later (after hasHdrContent @@ -3499,7 +3524,9 @@ async function executeRenderPipeline(input: { const framesDir = join(workDir, "captured-frames"); if (!existsSync(framesDir)) mkdirSync(framesDir, { recursive: true }); - updateJobStatus(job, "rendering", "Checking browser GPU", 25, onProgress); + updateJobStatus(job, "rendering", "Checking browser GPU", 25, onProgress, { + code: "start_browsers", + }); const resolvedBrowserGpuMode = await resolveBrowserGpuMode(cfg.browserGpuMode, { chromePath: resolveHeadlessShellPath(cfg), browserTimeout: cfg.browserTimeout, @@ -3899,7 +3926,9 @@ async function executeRenderPipeline(input: { !deInversionEligible && !deParallelRouterEligible ) { - updateJobStatus(job, "rendering", "Measuring capture speed", 25, onProgress); + updateJobStatus(job, "rendering", "Measuring capture speed", 25, onProgress, { + code: "start_browsers", + }); const outcome = await observeRenderStage( observability, "capture_calibration",