refactor(render): centralize execution ownership

Author: jrusso1020

packages/producer/src/services/render/cleanup.test.ts

@@ -1,144 +1,7 @@
-/**
- * Tests for the cancel/error-path helpers in `./cleanup.ts`.
- */
+/** Tests for render failure-detail construction. */
 
-import { describe, expect, it, vi } from "vitest";
-import { mkdtempSync, mkdirSync, rmSync, writeFileSync, existsSync } from "node:fs";
-import { tmpdir } from "node:os";
-import { join } from "node:path";
-import type { CaptureSession } from "@hyperframes/engine";
-import type { FileServerHandle } from "../fileServer.js";
-import { buildRenderErrorDetails, cleanupRenderResources, safeCleanup } from "./cleanup.js";
-
-function makeLog() {
-  return { error: vi.fn(), warn: vi.fn(), info: vi.fn(), debug: vi.fn() };
-}
-
-describe("safeCleanup", () => {
-  it("returns normally when the operation succeeds", async () => {
-    const log = makeLog();
-    const op = vi.fn().mockResolvedValue(undefined);
-    await safeCleanup("close x", op, log);
-    expect(op).toHaveBeenCalledOnce();
-    expect(log.debug).not.toHaveBeenCalled();
-  });
-
-  it("swallows thrown errors and logs them at debug", async () => {
-    const log = makeLog();
-    await safeCleanup(
-      "close x",
-      () => {
-        throw new Error("boom");
-      },
-      log,
-    );
-    expect(log.debug).toHaveBeenCalledWith("Cleanup failed (close x)", { error: "boom" });
-  });
-
-  it("swallows async rejections", async () => {
-    const log = makeLog();
-    await safeCleanup("close x", async () => Promise.reject(new Error("async boom")), log);
-    expect(log.debug).toHaveBeenCalledWith("Cleanup failed (close x)", { error: "async boom" });
-  });
-});
-
-describe("cleanupRenderResources", () => {
-  it("closes fileServer, probeSession, then removes workDir (non-debug)", async () => {
-    const log = makeLog();
-    const workDir = mkdtempSync(join(tmpdir(), "cleanup-test-"));
-    writeFileSync(join(workDir, "marker.txt"), "x");
-
-    const order: string[] = [];
-    const fileServer = {
-      close: () => {
-        order.push("fileServer.close");
-      },
-    } as unknown as FileServerHandle;
-    const probeSession = {
-      _markClosed: () => {
-        order.push("probeSession.close");
-      },
-    } as unknown as CaptureSession;
-
-    // closeCaptureSession is the engine helper; the helper itself isn't
-    // mockable per-call without intercepting the module import. Instead
-    // we verify the higher-level invariants: fileServer.close was called,
-    // and the workDir was rmSync'd.
-    await cleanupRenderResources({
-      fileServer,
-      probeSession: null, // skip probe to keep this test focused on the workDir invariant
-      workDir,
-      debug: false,
-      log,
-      label: "cancel",
-    });
-
-    expect(order).toEqual(["fileServer.close"]);
-    expect(existsSync(workDir)).toBe(false);
-    void probeSession; // suppress unused-var (kept as a doc of the surface)
-  });
-
-  it("keeps workDir when debug=true", async () => {
-    const log = makeLog();
-    const workDir = mkdtempSync(join(tmpdir(), "cleanup-debug-"));
-    writeFileSync(join(workDir, "marker.txt"), "x");
-
-    await cleanupRenderResources({
-      fileServer: null,
-      probeSession: null,
-      workDir,
-      debug: true,
-      log,
-      label: "error",
-    });
-
-    expect(existsSync(workDir)).toBe(true);
-    rmSync(workDir, { recursive: true, force: true });
-  });
-
-  it("is a no-op for missing workDir thanks to rmSync force:true", async () => {
-    const log = makeLog();
-    const workDir = join(tmpdir(), `cleanup-missing-${Date.now()}`);
-    expect(existsSync(workDir)).toBe(false);
-
-    await cleanupRenderResources({
-      fileServer: null,
-      probeSession: null,
-      workDir,
-      debug: false,
-      log,
-      label: "error",
-    });
-
-    // No throw; nothing logged at debug for the rmSync step.
-    expect(log.debug).not.toHaveBeenCalled();
-  });
-
-  it("logs (and continues past) a fileServer.close that throws", async () => {
-    const log = makeLog();
-    const workDir = mkdtempSync(join(tmpdir(), "cleanup-throw-"));
-
-    const fileServer = {
-      close: () => {
-        throw new Error("server stuck");
-      },
-    } as unknown as FileServerHandle;
-
-    await cleanupRenderResources({
-      fileServer,
-      probeSession: null,
-      workDir,
-      debug: false,
-      log,
-      label: "error",
-    });
-
-    expect(log.debug).toHaveBeenCalledWith("Cleanup failed (close file server (error))", {
-      error: "server stuck",
-    });
-    expect(existsSync(workDir)).toBe(false);
-  });
-});
+import { describe, expect, it } from "vitest";
+import { buildRenderErrorDetails } from "./cleanup.js";
 
 describe("buildRenderErrorDetails", () => {
   const baseDiagnostics = { videoExtractionFailures: 0, imageDecodeFailures: 0 };
@@ -235,6 +98,3 @@ describe("buildRenderErrorDetails", () => {
     expect(failed.hdrDiagnostics).toEqual({ videoExtractionFailures: 2, imageDecodeFailures: 0 });
   });
 });
