diff --git a/packages/producer/src/services/render/observability.ts b/packages/producer/src/services/render/observability.ts index af10d38b55..e03b301661 100644 --- a/packages/producer/src/services/render/observability.ts +++ b/packages/producer/src/services/render/observability.ts @@ -51,7 +51,7 @@ export interface RenderCaptureObservability { * render re-ran via screenshot. NARROWED semantics since the pinned-fallback * retry was widened (review): OOM- and generic-capture-error-triggered * fallbacks report FALSE here, with `deFallbackReason` โˆˆ {oom, - * capture_error}. The "any fallback fired" signal is `deFallbackReason` + * de_renderer_stall, encoder_death, parallel_stall, capture_error}. The "any fallback fired" signal is `deFallbackReason` * being set, NOT this flag โ€” dashboards keyed on `de_self_verify_fallback = * true` as any-fallback must migrate to `de_fallback_reason IS NOT NULL`. */ @@ -59,14 +59,15 @@ export interface RenderCaptureObservability { /** * Why the capture-stage retry (self-verify OR the pinned-worker-count * fallback) fired: "blank"/"psnr" for a real self-verify trip, - * "oom"/"capture_error" for the widened generic-failure retry. Set + * "oom"/"de_renderer_stall"/"encoder_death"/"parallel_stall"/"capture_error" + * for the widened generic-failure retry. Set * whenever a fallback is attempted, independent of whether that retry * itself later succeeds โ€” so a render that fails AFTER a fallback attempt * (perfSummary never built) is still distinguishable in failure-path * telemetry from one that never attempted any fallback. */ deFallbackReason?: string; - /** The failing PSNR (dB) when `deFallbackReason === "psnr"`; undefined for blank/oom/capture_error (no score exists). */ + /** The failing PSNR (dB) when `deFallbackReason === "psnr"`; undefined for every other reason (no score exists). */ deFallbackFailedDb?: number; /** Frame index the verification failure was detected at; set for both "psnr" and "blank" fallback reasons. */ deFallbackFrameIndex?: number; diff --git a/packages/producer/src/services/render/stages/captureStreamingStage.test.ts b/packages/producer/src/services/render/stages/captureStreamingStage.test.ts index 9e2044d895..57bf77731f 100644 --- a/packages/producer/src/services/render/stages/captureStreamingStage.test.ts +++ b/packages/producer/src/services/render/stages/captureStreamingStage.test.ts @@ -101,6 +101,9 @@ mock.module("@hyperframes/engine", () => ({ if (failWorkerTransient) { // One worker's Chrome died. The engine reports the ORIGINAL failure // through the hook, then the pool rejects with its flattened summary. + // That ordering is the engine's contract, pinned without a browser by + // parallelCoordinator.test.ts "createPoolFailureHandler"; this mock + // only replays it. injectedWorkerFailure = Object.assign( new Error("Protocol error (Page.captureScreenshot): Target closed"), { kind: "transient_browser" }, diff --git a/packages/producer/src/services/renderOrchestrator.test.ts b/packages/producer/src/services/renderOrchestrator.test.ts index 91ac23e160..221f47b808 100644 --- a/packages/producer/src/services/renderOrchestrator.test.ts +++ b/packages/producer/src/services/renderOrchestrator.test.ts @@ -43,6 +43,7 @@ import { isDeRendererStallError, isSequentialCaptureStallError, isParallelCaptureStallError, + isParallelStreamForced, fallbackCaptureModeLabel, isRetryableEncoderDeath, scanElementTags, @@ -3813,3 +3814,20 @@ describe("fallbackCaptureModeLabel: trace label when no probe session is open", ).toBe("drawelement"); }); }); + +describe("isParallelStreamForced: the plan carries the manual interleave opt-in", () => { + const flagsOff = { deParallelStreamForced: false, captureParallelStreamForced: false }; + + it("folds HF_DE_PARALLEL_STREAM=true into the plan flag so the retry drops to one worker", () => { + expect(isParallelStreamForced({ HF_DE_PARALLEL_STREAM: "true" }, flagsOff)).toBe(true); + }); + + it("otherwise follows the two routers' flags", () => { + expect(isParallelStreamForced({}, flagsOff)).toBe(false); + expect(isParallelStreamForced({ HF_DE_PARALLEL_STREAM: "false" }, flagsOff)).toBe(false); + expect(isParallelStreamForced({}, { ...flagsOff, deParallelStreamForced: true })).toBe(true); + expect(isParallelStreamForced({}, { ...flagsOff, captureParallelStreamForced: true })).toBe( + true, + ); + }); +}); diff --git a/packages/producer/src/services/renderOrchestrator.ts b/packages/producer/src/services/renderOrchestrator.ts index 48faf2c7c7..a9aaa0b120 100644 --- a/packages/producer/src/services/renderOrchestrator.ts +++ b/packages/producer/src/services/renderOrchestrator.ts @@ -626,7 +626,7 @@ export interface RenderPerfSummary { * `fallbackReason` being set is the "any fallback fired" signal. */ selfVerifyFallback: boolean; - /** What tripped the fallback retry: psnr | blank | oom | de_renderer_stall | capture_error. */ + /** What tripped the fallback retry: psnr | blank | oom | de_renderer_stall | encoder_death | parallel_stall | capture_error. */ fallbackReason?: string; /** The failing PSNR (dB) when `fallbackReason === "psnr"`; undefined for every other reason (no score exists). */ fallbackFailedDb?: number; @@ -2557,6 +2557,28 @@ export function isCaptureParallelStreamRouterEnabled( return true; } +/** + * Whether this render distributes frames interleaved across its workers. The + * two routers set their flags; `HF_DE_PARALLEL_STREAM=true` is the manual + * opt-in the streaming stage also honours on its own. Folding all three into + * the plan is what keeps the plan, and so the retry target and telemetry, in + * agreement with the distribution the stage actually picks: before this, an + * env-opt-in render below the DE router's frame floor carried + * `forceParallelStream: false`, so its retry kept N workers, which the env + * var then re-interleaved instead of dropping to the hardened single-worker + * path. Pure; exported for tests. + */ +export function isParallelStreamForced( + env: Readonly>, + flags: { deParallelStreamForced: boolean; captureParallelStreamForced: boolean }, +): boolean { + return ( + flags.deParallelStreamForced || + flags.captureParallelStreamForced || + env.HF_DE_PARALLEL_STREAM === "true" + ); +} + /** * The duration at which segmented capture is meant to become the default * (spec ยง5 Phase 2d). Not the shipped default: that flip is its own release @@ -4234,7 +4256,10 @@ async function executeRenderPipeline(input: { let capturePlan: CapturePlan = createCapturePlan({ workerCount, forceScreenshot: captureForceScreenshot, - forceParallelStream: deParallelStreamForced || captureParallelStreamForced, + forceParallelStream: isParallelStreamForced(process.env, { + deParallelStreamForced, + captureParallelStreamForced, + }), useStreamingEncode, // Segmented capture is opt-in in Phase 2a and single-worker only; the // excluded routes are the ones whose concat-copy or capture loop the @@ -4672,7 +4697,7 @@ async function executeRenderPipeline(input: { : isSequentialStall ? "[Render] sequential capture stalled; retrying on a fresh screenshot session" : isParallelStall - ? "[Render] parallel capture stalled; retrying on a fresh single-worker screenshot session" + ? "[Render] parallel capture stalled; retrying on a fresh screenshot session" : isEncoderDeath ? "[Render] streaming encoder died mid-render; retrying on a fresh screenshot session" : "[Render] capture failed; re-rendering via a fresh screenshot session", @@ -4687,7 +4712,7 @@ async function executeRenderPipeline(input: { : isSequentialStall ? "sequential capture stalled; retrying with a fresh screenshot session" : isParallelStall - ? "parallel capture stalled; retrying with a fresh single-worker screenshot session" + ? "parallel capture stalled; retrying with a fresh screenshot session" : isEncoderDeath ? "streaming encoder died; retrying with a fresh screenshot session" : "capture failed; retrying with a fresh screenshot session",