diff --git a/packages/cli/src/telemetry/events.test.ts b/packages/cli/src/telemetry/events.test.ts index d153d42046..9df3498cf0 100644 --- a/packages/cli/src/telemetry/events.test.ts +++ b/packages/cli/src/telemetry/events.test.ts @@ -361,6 +361,22 @@ describe("render telemetry events", () => { }), ); }); + + it("carries capture_parallel_stream on render_error via the shared payload", () => { + trackRenderError({ + fps: 30, + quality: "standard", + docker: false, + errorMessage: "worker crashed", + captureParallelStream: "beginframe", + }); + + expect(trackEvent).toHaveBeenCalledWith( + "render_error", + expect.objectContaining({ capture_parallel_stream: "beginframe" }), + undefined, + ); + }); }); describe("trackRenderFeedback", () => { diff --git a/packages/cli/src/telemetry/events.ts b/packages/cli/src/telemetry/events.ts index 2b8f4b904a..88196e247b 100644 --- a/packages/cli/src/telemetry/events.ts +++ b/packages/cli/src/telemetry/events.ts @@ -53,6 +53,9 @@ export interface RenderObservabilityTelemetryPayload { captureDePreRouterWorkers?: number; captureDeSelfVerifyFallback?: boolean; captureDeFallbackReason?: string; + /** Non-DE parallel-streaming router outcome ("screenshot" | "beginframe" — + * routed; "eligible_off" — would route but the kill switch is off). */ + captureParallelStream?: string; observabilityExtractVideoCount?: number; observabilityExtractedVideoCount?: number; observabilityExtractTotalFrames?: number; @@ -104,6 +107,7 @@ function renderObservabilityEventProperties(props: RenderObservabilityTelemetryP de_pre_router_workers: props.captureDePreRouterWorkers, de_self_verify_fallback: props.captureDeSelfVerifyFallback, de_fallback_reason: props.captureDeFallbackReason, + capture_parallel_stream: props.captureParallelStream, observability_extract_video_count: props.observabilityExtractVideoCount, observability_extracted_video_count: props.observabilityExtractedVideoCount, observability_extract_total_frames: props.observabilityExtractTotalFrames, diff --git a/packages/cli/src/telemetry/renderObservability.test.ts b/packages/cli/src/telemetry/renderObservability.test.ts index de53869458..0a64cc8661 100644 --- a/packages/cli/src/telemetry/renderObservability.test.ts +++ b/packages/cli/src/telemetry/renderObservability.test.ts @@ -82,3 +82,24 @@ describe("renderObservabilityTelemetryPayload — DE inversion/router cohort (fa expect(payload.captureDeFallbackReason).toBeUndefined(); }); }); + +describe("renderObservabilityTelemetryPayload — non-DE parallel-stream router", () => { + it("maps the router outcome", () => { + const payload = renderObservabilityTelemetryPayload( + makeSummary({ captureParallelStream: "beginframe" }), + ); + expect(payload.captureParallelStream).toBe("beginframe"); + }); + + it("maps the passive eligible_off cohort-sizing signal", () => { + const payload = renderObservabilityTelemetryPayload( + makeSummary({ captureParallelStream: "eligible_off" }), + ); + expect(payload.captureParallelStream).toBe("eligible_off"); + }); + + it("stays undefined when the router never fired", () => { + const payload = renderObservabilityTelemetryPayload(makeSummary({})); + expect(payload.captureParallelStream).toBeUndefined(); + }); +}); diff --git a/packages/cli/src/telemetry/renderObservability.ts b/packages/cli/src/telemetry/renderObservability.ts index 45288ce86c..25e17f118f 100644 --- a/packages/cli/src/telemetry/renderObservability.ts +++ b/packages/cli/src/telemetry/renderObservability.ts @@ -46,6 +46,7 @@ export function renderObservabilityTelemetryPayload( captureDePreRouterWorkers: capture.dePreRouterWorkers, captureDeSelfVerifyFallback: capture.deSelfVerifyFallback, captureDeFallbackReason: capture.deFallbackReason, + captureParallelStream: capture.captureParallelStream, observabilityExtractVideoCount: extraction?.videoCount, observabilityExtractedVideoCount: extraction?.extractedVideoCount, observabilityExtractTotalFrames: extraction?.totalFramesExtracted, diff --git a/packages/producer/src/services/render/observability.ts b/packages/producer/src/services/render/observability.ts index f793a47a41..c180f0481b 100644 --- a/packages/producer/src/services/render/observability.ts +++ b/packages/producer/src/services/render/observability.ts @@ -68,6 +68,16 @@ export interface RenderCaptureObservability { deParallelRouter?: "routed" | "reverted"; /** Worker count the resolver would have used absent the router; undefined if it never fired. */ dePreRouterWorkers?: number; + /** + * Non-DE parallel-streaming router outcome (HF_CAPTURE_PARALLEL_STREAM): + * "screenshot" | "beginframe" — the render passed every gate AND the kill + * switch was on, so it was routed through the interleaved streaming encoder + * (the value is the capture mode that streamed); "eligible_off" — the render + * passed every gate EXCEPT the kill switch (passive cohort-sizing signal for + * the default-off soak: how many renders WOULD route if enabled). Absent = + * ineligible regardless of the switch. + */ + captureParallelStream?: "screenshot" | "beginframe" | "eligible_off"; protocolTimeoutMs?: number; pageNavigationTimeoutMs?: number; playerReadyTimeoutMs?: number; diff --git a/packages/producer/src/services/renderOrchestrator.test.ts b/packages/producer/src/services/renderOrchestrator.test.ts index f3eb95841c..2472932db9 100644 --- a/packages/producer/src/services/renderOrchestrator.test.ts +++ b/packages/producer/src/services/renderOrchestrator.test.ts @@ -34,6 +34,7 @@ import { shouldRetryViaPinnedFallback, shouldPreferParallelDrawElement, shouldPreferSingleWorkerDrawElement, + shouldStreamParallelCapture, shouldUseStreamingEncode, } from "./renderOrchestrator.js"; import { ensureFrameWritten } from "./render/stages/captureHdrFrameShared.js"; @@ -2049,3 +2050,43 @@ describe("shouldRetryViaPinnedFallback (widen the self-verify retry to generic c ).toBe(false); }); }); + +describe("shouldStreamParallelCapture (non-DE parallel streaming router)", () => { + const eligible = { + routerEnabled: true, + workerCount: 3, + useDrawElement: false, + outputFormat: "mp4" as const, + streamingOk: true, + layeredOrEffectRoute: false, + }; + + it("routes an eligible multi-worker non-drawElement render", () => { + expect(shouldStreamParallelCapture(eligible)).toBe(true); + }); + + it("is disabled by default (kill switch off is the shipped default)", () => { + expect(shouldStreamParallelCapture({ ...eligible, routerEnabled: false })).toBe(false); + }); + + it("never fires for single-worker renders (those already stream)", () => { + expect(shouldStreamParallelCapture({ ...eligible, workerCount: 1 })).toBe(false); + }); + + it("never fires when drawElement will capture (the DE routers own that path)", () => { + expect(shouldStreamParallelCapture({ ...eligible, useDrawElement: true })).toBe(false); + }); + + it("only applies to mp4", () => { + expect(shouldStreamParallelCapture({ ...eligible, outputFormat: "webm" })).toBe(false); + expect(shouldStreamParallelCapture({ ...eligible, outputFormat: "png-sequence" })).toBe(false); + }); + + it("respects the streaming-encode config/duration gates", () => { + expect(shouldStreamParallelCapture({ ...eligible, streamingOk: false })).toBe(false); + }); + + it("skips HDR-layered and shader-transition routes", () => { + expect(shouldStreamParallelCapture({ ...eligible, layeredOrEffectRoute: true })).toBe(false); + }); +}); diff --git a/packages/producer/src/services/renderOrchestrator.ts b/packages/producer/src/services/renderOrchestrator.ts index c915ec5d29..74dcb3d17b 100644 --- a/packages/producer/src/services/renderOrchestrator.ts +++ b/packages/producer/src/services/renderOrchestrator.ts @@ -1299,6 +1299,54 @@ export function shouldRetryViaPinnedFallback(args: { return args.deWorkerInversion === "inverted" || args.deParallelRouter === "routed"; } +/** + * Parallel-streaming router for NON-drawElement capture (screenshot on + * macOS/Windows/forced-screenshot, BeginFrame on Linux): should this + * multi-worker render stream captured frame buffers straight into the single + * ffmpeg stdin encoder (interleaved distribution + ordered reorder-buffer + * writer — the PR #2056 machinery) instead of the parallel disk path (workers + * write JPEGs, a separate sequential encode pass reads them back)? + * + * Measured motivation (2026-07-10, macOS SS W3): the disk path's encode is a + * purely additive tail (~27% of wall clock on a 3,600-frame comp). Streaming + * overlapped it for 1.29x on a uniform-cost comp and was a wash (not a + * regression) on a 39%-static bimodal comp — the interleaved writer's + * near-lockstep coupling eats the encode win when frame costs are bimodal. + * v1 accepts the wash; static-aware routing is a documented follow-up. + * + * Unlike the DE router this deliberately does NOT require auto-resolved + * workers: streaming doesn't change the worker count, so an explicit + * `--workers 3` should benefit too. It requires !useDrawElement + * (post-resolveConfig — always true on Linux): DE parallel renders belong to + * the DE parallel router (HF_DE_PARALLEL_ROUTER) with its self-verify + * machinery; both DE predicates independently require useDrawElement, making + * the two routers mutually exclusive by construction. + */ +export function shouldStreamParallelCapture(args: { + /** HF_CAPTURE_PARALLEL_STREAM === "true" — kill switch, default OFF. */ + routerEnabled: boolean; + workerCount: number; + /** cfg.useDrawElement AFTER resolveConfig clamps. */ + useDrawElement: boolean; + outputFormat: NonNullable; + /** shouldUseStreamingEncode(cfg, format, 1, duration) at the call site — + * carries the enableStreamingEncode/format/duration-cap gates. */ + streamingOk: boolean; + /** HDR layered composite or shader transitions — bespoke pipelines + * (including page-side compositing, which only engages when + * hasShaderTransitions) that never stream. */ + layeredOrEffectRoute: boolean; +}): boolean { + return ( + args.routerEnabled && + args.workerCount > 1 && + !args.useDrawElement && + args.outputFormat === "mp4" && + args.streamingOk && + !args.layeredOrEffectRoute + ); +} + export function resolveCaptureForceScreenshotForPageSideCompositing(args: { forceScreenshot: boolean; usePageSideCompositing: boolean; @@ -1599,6 +1647,10 @@ export async function executeRenderJob( // render already executing in the same process. Threading this as a // local instead closes that cross-talk, not just the sequential leak. let deParallelStreamForced = false; + // Per-render (not process-global) signal that the NON-DE parallel-stream + // router fired — same threading discipline as deParallelStreamForced + // (see that flag's comment for why this must never be an env mutation). + let captureParallelStreamForced = false; let deSelfVerifyFallback = false; let deFallbackReason: string | undefined; let deDrainStats: import("./render/stages/captureStreamingStage.js").DeDrainStats | undefined; @@ -2190,6 +2242,50 @@ export async function executeRenderJob( deParallelRouter: deParallelRouter ?? "none", }); + // Non-DE parallel-streaming router — see shouldStreamParallelCapture. + // Mutually exclusive with the DE inversion/router above by construction + // (both DE predicates require useDrawElement; this requires its negation). + const captureParallelStreamRouterEnabled = process.env.HF_CAPTURE_PARALLEL_STREAM === "true"; + const captureParallelStreamArgs = { + workerCount, + useDrawElement: cfg.useDrawElement, + outputFormat, + streamingOk: shouldUseStreamingEncode(cfg, outputFormat, 1, job.duration), + layeredOrEffectRoute: hasHdrContent || compiled.hasShaderTransitions, + }; + const captureParallelStreamEligible = shouldStreamParallelCapture({ + routerEnabled: captureParallelStreamRouterEnabled, + ...captureParallelStreamArgs, + }); + if (captureParallelStreamEligible) { + captureParallelStreamForced = true; + // Which mode will stream: the engine picks beginframe only on Linux with + // headless-shell and no forced screenshot (frameCapture.ts preMode); + // everything else is screenshot. Recorded for telemetry cohorting. + const captureParallelStream = + process.platform === "linux" && !captureForceScreenshot ? "beginframe" : "screenshot"; + log.info( + `[Render] Parallel ${captureParallelStream} capture will stream to the encoder ` + + `(interleaved, ${workerCount} workers) instead of the disk path. ` + + "Set HF_CAPTURE_PARALLEL_STREAM=false to disable.", + ); + updateCaptureObservability({ captureParallelStream }); + // NOTE: no string data on the checkpoint — RenderObservationData string + // values are dropped unless the key is in observability.ts's + // ALLOWED_STRING_DATA_KEYS allow-list. The message carries the detail. + observability.checkpoint( + "worker_resolution", + `parallel ${captureParallelStream} capture routed to streaming`, + ); + } else if (shouldStreamParallelCapture({ routerEnabled: true, ...captureParallelStreamArgs })) { + // The kill switch is the ONLY failed gate: emit a passive cohort-sizing + // signal (capture_parallel_stream = "eligible_off") so the default-off + // soak can measure how many fleet renders WOULD route before anyone + // enables the flag. Observability-only — no behavior change, no log + // noise on the default path. + updateCaptureObservability({ captureParallelStream: "eligible_off" }); + } + if (workerCount > 1 && probeSession) { lastBrowserConsole = probeSession.browserConsoleBuffer; await closeCaptureSession(probeSession); @@ -2206,7 +2302,7 @@ export async function executeRenderJob( outputFormat, workerCount, job.duration, - deParallelStreamForced, + deParallelStreamForced || captureParallelStreamForced, ); log.info("streaming-encode gate", { enabled: useStreamingEncode, @@ -2450,7 +2546,7 @@ export async function executeRenderJob( workerCount, probeSession, outputFormat, - forceParallelStream: deParallelStreamForced, + forceParallelStream: deParallelStreamForced || captureParallelStreamForced, streamingEncoderOptions: { fps: job.config.fps, width,