feat(producer): route eligible parallel screenshot/beginframe renders to streaming

This commit is contained in:
vanceingalls
2026-07-11 22:34:46 +00:00
parent 2416e9bcd5
commit 6d5a7711f3
2 changed files with 46 additions and 2 deletions
@@ -68,6 +68,13 @@ export interface RenderCaptureObservability {
deParallelRouter?: "routed" | "reverted"; deParallelRouter?: "routed" | "reverted";
/** Worker count the resolver would have used absent the router; undefined if it never fired. */ /** Worker count the resolver would have used absent the router; undefined if it never fired. */
dePreRouterWorkers?: number; dePreRouterWorkers?: number;
/**
* Non-DE parallel-streaming router outcome (HF_CAPTURE_PARALLEL_STREAM):
* set when a multi-worker screenshot/BeginFrame render was routed through
* the interleaved streaming encoder instead of the parallel disk path.
* The value is the capture mode that streamed. Absent = not routed.
*/
captureParallelStream?: "screenshot" | "beginframe";
protocolTimeoutMs?: number; protocolTimeoutMs?: number;
pageNavigationTimeoutMs?: number; pageNavigationTimeoutMs?: number;
playerReadyTimeoutMs?: number; playerReadyTimeoutMs?: number;
@@ -1647,6 +1647,10 @@ export async function executeRenderJob(
// render already executing in the same process. Threading this as a // render already executing in the same process. Threading this as a
// local instead closes that cross-talk, not just the sequential leak. // local instead closes that cross-talk, not just the sequential leak.
let deParallelStreamForced = false; 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 deSelfVerifyFallback = false;
let deFallbackReason: string | undefined; let deFallbackReason: string | undefined;
let deDrainStats: import("./render/stages/captureStreamingStage.js").DeDrainStats | undefined; let deDrainStats: import("./render/stages/captureStreamingStage.js").DeDrainStats | undefined;
@@ -2238,6 +2242,39 @@ export async function executeRenderJob(
deParallelRouter: deParallelRouter ?? "none", 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 captureParallelStreamEligible = shouldStreamParallelCapture({
routerEnabled: process.env.HF_CAPTURE_PARALLEL_STREAM === "true",
workerCount,
useDrawElement: cfg.useDrawElement,
outputFormat,
streamingOk: shouldUseStreamingEncode(cfg, outputFormat, 1, job.duration),
layeredOrEffectRoute: hasHdrContent || compiled.hasShaderTransitions,
});
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`,
);
}
if (workerCount > 1 && probeSession) { if (workerCount > 1 && probeSession) {
lastBrowserConsole = probeSession.browserConsoleBuffer; lastBrowserConsole = probeSession.browserConsoleBuffer;
await closeCaptureSession(probeSession); await closeCaptureSession(probeSession);
@@ -2254,7 +2291,7 @@ export async function executeRenderJob(
outputFormat, outputFormat,
workerCount, workerCount,
job.duration, job.duration,
deParallelStreamForced, deParallelStreamForced || captureParallelStreamForced,
); );
log.info("streaming-encode gate", { log.info("streaming-encode gate", {
enabled: useStreamingEncode, enabled: useStreamingEncode,
@@ -2498,7 +2535,7 @@ export async function executeRenderJob(
workerCount, workerCount,
probeSession, probeSession,
outputFormat, outputFormat,
forceParallelStream: deParallelStreamForced, forceParallelStream: deParallelStreamForced || captureParallelStreamForced,
streamingEncoderOptions: { streamingEncoderOptions: {
fps: job.config.fps, fps: job.config.fps,
width, width,