From 0045e8b3c5a44eeb95beb05dbf550ebaed501a43 Mon Sep 17 00:00:00 2001 From: Vance Ingalls Date: Thu, 9 Jul 2026 00:54:09 -0700 Subject: [PATCH] fix(producer): stop router from mutating process.env for cross-render state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HF_DE_PARALLEL_STREAM was restored on every exit path, but the producer server allows concurrent renders in one process — a router-eligible job's mutation was still visible to an unrelated render already executing during that window. Thread the router's decision as a per-render local instead of a global env var; HF_DE_PARALLEL_STREAM stays as the manual opt-in. Co-Authored-By: Claude Sonnet 5 --- .../render/stages/captureStreamingStage.ts | 24 ++-- .../src/services/renderOrchestrator.test.ts | 70 ++++-------- .../src/services/renderOrchestrator.ts | 108 ++++++++---------- 3 files changed, 86 insertions(+), 116 deletions(-) diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.ts b/packages/producer/src/services/render/stages/captureStreamingStage.ts index 5305ffb82..17ffd26c9 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.ts @@ -113,6 +113,14 @@ export interface CaptureStreamingStageInput { log: ProducerLogger; workerCount: number; probeSession: CaptureSession | null; + /** + * Per-render override from the DE parallel router — see + * deParallelStreamForced's declaration in renderOrchestrator.ts. Distinct + * from the `HF_DE_PARALLEL_STREAM` manual opt-in (still read directly by + * this stage) because the router's decision must not leak across + * concurrently-running renders sharing this process via a global env var. + */ + forceParallelStream?: boolean; /** For the spawn-failure log message context only. */ outputFormat: string; /** Pre-built encoder options; passed straight to `spawnStreamingEncoder`. */ @@ -456,6 +464,7 @@ export async function runCaptureStreamingStage( assertNotAborted, onProgress, dedupPerfs, + forceParallelStream, } = input; let { workerCount, probeSession } = input; let lastBrowserConsole: string[] = []; @@ -507,13 +516,14 @@ export async function runCaptureStreamingStage( if (workerCount > 1) { // Parallel capture → streaming encode - // HF_DE_PARALLEL_STREAM (opt-in): interleaved distribution — worker i - // takes frames i, i+N, i+2N… so the ordered writer's reorder window is - // N frames and workers run in lockstep instead of serializing behind - // contiguous ranges (see distributeFramesInterleaved). Each worker runs - // the depth-2 pipelined drawElement produce when its session initialized - // in drawelement mode. - const deParallelStream = process.env.HF_DE_PARALLEL_STREAM === "true"; + // HF_DE_PARALLEL_STREAM (manual opt-in) / forceParallelStream (router): + // interleaved distribution — worker i takes frames i, i+N, i+2N… so the + // ordered writer's reorder window is N frames and workers run in + // lockstep instead of serializing behind contiguous ranges (see + // distributeFramesInterleaved). Each worker runs the depth-2 pipelined + // drawElement produce when its session initialized in drawelement mode. + const deParallelStream = + forceParallelStream === true || process.env.HF_DE_PARALLEL_STREAM === "true"; const tasks = deParallelStream ? distributeFramesInterleaved(totalFrames, workerCount, workDir) : distributeFrames(totalFrames, workerCount, workDir); diff --git a/packages/producer/src/services/renderOrchestrator.test.ts b/packages/producer/src/services/renderOrchestrator.test.ts index 58b522b3c..e6c5bd182 100644 --- a/packages/producer/src/services/renderOrchestrator.test.ts +++ b/packages/producer/src/services/renderOrchestrator.test.ts @@ -30,7 +30,6 @@ import { shouldDiscardProbeSessionForPageSideCompositing, resolveInversionRetryPlan, resolveParallelRouterRetryPlan, - restoreEnv, shouldPreferParallelDrawElement, shouldPreferSingleWorkerDrawElement, shouldUseStreamingEncode, @@ -405,6 +404,14 @@ describe("shouldUseStreamingEncode", () => { expect(shouldUseStreamingEncode(streamingEnabledConfig, "mp4", 2, 240)).toBe(false); }); + it("forceParallelStream overrides the parallel-capture clamp for verified multi-worker streaming", () => { + expect(shouldUseStreamingEncode(streamingEnabledConfig, "mp4", 3, 240, true)).toBe(true); + expect(shouldUseStreamingEncode(streamingEnabledConfig, "mp4", 3, 240, false)).toBe(false); + expect(shouldUseStreamingEncode(streamingEnabledConfig, "png-sequence", 3, 240, true)).toBe( + false, + ); + }); + it("keeps renders over the configured max duration on normal encoding", () => { expect(shouldUseStreamingEncode(streamingEnabledConfig, "mp4", 1, 240)).toBe(true); expect(shouldUseStreamingEncode(streamingEnabledConfig, "mp4", 1, 240.001)).toBe(false); @@ -1811,54 +1818,17 @@ describe("resolveParallelRouterRetryPlan (self-verify retry rollback)", () => { }); 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 { - restoreEnv("HF_DE_PARALLEL_STREAM", prevEnv); - } - }); -}); - -describe("restoreEnv (DE parallel-router leak-safety primitive)", () => { - const VAR = "HF_DE_PARALLEL_STREAM_TEST_VAR"; - const prev = process.env[VAR]; - - afterEach(() => { - if (prev === undefined) delete process.env[VAR]; - else process.env[VAR] = prev; - }); - - it("deletes the var when the captured prior value was undefined", () => { - process.env[VAR] = "true"; - restoreEnv(VAR, undefined); - expect(process.env[VAR]).toBeUndefined(); - }); - - it("restores the exact prior string value, including a falsy-looking one", () => { - process.env[VAR] = "true"; - restoreEnv(VAR, "false"); - expect(process.env[VAR]).toBe("false"); - }); - - it("is a no-op shape when there was nothing to restore", () => { - delete process.env[VAR]; - restoreEnv(VAR, undefined); - expect(process.env[VAR]).toBeUndefined(); + const plan = resolveParallelRouterRetryPlan({ + deParallelRouter: "routed", + preRouterWorkerCount: 5, + cfg, + outputFormat: "mp4", + durationSeconds: 80, + }); + expect(plan).toEqual({ + workerCount: 5, + useStreamingEncode: false, + deParallelRouter: "reverted", + }); }); }); diff --git a/packages/producer/src/services/renderOrchestrator.ts b/packages/producer/src/services/renderOrchestrator.ts index 5a8c8299e..614ebb75c 100644 --- a/packages/producer/src/services/renderOrchestrator.ts +++ b/packages/producer/src/services/renderOrchestrator.ts @@ -955,38 +955,29 @@ function replaceBodyWithRenderClone(body: HTMLElement, renderClone: Element): vo body.appendChild(renderClone); } -/** - * Restore an env var to a previously-captured value (`undefined` deletes it - * rather than setting the literal string "undefined"). Used to make the DE - * parallel-router's HF_DE_PARALLEL_STREAM mutation safe to leave uncleared on - * any exit path other than the one that explicitly calls this — see the - * outer `finally` in executeRenderJob (review: an unrestored env var leaks - * into the next render sharing this process). - */ -export function restoreEnv(name: string, prev: string | undefined): void { - if (prev === undefined) delete process.env[name]; - else process.env[name] = prev; -} - export function shouldUseStreamingEncode( cfg: Pick, outputFormat: NonNullable, workerCount: number, // Composition timeline duration in seconds. durationSeconds: number, + // Per-render override (set by the DE parallel router) — see + // deParallelStreamForced's declaration in executeRenderJob for why this is + // a parameter instead of an env-var read. + forceParallelStream = false, ): boolean { if (!cfg.enableStreamingEncode) return false; if (outputFormat === "png-sequence") return false; if (outputFormat === "gif") return false; if (!Number.isFinite(durationSeconds) || durationSeconds <= 0) return false; if (durationSeconds > cfg.streamingEncodeMaxDurationSeconds) return false; - // SPIKE (HF_DE_PARALLEL_STREAM): allow multi-worker streaming for the - // interleaved drawElement produce experiment. Contiguous-chunk parallel - // streaming stalls (worker k+1's first frame waits for ALL of worker k's), - // so this only makes sense with the interleaved distribution the capture - // stage selects under the same flag. Explicit opt-in, unverified — do not - // ship default-on without threading guardFrame through onFrameBuffer. - if (process.env.HF_DE_PARALLEL_STREAM === "true") return true; + // HF_DE_PARALLEL_STREAM (manual opt-in) / forceParallelStream (router): + // allow multi-worker streaming for the interleaved drawElement produce + // path. Contiguous-chunk parallel streaming stalls (worker k+1's first + // frame waits for ALL of worker k's), so this only makes sense with the + // interleaved distribution the capture stage selects under the same + // condition. + if (forceParallelStream || process.env.HF_DE_PARALLEL_STREAM === "true") return true; return workerCount === 1; } @@ -1141,12 +1132,11 @@ export function shouldPreferParallelDrawElement(args: { * 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. + * `resolveInversionRetryPlan`, the caller must also clear the router's + * `deParallelStreamForced` local BEFORE calling this — `shouldUseStreamingEncode` + * takes it as a direct argument, so a stale `true` 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; @@ -1324,13 +1314,9 @@ export async function executeRenderJob( // between declaration and the try-block (currently impossible, but // defensible if more setup ever lands here) can't leak the interval. let memSampler: MemorySampler | null = null; - // Same reason these two live out here rather than with the rest of the DE - // state below (which is fine staying try-scoped, nothing outside the try - // reads it): a `let` declared inside `try {}` is NOT visible in the - // sibling `finally {}` block in JS — they're independent block scopes — - // so the leak-safety restore in `finally` needs these declared here. + // "routed" = the parallel router fired and held; "reverted" = fired but + // the self-verify retry rolled back; undefined = never fired. let deParallelRouter: "routed" | "reverted" | undefined; - let deParallelStreamEnvBefore: string | undefined; try { memSampler = createMemorySampler(); @@ -1455,12 +1441,20 @@ 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; - // deParallelRouter: "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). Declared above the try (with - // deParallelStreamEnvBefore) so the outer `finally` can read it. + // deParallelRouter is mutually exclusive with deWorkerInversion — the + // router takes priority when both would be eligible (see + // shouldPreferParallelDrawElement). + // + // Per-render (not process-global) signal that the router wants parallel + // drawElement streaming. `HF_DE_PARALLEL_STREAM` env var stays as the + // manual opt-in for local testing (read directly by + // shouldUseStreamingEncode / the capture stage), but the router itself + // must NOT mutate process.env: the producer server runs concurrent + // renders in one process (PRODUCER_MAX_CONCURRENT_RENDERS), and a global + // flag set by one render's router decision would leak into an unrelated + // 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; let deSelfVerifyFallback = false; let deFallbackReason: string | undefined; let deDrainStats: import("./render/stages/captureStreamingStage.js").DeDrainStats | undefined; @@ -1965,11 +1959,7 @@ export async function executeRenderJob( "Set HF_DE_PARALLEL_ROUTER=false or --workers N to override.", ); workerCount = ROUTER_WORKER_COUNT; - // Captured BEFORE the mutation so the outer `finally` can restore the - // exact prior value on every exit path, not just the self-verify - // retry below (review) — see deParallelStreamEnvBefore's declaration. - deParallelStreamEnvBefore = process.env.HF_DE_PARALLEL_STREAM; - process.env.HF_DE_PARALLEL_STREAM = "true"; + deParallelStreamForced = true; } else if (deInversionEligible && workerCount > 1) { deWorkerInversion = "inverted"; log.info( @@ -1998,7 +1988,13 @@ export async function executeRenderJob( // let auto-parallel renders use disk frames: the current ordered streaming // writer would otherwise stall later workers behind earlier frame ranges. // png-sequence has no encoded video output, so streaming is always bypassed. - let useStreamingEncode = shouldUseStreamingEncode(cfg, outputFormat, workerCount, job.duration); + let useStreamingEncode = shouldUseStreamingEncode( + cfg, + outputFormat, + workerCount, + job.duration, + deParallelStreamForced, + ); log.info("streaming-encode gate", { enabled: useStreamingEncode, configFlag: cfg.enableStreamingEncode, @@ -2017,7 +2013,9 @@ export async function executeRenderJob( // drain guard), so the confinement rule is satisfied and the parallel // clamp does not apply. The disk path stays clamped. const deParallelStreamVerified = - process.env.HF_DE_PARALLEL_STREAM === "true" && useStreamingEncode && workerCount > 1; + (deParallelStreamForced || process.env.HF_DE_PARALLEL_STREAM === "true") && + useStreamingEncode && + workerCount > 1; if ( cfg.useDrawElement && process.env.PRODUCER_EXPERIMENTAL_FAST_CAPTURE !== "true" && @@ -2238,6 +2236,7 @@ export async function executeRenderJob( workerCount, probeSession, outputFormat, + forceParallelStream: deParallelStreamForced, streamingEncoderOptions: { fps: job.config.fps, width, @@ -2287,12 +2286,11 @@ export async function executeRenderJob( deSelfVerifyFallback: true, }); probeSession = null; - // HF_DE_PARALLEL_STREAM must be restored 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") - restoreEnv("HF_DE_PARALLEL_STREAM", deParallelStreamEnvBefore); + // Must clear BEFORE resolveParallelRouterRetryPlan recomputes + // useStreamingEncode, or shouldUseStreamingEncode would keep + // resolving to the parallel-streaming shape on the retry instead + // of the well-tested parallel-disk fallback. + if (deParallelRouter === "routed") deParallelStreamForced = false; const inversionRetryPlan = resolveInversionRetryPlan({ deWorkerInversion, preInversionWorkerCount: preRoutingWorkerCount, @@ -2717,13 +2715,5 @@ export async function executeRenderJob( throw error; } finally { memSampler?.stop(); - // Guaranteed restore regardless of exit path (success, any thrown error, - // abort) — the DE self-verify retry path also restores this mid-render - // (so ITS OWN recomputation sees the right value), but that branch is - // only one of several ways this render can end; this is the actual - // leak fix (review). - if (deParallelStreamEnvBefore !== undefined || deParallelRouter) { - restoreEnv("HF_DE_PARALLEL_STREAM", deParallelStreamEnvBefore); - } } }