fix(engine): self-verify parallel disk drawElement samples (PRINFRA-352)

Author: vanceingalls

packages/engine/src/index.ts

@@ -209,6 +209,7 @@ export type {
 export {
   calculateOptimalWorkers,
   computeWorkerSizing,
+  selectVerifySampleIndicesForTask,
   distributeFrames,
   distributeFramesInterleaved,
   executeParallelCapture,
@@ -273,6 +274,10 @@ export {
 
 export { trackChildProcess, killTrackedProcesses } from "./utils/processTracker.js";
 
+// drawElement self-verify comparison — shared by the streaming drain
+// (producer) and the parallel disk-path verify (parallelCoordinator).
+export { psnrDb, resolveDeVerifyMinDb } from "./utils/psnr.js";
+
 export {
   decodePng,
   decodePngToRgb48le,

packages/engine/src/services/parallelCoordinator.test.ts

@@ -6,6 +6,7 @@ import {
   expectedFramesForTask,
   flagSilentWorkerExits,
   formatWorkerFailure,
+  selectVerifySampleIndicesForTask,
   selectWorkerDiagnostics,
   shouldDisableBrowserPoolForParallelWorker,
   shouldVerifyWorkerGpu,
@@ -349,6 +350,35 @@ describe("flagSilentWorkerExits", () => {
   });
 });
 
+describe("selectVerifySampleIndicesForTask", () => {
+  it("keeps only samples inside the task's contiguous range, sorted", () => {
+    // 2-worker split of 3032 frames: worker 1 owns [1516, 3032).
+    expect(
+      selectVerifySampleIndicesForTask([2274, 758, 1516, 3031, 3032], {
+        startFrame: 1516,
+        endFrame: 3032,
+      }),
+    ).toEqual([1516, 2274, 3031]);
+  });
+
+  it("respects the stride lattice for interleaved tasks", () => {
+    // Worker 1 of a 3-way interleave over [1, 30): captures 1, 4, 7, ...
+    expect(
+      selectVerifySampleIndicesForTask([1, 2, 4, 6, 7, 28, 29], {
+        startFrame: 1,
+        endFrame: 30,
+        frameStride: 3,
+      }),
+    ).toEqual([1, 4, 7, 28]);
+  });
+
+  it("returns empty when no samples fall in the range", () => {
+    expect(
+      selectVerifySampleIndicesForTask([0, 10, 20], { startFrame: 100, endFrame: 200 }),
+    ).toEqual([]);
+  });
+});
+
 describe("shouldVerifyWorkerGpu", () => {
   const softwareConfig: Partial<EngineConfig> = { browserGpuMode: "software" };
 

packages/engine/src/services/parallelCoordinator.ts

@@ -7,7 +7,7 @@
 
 import { cpus, freemem } from "os";
 import { existsSync, mkdirSync, readdirSync } from "fs";
-import { copyFile, rename } from "fs/promises";
+import { copyFile, readFile, rename } from "fs/promises";
 import { join } from "path";
 import { getHeapStatistics } from "v8";
 
@@ -19,11 +19,13 @@ import {
   captureFrameToBufferPipelined,
   captureFrameToBuffer,
   getCapturePerfSummary,
+  DrawElementVerificationError,
   type CaptureSession,
   type CaptureOptions,
   type CapturePerfSummary,
   type BeforeCaptureHook,
 } from "./frameCapture.js";
+import { psnrDb, resolveDeVerifyMinDb } from "../utils/psnr.js";
 import { DEFAULT_CONFIG, type EngineConfig } from "../config.js";
 import { assertSwiftShader } from "../utils/assertSwiftShader.js";
 import { readWebGlVendorInfoFromCanvas } from "../utils/readWebGlVendorInfoFromCanvas.js";
@@ -551,6 +553,126 @@ async function captureFrameRange(
   return framesCaptured;
 }
 
+/**
+ * The armed self-verify sample indices this task actually captured: inside
+ * `[startFrame, endFrame)` and on the task's stride lattice. Mirrors the
+ * capture loop in `captureFrameRange` (`i += stride` from `startFrame`).
+ */
+export function selectVerifySampleIndicesForTask(
+  sampleIndices: Iterable<number>,
+  task: Pick<WorkerTask, "startFrame" | "endFrame" | "frameStride">,
+): number[] {
+  const stride = task.frameStride ?? 1;
+  const selected: number[] = [];
+  for (const idx of sampleIndices) {
+    if (idx < task.startFrame || idx >= task.endFrame) continue;
+    if ((idx - task.startFrame) % stride !== 0) continue;
+    selected.push(idx);
+  }
+  return selected.sort((a, b) => a - b);
+}
+
+/**
+ * Disk-path drawElement self-verification (PRINFRA-352). Parallel DISK
+ * workers arm the same pre-injection ground-truth samples as the streaming
+ * path (`resolveParallelDeVerifySamples` even raises the density for
+ * multi-worker capture) — but only the streaming drain ever CHECKED them,
+ * so an explicit `--experimental-fast-capture --workers N` render shipped
+ * unverified drawElement frames. On a 16GB host, two concurrent
+ * hardware-GPU Chrome instances hit the documented compositor-tile-eviction
+ * damage class (frames displaced into vertical strips for one worker's
+ * whole range — reads as "corruption from the exact worker boundary").
+ *
+ * After a worker's range completes, re-read its captured files for the
+ * sampled indices and PSNR-compare against the session's ground truth.
+ * A breach throws `DrawElementVerificationError`, which the orchestrator's
+ * existing pinned-fallback retry converts into a screenshot re-render —
+ * the same recovery the streaming drain gets.
+ */
+/** HF_DE_PAR_DEBUG=1 gated per-worker trace line (message built lazily). */
+function logParDebug(message: () => string): void {
+  if (process.env.HF_DE_PAR_DEBUG === "1") console.log(message());
+}
+
+/**
+ * Throw the verification error for a sample below the PSNR floor; log the
+ * pass otherwise. Split from the sampling loop for the complexity gate.
+ */
+function assertDiskSampleAboveFloor(
+  db: number,
+  verifyMinDb: number,
+  idx: number,
+  workerId: number,
+): void {
+  if (db < verifyMinDb) {
+    // Message keeps the contiguous "drawElement self-verify" phrase —
+    // captureFailure's VERIFICATION_ERROR_PATTERNS classifies on it.
+    throw new DrawElementVerificationError(
+      `drawElement self-verify failed at frame ${idx} (disk path, worker ${workerId}): ` +
+        `${db.toFixed(1)}dB < ${verifyMinDb}dB vs pre-injection screenshot`,
+      { kind: "psnr", frameIndex: idx, failedDb: db, verifyThresholdDb: verifyMinDb },
+    );
+  }
+  console.log(
+    `[Parallel] drawElement disk self-verify passed (worker ${workerId}, frame ${idx}, ` +
+      `${db === Infinity ? "inf" : db.toFixed(1)}dB)`,
+  );
+}
+
+/**
+ * Compare one captured frame file against its ground truth. Returns the
+ * PSNR, or null on infrastructure failure (missing file already surfaces
+ * via the frame completeness check; ffmpeg spawn/tmpdir here) — a skipped
+ * sample is not damage evidence and must not fail the capture.
+ */
+async function psnrForDiskSample(
+  framePath: string,
+  truth: Buffer,
+  workerId: number,
+  idx: number,
+): Promise<number | null> {
+  try {
+    return await psnrDb(await readFile(framePath), truth);
+  } catch (err) {
+    console.warn(
+      `[Parallel] drawElement disk self-verify sample skipped (worker ${workerId}, ` +
+        `frame ${idx}): ${err instanceof Error ? err.message : String(err)}`,
+    );
+    return null;
+  }
+}
+
+// Branches are the gate conditions themselves (mode/armed/streaming guards +
+// per-sample skip/breach) — already decomposed into psnrForDiskSample +
+// assertDiskSampleAboveFloor; further splitting obscures the check.
+// fallow-ignore-next-line complexity
+async function verifyDiskDrawElementSamples(
+  session: CaptureSession,
+  task: WorkerTask,
+  streaming: boolean,
+): Promise<void> {
+  // Streaming capture verifies every sampled frame in the drain guard already.
+  if (streaming || session.captureMode !== "drawelement") return;
+  const truths = session.deVerifyFrames;
+  if (!truths || truths.size === 0) return;
+  const verifyMinDb = resolveDeVerifyMinDb();
+  const ext = session.options.format === "png" ? "png" : "jpg";
+  const offset = task.outputFrameOffset ?? 0;
+  for (const idx of selectVerifySampleIndicesForTask(truths.keys(), task)) {
+    const truth = truths.get(idx);
+    if (!truth) continue;
+    const framePath = join(task.outputDir, `frame_${String(idx - offset).padStart(6, "0")}.${ext}`);
+    const db = await psnrForDiskSample(framePath, truth, task.workerId, idx);
+    if (db === null) continue;
+    assertDiskSampleAboveFloor(db, verifyMinDb, idx, task.workerId);
+  }
+}
+
+// Inherited worker-lifecycle shape (session create → verify GPU → init →
+// capture → self-verify → perf, with a classifying catch + closing finally);
+// flagged only because the disk self-verify call shifted its line range into
+// the changed-code audit. Not restructured by this PR.
+// fallow-ignore-next-line complexity
 async function executeWorkerTask(
   task: WorkerTask,
   serverUrl: string,
@@ -590,19 +712,16 @@ async function executeWorkerTask(
       createBeforeCaptureHook(),
       workerConfig,
     );
-    if (process.env.HF_DE_PAR_DEBUG === "1") {
-      console.log(`[par:w${task.workerId}] session created`);
-    }
+    logParDebug(() => `[par:w${task.workerId}] session created`);
     // Worker-0-only SwiftShader assertion — see `shouldVerifyWorkerGpu` and #955.
     if (shouldVerifyWorkerGpu(task.workerId, workerConfig)) {
       await assertSwiftShader(session.page, readWebGlVendorInfoFromCanvas);
     }
     await initializeSession(session);
-    if (process.env.HF_DE_PAR_DEBUG === "1") {
-      console.log(
-        `[par:w${task.workerId}] init done (mode=${session.captureMode} workerEncode=${session.workerEncodeEnabled === true})`,
-      );
-    }
+    logParDebug(
+      () =>
+        `[par:w${task.workerId}] init done (mode=${session?.captureMode} workerEncode=${session?.workerEncodeEnabled === true})`,
+    );
     framesCaptured = await captureFrameRange(
       session,
       task,
@@ -612,6 +731,8 @@ async function executeWorkerTask(
       onFrameBuffer,
     );
 
+    await verifyDiskDrawElementSamples(session, task, Boolean(onFrameBuffer));
+
     perf = getCapturePerfSummary(session);
     return {
       workerId: task.workerId,

packages/engine/src/utils/psnr.ts

@@ -0,0 +1,42 @@
+import { execFile } from "node:child_process";
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { promisify } from "node:util";
+import { getFfmpegBinary } from "./ffmpegBinaries.js";
+
+const execFileP = promisify(execFile);
+
+/**
+ * PSNR (average, dB) between two same-dimension encoded images via ffmpeg.
+ * Infinity means bit-identical pixels. Single source of truth for every
+ * drawElement self-verify comparison (streaming drain + parallel disk path).
+ */
+export async function psnrDb(a: Buffer, b: Buffer): Promise<number> {
+  const dir = await mkdtemp(join(tmpdir(), "hf-de-verify-"));
+  try {
+    const pa = join(dir, "a.jpg");
+    const pb = join(dir, "b.jpg");
+    await Promise.all([writeFile(pa, a), writeFile(pb, b)]);
+    const { stderr } = await execFileP(
+      getFfmpegBinary(),
+      ["-hide_banner", "-i", pa, "-i", pb, "-lavfi", "psnr", "-f", "null", "-"],
+      { maxBuffer: 4 * 1024 * 1024 },
+    );
+    const m = /average:(inf|[\d.]+)/.exec(stderr);
+    if (!m) throw new Error(`psnr parse failed: ${stderr.slice(-300)}`);
+    return m[1] === "inf" ? Infinity : Number(m[1]);
+  } finally {
+    await rm(dir, { recursive: true, force: true }).catch(() => {});
+  }
+}
+
+/**
+ * The drawElement self-verify PSNR floor (dB). HF_DE_VERIFY_MIN_DB overrides,
+ * clamped to [10, 60]; out-of-range or unset falls back to 32 — the threshold
+ * every prior eval used to separate real compositor damage from encoder noise.
+ */
+export function resolveDeVerifyMinDb(): number {
+  const raw = Number(process.env.HF_DE_VERIFY_MIN_DB ?? "32");
+  return Number.isFinite(raw) && raw >= 10 && raw <= 60 ? raw : 32;
+}

packages/producer/src/services/render/stages/captureStreamingStage.ts

@@ -41,11 +41,9 @@
  * into a shared module so the stages can import without reaching back.
  */
 
-import { execFile } from "node:child_process";
-import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { mkdtemp, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
-import { promisify } from "node:util";
 import {
   type BeforeCaptureHook,
   type CaptureOptions,
@@ -64,7 +62,7 @@ import {
   distributeFramesInterleaved,
   executeParallelCapture,
   getCapturePerfSummary,
-  getFfmpegBinary,
+  psnrDb,
   recaptureDrawElementFrameForVerify,
   completeDeferredDrawElementInit,
   initializeSession,
@@ -223,27 +221,9 @@ export type CaptureStreamingStageResult =
       success: false;
     };
 
-const execFileP = promisify(execFile);
-
-/** PSNR (average, dB) between two same-dimension encoded images via ffmpeg. */
-async function psnrDb(a: Buffer, b: Buffer): Promise<number> {
-  const dir = await mkdtemp(join(tmpdir(), "hf-de-verify-"));
-  try {
-    const pa = join(dir, "a.jpg");
-    const pb = join(dir, "b.jpg");
-    await Promise.all([writeFile(pa, a), writeFile(pb, b)]);
-    const { stderr } = await execFileP(
-      getFfmpegBinary(),
-      ["-hide_banner", "-i", pa, "-i", pb, "-lavfi", "psnr", "-f", "null", "-"],
-      { maxBuffer: 4 * 1024 * 1024 },
-    );
-    const m = /average:(inf|[\d.]+)/.exec(stderr);
-    if (!m) throw new Error(`psnr parse failed: ${stderr.slice(-300)}`);
-    return m[1] === "inf" ? Infinity : Number(m[1]);
-  } finally {
-    await rm(dir, { recursive: true, force: true }).catch(() => {});
-  }
-}
+// psnrDb moved to @hyperframes/engine (utils/psnr.ts) so the parallel
+// disk-path verify (parallelCoordinator) and this drain guard share one
+// comparison implementation.
 
 // ── drawElement drain-time safety checks (ungated-release safety net) ──
 // Shared by the sequential worker-encode loop and the interleaved parallel