-
-// Quiet unused-import warning — these are referenced via type-only paths.
-void mkdirSync;

packages/producer/src/services/render/cleanup.ts

@@ -1,80 +1,11 @@
-/**
- * Sequencer cleanup + error-details helpers shared by the cancel and
- * error paths in `executeRenderJob`.
- */
+/** Error-detail construction shared by render failure paths. */
 
-import { rmSync } from "node:fs";
 import { freemem } from "node:os";
-import {
-  type CaptureSession,
-  type SubTimelineWaitOutcome,
-  closeCaptureSession,
-} from "@hyperframes/engine";
-import type { FileServerHandle } from "../fileServer.js";
-import { defaultLogger, type ProducerLogger } from "../../logger.js";
+import { type SubTimelineWaitOutcome } from "@hyperframes/engine";
 import type { HdrDiagnostics, RenderJob } from "../renderOrchestrator.js";
 import { normalizeErrorMessage } from "../../utils/errorMessage.js";
 import type { RenderObservabilitySummary } from "./observability.js";
 
-/**
- * Wrap a cleanup operation so it never throws, but logs any failure.
- * The sequencer needs to keep tearing down resources even when one of
- * them is stuck (e.g. a `fileServer.close()` hitting a TCP race); a
- * thrown cleanup error would mask the original render failure.
- */
-export async function safeCleanup(
-  label: string,
-  fn: () => Promise<void> | void,
-  log: ProducerLogger = defaultLogger,
-): Promise<void> {
-  try {
-    await fn();
-  } catch (err) {
-    log.debug(`Cleanup failed (${label})`, {
-      error: err instanceof Error ? err.message : String(err),
-    });
-  }
-}
-
-/**
- * Close the file server, close the probe session, and remove the
- * working directory. Each step runs through `safeCleanup` so a stuck
- * resource doesn't mask the original render error.
- */
-export async function cleanupRenderResources(input: {
-  fileServer: FileServerHandle | null;
-  probeSession: CaptureSession | null;
-  workDir: string;
-  debug: boolean;
-  log: ProducerLogger;
-  /** Suffix appended to safeCleanup labels. Pinned to the existing diagnostic payloads. */
-  label: "cancel" | "error";
-}): Promise<void> {
-  const { fileServer, probeSession, workDir, debug, log, label } = input;
-  if (fileServer) {
-    const fs = fileServer;
-    await safeCleanup(
-      `close file server (${label})`,
-      () => {
-        fs.close();
-      },
-      log,
-    );
-  }
-  if (probeSession) {
-    const session = probeSession;
-    await safeCleanup(`close probe session (${label})`, () => closeCaptureSession(session), log);
-  }
-  if (!debug) {
-    // `force: true` swallows ENOENT, so no need to existsSync first.
-    await safeCleanup(
-      `remove workDir (${label})`,
-      () => rmSync(workDir, { recursive: true, force: true, maxRetries: 3, retryDelay: 100 }),
-      log,
-    );
-  }
-}
-
 /**
  * Build the `RenderJob.errorDetails` shape downstream consumers (SSE,
  * sync `/render` response, queue introspection) read on failure.

packages/producer/src/services/render/renderExecutionContext.test.ts

@@ -0,0 +1,137 @@
+import { describe, expect, it, vi } from "vitest";
+import type { ProducerLogger } from "../../logger.js";
+import type { RenderJob } from "../renderOrchestrator.js";
+import { RenderExecutionContext } from "./renderExecutionContext.js";
+
+function logger(): ProducerLogger {
+  return { error: vi.fn(), warn: vi.fn(), info: vi.fn(), debug: vi.fn() };
+}
+
+function job(): RenderJob {
+  return {
+    id: "job-1",
+    status: "queued",
+    outcome: null,
+    warnings: [],
+    progress: 0,
+    currentStage: "queued",
+    config: { fps: { num: 30, den: 1 }, quality: "standard" },
+    createdAt: new Date(),
+  };
+}
+
+describe("RenderExecutionContext", () => {
+  it("freezes request identity and scopes every log record", () => {
+    const base = logger();
+    const context = new RenderExecutionContext({
+      request: { renderJobId: "job-1", projectDir: "/project", outputPath: "/out.mp4" },
+      logger: base,
+    });
+
+    expect(Object.isFrozen(context.request)).toBe(true);
+    context.logger.info("started", { phase: "compile" });
+    expect(base.info).toHaveBeenCalledWith("started", {
+      renderJobId: "job-1",
+      phase: "compile",
+    });
+  });
+
+  it("runs disposers once in reverse acquisition order and contains failures", async () => {
+    const base = logger();
+    const calls: string[] = [];
+    const context = new RenderExecutionContext({
+      request: { renderJobId: "job-1", projectDir: "/project", outputPath: "/out.mp4" },
+      logger: base,
+    });
+    context.defer("first", () => calls.push("first"));
+    context.defer("broken", () => {
+      calls.push("broken");
+      throw new Error("close failed");
+    });
+    context.defer("last", async () => calls.push("last"));
+
+    await Promise.all([context.dispose(), context.dispose()]);
+
+    expect(calls).toEqual(["last", "broken", "first"]);
+    expect(base.debug).toHaveBeenCalledWith(
+      "Cleanup failed (broken)",
+      expect.objectContaining({ renderJobId: "job-1", error: "close failed" }),
+    );
+  });
+
+  it("can release ownership after a resource is closed early", async () => {
+    const dispose = vi.fn();
+    const context = new RenderExecutionContext({
+      request: { renderJobId: "job-1", projectDir: "/project", outputPath: "/out.mp4" },
+      logger: logger(),
+    });
+    const release = context.defer("resource", dispose);
+    release();
+
+    await context.dispose();
+
+    expect(dispose).not.toHaveBeenCalled();
+  });
+
+  it("serializes progress before disposal completes", async () => {
+    const deliveries: string[] = [];
+    const context = new RenderExecutionContext({
+      request: { renderJobId: "job-1", projectDir: "/project", outputPath: "/out.mp4" },
+      logger: logger(),
+      progressSink: async (_job, message) => {
+        await Promise.resolve();
+        deliveries.push(message);
+      },
+    });
+    context.defer("resource", () => deliveries.push("disposed"));
+    const renderJob = job();
+    context.onProgress?.(renderJob, "first");
+    context.onProgress?.(renderJob, "second");
+
+    await context.dispose();
+
+    expect(deliveries).toEqual(["first", "second", "disposed"]);
+  });
+
+  it("does not let a broken cleanup logger reject disposal", async () => {
+    const base = logger();
+    vi.mocked(base.debug).mockImplementation(() => {
+      throw new Error("logger unavailable");
+    });
+    const context = new RenderExecutionContext({
+      request: { renderJobId: "job-1", projectDir: "/project", outputPath: "/out.mp4" },
+      logger: base,
+    });
+    context.defer("broken", () => {
+      throw new Error("close failed");
+    });
+
+    await expect(context.dispose()).resolves.toBeUndefined();
+  });
+
+  it("composes caller cancellation with a deadline", async () => {
+    const controller = new AbortController();
+    const context = new RenderExecutionContext({
+      request: { renderJobId: "job-1", projectDir: "/project", outputPath: "/out.mp4" },
+      logger: logger(),
+      signal: controller.signal,
+      deadlineAtMs: Date.now() + 60_000,
+    });
+    controller.abort();
+
+    expect(() => context.assertActive()).toThrow("render_cancelled");
+    await context.dispose();
+  });
+
+  it("is already aborted when its deadline has elapsed", async () => {
+    const context = new RenderExecutionContext({
+      request: { renderJobId: "job-1", projectDir: "/project", outputPath: "/out.mp4" },
+      logger: logger(),
+      deadlineAtMs: Date.now() - 1,
+    });
+
+    expect(context.signal?.aborted).toBe(true);
+    expect(() => context.assertActive()).toThrow("render_cancelled");
+    await context.dispose();
+  });
+});