Skip to content

refactor(render): centralize execution ownership #2158

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jrusso1020 wants to merge 1 commit into
base: 07-10-refactor_core_unify_composition_contract
Choose a base branch
from
Open

refactor(render): centralize execution ownership #2158

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
146 changes: 3 additions & 143 deletions packages/producer/src/services/render/cleanup.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 };
Expand Down Expand Up @@ -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;
73 changes: 2 additions & 71 deletions packages/producer/src/services/render/cleanup.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
Expand Down
137 changes: 137 additions & 0 deletions packages/producer/src/services/render/renderExecutionContext.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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();
});
});
Loading
Loading