diff --git a/packages/cli/src/commands/render.ts b/packages/cli/src/commands/render.ts index 8400cdddd..e4a7af554 100644 --- a/packages/cli/src/commands/render.ts +++ b/packages/cli/src/commands/render.ts @@ -1688,6 +1688,8 @@ function trackRenderMetrics( deClampReason: perf?.drawElement?.clampReason, deWorkerInversion: perf?.drawElement?.workerInversion, dePreInversionWorkers: perf?.drawElement?.preInversionWorkers, + deParallelRouter: perf?.drawElement?.parallelRouter, + dePreRouterWorkers: perf?.drawElement?.preRouterWorkers, deGateReason: perf?.drawElement?.gateReason, deWorkerEncode: perf?.drawElement?.workerEncode, deVerifyArmed: perf?.drawElement?.verifyArmed, diff --git a/packages/cli/src/telemetry/events.ts b/packages/cli/src/telemetry/events.ts index c90392ce4..89462246a 100644 --- a/packages/cli/src/telemetry/events.ts +++ b/packages/cli/src/telemetry/events.ts @@ -126,6 +126,8 @@ export function trackRenderComplete( deClampReason?: string; deWorkerInversion?: string; dePreInversionWorkers?: number; + deParallelRouter?: string; + dePreRouterWorkers?: number; deGateReason?: string; deWorkerEncode?: boolean; deVerifyArmed?: number; @@ -204,6 +206,8 @@ export function trackRenderComplete( de_clamp_reason: props.deClampReason, de_worker_inversion: props.deWorkerInversion, de_pre_inversion_workers: props.dePreInversionWorkers, + de_parallel_router: props.deParallelRouter, + de_pre_router_workers: props.dePreRouterWorkers, de_gate_reason: props.deGateReason, de_worker_encode: props.deWorkerEncode, de_verify_armed: props.deVerifyArmed, diff --git a/packages/producer/src/services/render/observability.ts b/packages/producer/src/services/render/observability.ts index 498b2edf0..8e486dc1a 100644 --- a/packages/producer/src/services/render/observability.ts +++ b/packages/producer/src/services/render/observability.ts @@ -44,6 +44,8 @@ export interface RenderCaptureObservability { deSelfVerifyFallback?: boolean; /** Auto-parallel inversion outcome: "inverted" (fired, held) | "reverted" (fired, self-verify retry rolled back). */ deWorkerInversion?: "inverted" | "reverted"; + /** DE parallel-router outcome: "routed" (fired, held) | "reverted" (fired, self-verify retry rolled back). */ + deParallelRouter?: "routed" | "reverted"; protocolTimeoutMs?: number; pageNavigationTimeoutMs?: number; playerReadyTimeoutMs?: number; diff --git a/packages/producer/src/services/render/perfSummary.ts b/packages/producer/src/services/render/perfSummary.ts index b449c4c4b..0200ae460 100644 --- a/packages/producer/src/services/render/perfSummary.ts +++ b/packages/producer/src/services/render/perfSummary.ts @@ -67,6 +67,9 @@ export interface DrawElementPerfInput { workerInversion?: "inverted" | "reverted"; /** Auto-resolved worker count before the inversion pinned it to 1 (set only when the inversion fired). */ preInversionWorkers?: number; + parallelRouter?: "routed" | "reverted"; + /** Auto-resolved worker count before the router pinned it to 3 (set only when the router fired). */ + preRouterWorkers?: number; selfVerifyFallback: boolean; fallbackReason?: string; drainStats?: { @@ -96,6 +99,8 @@ function aggregateDrawElement( clampReason: de.clampReason, workerInversion: de.workerInversion ?? "none", preInversionWorkers: de.preInversionWorkers, + parallelRouter: de.parallelRouter ?? "none", + preRouterWorkers: de.preRouterWorkers, gateReason: gateReasons.length > 0 ? gateReasons.join("|") : undefined, workerEncode: perfs.some((p) => p.deWorkerEncode), verifyArmed: perfs.reduce((sum, p) => sum + (p.deVerifyArmed ?? 0), 0), diff --git a/packages/producer/src/services/renderOrchestrator.test.ts b/packages/producer/src/services/renderOrchestrator.test.ts index 1af6be66b..147258f1b 100644 --- a/packages/producer/src/services/renderOrchestrator.test.ts +++ b/packages/producer/src/services/renderOrchestrator.test.ts @@ -29,6 +29,8 @@ import { resolveCaptureForceScreenshotForPageSideCompositing, shouldDiscardProbeSessionForPageSideCompositing, resolveInversionRetryPlan, + resolveParallelRouterRetryPlan, + shouldPreferParallelDrawElement, shouldPreferSingleWorkerDrawElement, shouldUseStreamingEncode, } from "./renderOrchestrator.js"; @@ -1702,3 +1704,133 @@ describe("resolveInversionRetryPlan (self-verify retry rollback)", () => { }); }); }); + +describe("shouldPreferParallelDrawElement (DE parallel router)", () => { + const eligible = { + workerCount: 5, + requestedWorkers: "auto" as const, + useDrawElement: true, + deCompileGate: undefined, + forceScreenshot: false, + outputFormat: "mp4" as const, + totalFrames: 2381, + minFrames: 2000, + layeredOrEffectRoute: false, + supersampling: false, + probeDeGated: false, + experimentalParallelDeOptIn: false, + routerEnabled: true, + }; + + it("routes an auto-resolved multi-worker render for an eligible long comp", () => { + expect(shouldPreferParallelDrawElement(eligible)).toBe(true); + }); + + it("is disabled by default (routerEnabled: false is the shipped default)", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, routerEnabled: false })).toBe(false); + }); + + it("honors explicitly requested workers", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, requestedWorkers: 3 })).toBe(false); + }); + + it("routes for requestedWorkers undefined — the value production actually passes for auto", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, requestedWorkers: undefined })).toBe( + true, + ); + }); + + it("skips comps routed to layered/HDR/shader paths (drawElement never runs there)", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, layeredOrEffectRoute: true })).toBe( + false, + ); + }); + + it("skips supersampled renders (engine init-time DE gate)", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, supersampling: true })).toBe(false); + }); + + it("skips when the probe session already shows DE gated out", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, probeDeGated: true })).toBe(false); + }); + + it("honors the explicit experimental parallel-DE opt-in (already parallel, router is a no-op)", () => { + expect( + shouldPreferParallelDrawElement({ ...eligible, experimentalParallelDeOptIn: true }), + ).toBe(false); + }); + + it("skips below the amortization threshold (benchmark: real-work comps clear 1.25x at ~2,000+ frames)", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, totalFrames: 915 })).toBe(false); + expect(shouldPreferParallelDrawElement({ ...eligible, totalFrames: 2000 })).toBe(true); + }); + + it("is disabled by minFrames <= 0 (HF_DE_PARALLEL_MIN_FRAMES=0 kill switch)", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, minFrames: 0 })).toBe(false); + expect(shouldPreferParallelDrawElement({ ...eligible, minFrames: -1 })).toBe(false); + }); + + it("requires drawElement to be enabled and ungated", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, useDrawElement: false })).toBe(false); + expect(shouldPreferParallelDrawElement({ ...eligible, deCompileGate: "3d" })).toBe(false); + expect(shouldPreferParallelDrawElement({ ...eligible, forceScreenshot: true })).toBe(false); + }); + + it("only applies to mp4 (the benchmarked configuration)", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, outputFormat: "webm" })).toBe(false); + }); + + it("is a no-op when workers already resolved to 1", () => { + expect(shouldPreferParallelDrawElement({ ...eligible, workerCount: 1 })).toBe(false); + }); +}); + +describe("resolveParallelRouterRetryPlan (self-verify retry rollback)", () => { + const cfg = { enableStreamingEncode: true, streamingEncodeMaxDurationSeconds: 240 }; + + it("returns null when the render was never router-routed", () => { + expect( + resolveParallelRouterRetryPlan({ + deParallelRouter: undefined, + preRouterWorkerCount: 5, + cfg, + outputFormat: "mp4", + durationSeconds: 80, + }), + ).toBe(null); + expect( + resolveParallelRouterRetryPlan({ + deParallelRouter: "reverted", + preRouterWorkerCount: 5, + cfg, + outputFormat: "mp4", + durationSeconds: 80, + }), + ).toBe(null); + }); + + it("restores the pre-router worker count and routes multi-worker retries to disk", () => { + // Caller is responsible for clearing HF_DE_PARALLEL_STREAM before calling + // this (see the function's own doc comment) — simulated here since + // shouldUseStreamingEncode reads it directly. + const prevEnv = process.env.HF_DE_PARALLEL_STREAM; + delete process.env.HF_DE_PARALLEL_STREAM; + try { + const plan = resolveParallelRouterRetryPlan({ + deParallelRouter: "routed", + preRouterWorkerCount: 5, + cfg, + outputFormat: "mp4", + durationSeconds: 80, + }); + expect(plan).toEqual({ + workerCount: 5, + useStreamingEncode: false, + deParallelRouter: "reverted", + }); + } finally { + if (prevEnv === undefined) delete process.env.HF_DE_PARALLEL_STREAM; + else process.env.HF_DE_PARALLEL_STREAM = prevEnv; + } + }); +}); diff --git a/packages/producer/src/services/renderOrchestrator.ts b/packages/producer/src/services/renderOrchestrator.ts index 4e241748d..34f3a0a89 100644 --- a/packages/producer/src/services/renderOrchestrator.ts +++ b/packages/producer/src/services/renderOrchestrator.ts @@ -387,6 +387,10 @@ export interface RenderPerfSummary { workerInversion?: string; /** Worker count the auto-resolution chose BEFORE the inversion pinned it to 1 — the parallel counterfactual for speedup math. Only set when the inversion fired. */ preInversionWorkers?: number; + /** DE parallel-router outcome: "routed" (fired, held), "reverted" (fired, self-verify retry rolled back), "none". Mutually exclusive with workerInversion. */ + parallelRouter?: string; + /** Worker count the auto-resolution chose BEFORE the router pinned it to 3 — the single-worker-inversion counterfactual. Only set when the router fired. */ + preRouterWorkers?: number; /** Engine init-time gate: swiftshader | css_effect:* | at_risk_timeline | 3d_init_failed | supersampling | render_mode_hint. */ gateReason?: string; /** Worker-encode drain (the verified path) was active. */ @@ -1065,6 +1069,96 @@ export function resolveInversionRetryPlan(args: { }; } +/** + * DE parallel-router predicate: should an AUTO-resolved multi-worker render + * use VERIFIED PARALLEL drawElement streaming (HF_DE_PARALLEL_STREAM) instead + * of the #2026 single-worker inversion? + * + * Benchmarked 2026-07-08 (clean, quiet-machine re-run): par3/single 1.16–1.36x + * on real-work comps ≥2,000 frames (2,381f GSAP graphics 1.36x, 3,245f rAF + * high-variance 1.29x, 915f crossover probe 1.27x); the one comp that didn't + * clear 1.25x (3,600f, 39% static/dedup-heavy) still didn't LOSE to single- + * worker (1.16x) — dedup already skips the capture work parallelism would + * split, so there's mechanically less headroom, not a regression. No comp + * anywhere showed par3 < single. Default-off (HF_DE_PARALLEL_ROUTER): this + * promotes the opt-in mechanism from #2056 into the auto-routing decision, + * but the decision itself stays gated behind its own flag pending the + * telemetry soak (revert rate, de_verify_min_db distribution) on real wild + * traffic — there is currently none, since nothing routes here by default. + * Takes priority over the single-worker inversion when both would fire (a + * higher minFrames than HF_DE_SINGLE_MIN_FRAMES is the intended shape: this + * only picks up the long tail the inversion's own benchmark didn't cover). + */ +export function shouldPreferParallelDrawElement(args: { + workerCount: number; + /** job.config.workers — a number means the user explicitly chose. */ + requestedWorkers: number | "auto" | undefined; + useDrawElement: boolean; + deCompileGate: string | undefined; + forceScreenshot: boolean; + outputFormat: NonNullable; + totalFrames: number; + /** Amortization threshold; <=0 disables the router. */ + minFrames: number; + layeredOrEffectRoute: boolean; + supersampling: boolean; + probeDeGated: boolean; + experimentalParallelDeOptIn: boolean; + /** HF_DE_PARALLEL_ROUTER === "true" — the router's own kill switch, default off. */ + routerEnabled: boolean; +}): boolean { + return ( + args.routerEnabled && + args.workerCount > 1 && + typeof args.requestedWorkers !== "number" && + args.useDrawElement && + !args.deCompileGate && + !args.forceScreenshot && + args.outputFormat === "mp4" && + args.minFrames > 0 && + args.totalFrames >= args.minFrames && + !args.layeredOrEffectRoute && + !args.supersampling && + !args.probeDeGated && + !args.experimentalParallelDeOptIn + ); +} + +/** + * Plan the self-verify retry for a router-routed render: the bet on verified + * parallel drawElement streaming lost, so the re-render falls back to the + * pre-router worker count on the ordinary (non-DE) parallel path. Unlike + * `resolveInversionRetryPlan`, the caller must also clear + * `process.env.HF_DE_PARALLEL_STREAM` (set by the router) BEFORE calling + * this — `shouldUseStreamingEncode` reads that env var directly, so an + * uncleared flag would keep resolving to the parallel-streaming shape on the + * retry instead of the well-tested parallel-disk fallback. Returns null when + * the render was not router-routed. + */ +export function resolveParallelRouterRetryPlan(args: { + deParallelRouter: "routed" | "reverted" | undefined; + preRouterWorkerCount: number; + cfg: Pick; + outputFormat: NonNullable; + durationSeconds: number; +}): { + workerCount: number; + useStreamingEncode: boolean; + deParallelRouter: "reverted"; +} | null { + if (args.deParallelRouter !== "routed") return null; + return { + workerCount: args.preRouterWorkerCount, + useStreamingEncode: shouldUseStreamingEncode( + args.cfg, + args.outputFormat, + args.preRouterWorkerCount, + args.durationSeconds, + ), + deParallelRouter: "reverted", + }; +} + export function resolveCaptureForceScreenshotForPageSideCompositing(args: { forceScreenshot: boolean; usePageSideCompositing: boolean; @@ -1341,6 +1435,11 @@ export async function executeRenderJob( // "inverted" = fired and held; "reverted" = fired but the self-verify // retry rolled back to the parallel path; undefined = never fired. let deWorkerInversion: "inverted" | "reverted" | undefined; + // "routed" = the parallel router fired and held; "reverted" = fired but + // the self-verify retry rolled back; undefined = never fired. Mutually + // exclusive with deWorkerInversion — the router takes priority when both + // would be eligible (see shouldPreferParallelDrawElement). + let deParallelRouter: "routed" | "reverted" | undefined; let deSelfVerifyFallback = false; let deFallbackReason: string | undefined; let deDrainStats: import("./render/stages/captureStreamingStage.js").DeDrainStats | undefined; @@ -1718,12 +1817,46 @@ export async function executeRenderJob( // Verified parallel DE streaming (opt-in) wants its parallelism kept. process.env.HF_DE_PARALLEL_STREAM === "true", }); + // DE parallel-router eligibility — see shouldPreferParallelDrawElement. + // Default-off (HF_DE_PARALLEL_ROUTER); HF_DE_PARALLEL_MIN_FRAMES defaults + // higher than the single-worker inversion's threshold since it targets + // the long tail the inversion's own benchmark didn't cover. + const deParallelRouterEnabled = process.env.HF_DE_PARALLEL_ROUTER === "true"; + const deParallelMinFramesRaw = process.env.HF_DE_PARALLEL_MIN_FRAMES; + const deParallelMinFramesNum = + deParallelMinFramesRaw === undefined || deParallelMinFramesRaw.trim() === "" + ? 2000 + : Number(deParallelMinFramesRaw); + const deParallelMinFrames = Number.isFinite(deParallelMinFramesNum) + ? deParallelMinFramesNum + : 2000; + const deParallelRouterEligible = shouldPreferParallelDrawElement({ + workerCount: WOULD_RESOLVE_MULTI_WORKER, + requestedWorkers: job.config.workers, + useDrawElement: cfg.useDrawElement, + deCompileGate, + forceScreenshot: captureForceScreenshot, + outputFormat, + totalFrames, + minFrames: deParallelMinFrames, + layeredOrEffectRoute: hasHdrContent || compiled.hasShaderTransitions, + supersampling: deviceScaleFactor > 1, + probeDeGated: + probeSession !== null && + probeSession.captureMode !== "drawelement" && + !probeSession.deInitDeferred, + experimentalParallelDeOptIn: + process.env.PRODUCER_EXPERIMENTAL_FAST_CAPTURE === "true" || + process.env.HF_DE_PARALLEL_STREAM === "true", + routerEnabled: deParallelRouterEnabled, + }); if ( job.config.workers === undefined && totalFrames >= 60 && !htmlInCanvasDetected && !cfg.lowMemoryMode && - !deInversionEligible + !deInversionEligible && + !deParallelRouterEligible ) { const outcome = await observeRenderStage( observability, @@ -1763,6 +1896,7 @@ export async function executeRenderJob( htmlInCanvasDetected, lowMemoryMode: Boolean(cfg.lowMemoryMode), deInversionEligible, + deParallelRouterEligible, }); } @@ -1785,7 +1919,25 @@ export async function executeRenderJob( // `preInversionWorkerCount` lets the self-verify retry return to the // parallel path when the drawElement bet loses. const preInversionWorkerCount = workerCount; - if (deInversionEligible && workerCount > 1) { + // Router takes priority over the single-worker inversion when both would + // fire — its higher frame threshold means this only ever picks up long- + // tail comps the inversion's own benchmark didn't cover (see + // shouldPreferParallelDrawElement). + if (deParallelRouterEligible && workerCount > 1) { + deParallelRouter = "routed"; + // Fixed at 3, not calibration-derived: the benchmark validated exactly + // this worker count (par3 beat par2 consistently; W4/W5 unmeasured for + // this path), same shape as the single-worker inversion pinning to a + // fixed 1 rather than a calibrated count. + const ROUTER_WORKER_COUNT = 3; + log.info( + "[Render] Fast capture: verified parallel drawElement streaming preferred over " + + `single-worker inversion (${totalFrames} frames >= ${deParallelMinFrames}; ` + + "benchmark-validated at 3 workers). Set HF_DE_PARALLEL_ROUTER=false or --workers N to override.", + ); + workerCount = ROUTER_WORKER_COUNT; + process.env.HF_DE_PARALLEL_STREAM = "true"; + } else if (deInversionEligible && workerCount > 1) { deWorkerInversion = "inverted"; log.info( "[Render] Fast capture: single-worker drawElement streaming preferred over " + @@ -1795,10 +1947,11 @@ export async function executeRenderJob( ); workerCount = 1; } - updateCaptureObservability({ workerCount, deWorkerInversion }); + updateCaptureObservability({ workerCount, deWorkerInversion, deParallelRouter }); observability.checkpoint("worker_resolution", "resolved", { workerCount, deWorkerInversion: deWorkerInversion ?? "none", + deParallelRouter: deParallelRouter ?? "none", }); if (workerCount > 1 && probeSession) { @@ -2101,6 +2254,11 @@ export async function executeRenderJob( deSelfVerifyFallback: true, }); probeSession = null; + // HF_DE_PARALLEL_STREAM must be cleared BEFORE resolveParallelRouterRetryPlan + // recomputes useStreamingEncode, or shouldUseStreamingEncode's own env + // check would keep resolving to the parallel-streaming shape on the + // retry instead of the well-tested parallel-disk fallback. + if (deParallelRouter === "routed") delete process.env.HF_DE_PARALLEL_STREAM; const inversionRetryPlan = resolveInversionRetryPlan({ deWorkerInversion, preInversionWorkerCount, @@ -2108,6 +2266,13 @@ export async function executeRenderJob( outputFormat, durationSeconds: job.duration, }); + const parallelRouterRetryPlan = resolveParallelRouterRetryPlan({ + deParallelRouter, + preRouterWorkerCount: preInversionWorkerCount, + cfg, + outputFormat, + durationSeconds: job.duration, + }); if (inversionRetryPlan) { // The inversion bet on drawElement and lost — re-render on the // pre-inversion parallel screenshot path instead of single-worker @@ -2126,6 +2291,23 @@ export async function executeRenderJob( `[Render] Reverting worker inversion for the retry: ${workerCount} workers, ` + `streaming=${useStreamingEncode}.`, ); + } else if (parallelRouterRetryPlan) { + // The router's bet on verified parallel streaming lost — re-render + // on the ordinary (non-DE) parallel path at the pre-router worker + // count, same "reverted, not cleared" telemetry contract as the + // inversion above. + deParallelRouter = parallelRouterRetryPlan.deParallelRouter; + workerCount = parallelRouterRetryPlan.workerCount; + useStreamingEncode = parallelRouterRetryPlan.useStreamingEncode; + updateCaptureObservability({ + workerCount, + useStreamingEncode, + deParallelRouter, + }); + log.info( + `[Render] Reverting parallel router for the retry: ${workerCount} workers, ` + + `streaming=${useStreamingEncode}.`, + ); } if (useStreamingEncode) { streamingRes = await invokeStreaming(); @@ -2338,6 +2520,8 @@ export async function executeRenderJob( clampReason: deClampReason, workerInversion: deWorkerInversion, preInversionWorkers: deWorkerInversion ? preInversionWorkerCount : undefined, + parallelRouter: deParallelRouter, + preRouterWorkers: deParallelRouter ? preInversionWorkerCount : undefined, selfVerifyFallback: deSelfVerifyFallback, fallbackReason: deFallbackReason, drainStats: deDrainStats,