feat(pipeline): render queue, worker service and automated preview generation

One "Queue Export" click now renders the EXR sequence, then rebuilds the shot headlessly with the studio slate/overlay template to produce the delivery MOV and review MP4. Implements RenderPipeline2 phases 1-2 plus the preview stage. Server: - New models Export, RenderJob, ExportEvent, Machine, WorkerHeartbeat, plus Project.deliveryConfig and per-submission slate fields (Export.vfxScope, Export.submissionNote, inherited from the shot's previous export). Both migrations are purely additive; no existing column is touched. - lib/render-pipeline: server-enforced state machine, transactional version increment with supersede, atomic FOR UPDATE SKIP LOCKED claim gated by machine availability windows, and a lease reaper run from instrumentation.ts. - /api/ext/* endpoints for the panel and workers; session-auth mirrors under /api/render and /api/machines for the web UI. - Pipeline pages: render queue, export detail, machine monitoring, plus an Exports tab on shot detail. RenderWorker (.NET 8 Windows service, new): - Registration, heartbeat as cancel channel, claim loop, aerender runner with progress parsing and stall watchdog, crash recovery and disk-spooled reporting that survives server downtime. - Preview stage: headless AE assembles the preview comp into a throwaway AEP with both output modules queued, then a single aerender pass renders them. Preview jobs are not claimed while an interactive AE session is open, so an artist's project is never taken over. AE panel: Queue Export with live status polling, urgent flag, retry, and the VFX Scope / Submission Note fields. Every existing panel action is unchanged. Preview chaining ships disabled behind SystemConfig preview.enabled. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-02 14:34:34 +02:00
parent 6b15bae62a
commit cc89415a29
71 changed files with 11767 additions and 1 deletions
@@ -0,0 +1,504 @@
+import { ExportStatus, Prisma } from "@prisma/client";
+import { db } from "@/lib/db";
+import { PipelineError } from "./errors";
+import { renderManifestSchema, type RenderManifest } from "./manifest";
+import { applyTransition, type TransitionActor } from "./transitions";
+import { getConfigNumber, getConfigValue } from "./config";
+
+/** Statuses in which a shot has a live export in flight (§6.4 E1 409 rule). */
+const ACTIVE_STATUSES: ExportStatus[] = [
+  "QUEUED",
+  "RENDERING",
+  "VALIDATING",
+  "GENERATING_PREVIEW",
+];
+
+/** Non-terminal statuses auto-superseded when a newer Export is queued (T1/T17). */
+const SUPERSEDABLE_STATUSES: ExportStatus[] = [
+  "QUEUED",
+  "RENDER_FAILED",
+  "VALIDATION_FAILED",
+  "PREVIEW_FAILED",
+  "READY_FOR_QC",
+  "QC_FAILED",
+  "READY_FOR_DELIVERY",
+];
+
+const OPEN_JOB_STATUSES = ["QUEUED", "CLAIMED", "RUNNING"] as const;
+
+export function parseVersionString(v: string | null | undefined): number {
+  const m = /^v(\d+)$/i.exec(v?.trim() ?? "");
+  return m ? parseInt(m[1], 10) : 0;
+}
+
+export function formatVersionString(n: number): string {
+  return `v${String(n).padStart(3, "0")}`;
+}
+
+/** `UNG_..._cmp_TT_v003` + 4 → `UNG_..._cmp_TT_v004`; falls back to house convention. */
+function nextExrOutputBase(current: string | null, shotCode: string, versionString: string): string {
+  if (current && /_v\d+$/i.test(current)) return current.replace(/_v\d+$/i, `_${versionString}`);
+  if (current) return `${current}_${versionString}`;
+  return renderOutputBase(shotCode, versionString);
+}
+
+/**
+ * Rendered EXR naming for the pipeline — matches the panel's
+ * "Queue EXR (Review)" convention (`{shotCode}_cmp_TT_{version}`), which is
+ * shot-code based and deliberately excludes the source clip name that
+ * `Shot.exrOutput` carries from the EDL import.
+ */
+function renderOutputBase(shotCode: string, versionString: string): string {
+  return `${shotCode}_cmp_TT_${versionString}`;
+}
+
+export interface CreateExportInput {
+  manifest: unknown;
+  submittedByEmail?: string | null;
+  priority?: number;
+  force?: boolean;
+  /**
+   * Per-submission slate fields. `undefined` inherits the previous export's
+   * value for this shot; `null` or "" explicitly clears it.
+   */
+  vfxScope?: string | null;
+  submissionNote?: string | null;
+}
+
+/**
+ * E1 — Queue Export (§6.4). Transactionally: decides the new version number
+ * (server-side, race-free), updates Shot.shotVersion/exrOutput exactly as the
+ * panel's legacy PATCH did, supersedes older non-terminal exports, creates
+ * Export(QUEUED) + RenderJob attempt 1 with the manifest snapshot.
+ */
+export async function createExport(input: CreateExportInput) {
+  const parsed = renderManifestSchema.safeParse(input.manifest);
+  if (!parsed.success) {
+    throw new PipelineError(422, "Manifest validation error", parsed.error.issues);
+  }
+  const manifest = parsed.data;
+
+  const shot = await db.shot.findUnique({
+    where: { id: manifest.shotId },
+    include: {
+      project: { select: { id: true, code: true, showId: true } },
+      tasks: { where: { type: "COMP" }, orderBy: { sortOrder: "asc" }, take: 1, select: { id: true } },
+    },
+  });
+  if (!shot) throw new PipelineError(404, `Shot not found: ${manifest.shotId}`);
+
+  // Cross-checks (§6.2): manifest vs DB truth
+  if (shot.shotCode !== manifest.shotCode) {
+    throw new PipelineError(422, `Manifest shotCode "${manifest.shotCode}" does not match shot ${shot.id} ("${shot.shotCode}")`);
+  }
+  if (shot.project.code !== manifest.projectCode) {
+    throw new PipelineError(422, `Manifest projectCode "${manifest.projectCode}" does not match shot's project ("${shot.project.code}")`);
+  }
+  // A 0/0 shot range means "never imported", not a real single-frame shot at 0
+  const shotRangeKnown =
+    shot.frameStart != null && shot.frameEnd != null && !(shot.frameStart === 0 && shot.frameEnd === 0);
+  if (
+    !input.force &&
+    shotRangeKnown &&
+    (shot.frameStart !== manifest.frameStart || shot.frameEnd !== manifest.frameEnd)
+  ) {
+    throw new PipelineError(422, `Manifest frame range ${manifest.frameStart}-${manifest.frameEnd} does not match shot record ${shot.frameStart}-${shot.frameEnd} (pass force to override)`);
+  }
+
+  const activeExport = await db.export.findFirst({
+    where: { shotId: shot.id, status: { in: ACTIVE_STATUSES } },
+    select: { id: true, status: true, versionString: true },
+  });
+  if (activeExport && !input.force) {
+    throw new PipelineError(
+      409,
+      `An active export (${activeExport.versionString}, ${activeExport.status}) already exists for this shot — pass force to supersede it`,
+      { activeExportId: activeExport.id }
+    );
+  }
+
+  const submittedBy = input.submittedByEmail
+    ? await db.user.findUnique({ where: { email: input.submittedByEmail }, select: { id: true, name: true } })
+    : null;
+
+  const [maxAttempts, outputRoot] = await Promise.all([
+    getConfigNumber("render.maxAttempts"),
+    getConfigValue("render.outputRoot"),
+  ]);
+
+  // Retry loop: @@unique([shotId, versionNumber]) guards the version-increment
+  // race between two concurrent E1 calls; loser recomputes and retries.
+  for (let tryNo = 0; tryNo < 3; tryNo++) {
+    try {
+      return await db.$transaction(async (tx) => {
+        const latest = await tx.export.aggregate({
+          where: { shotId: shot.id },
+          _max: { versionNumber: true },
+        });
+        const freshShot = await tx.shot.findUniqueOrThrow({
+          where: { id: shot.id },
+          select: { shotVersion: true, exrOutput: true },
+        });
+        // Slate fields carry forward from the shot's previous submission
+        // unless the caller supplies new ones.
+        const previous = await tx.export.findFirst({
+          where: { shotId: shot.id },
+          orderBy: { versionNumber: "desc" },
+          select: { vfxScope: true, submissionNote: true },
+        });
+        const vfxScope = input.vfxScope !== undefined ? input.vfxScope : (previous?.vfxScope ?? null);
+        const submissionNote =
+          input.submissionNote !== undefined ? input.submissionNote : (previous?.submissionNote ?? null);
+        const versionNumber =
+          Math.max(latest._max.versionNumber ?? 0, parseVersionString(freshShot.shotVersion)) + 1;
+        const versionString = formatVersionString(versionNumber);
+        const exrBase = nextExrOutputBase(freshShot.exrOutput, shot.shotCode, versionString);
+
+        const outputDir =
+          manifest.outputDir === "auto"
+            ? [outputRoot, shot.project.code, ...(shot.episode ? [shot.episode] : []), shot.shotCode, versionString].join("/")
+            : manifest.outputDir;
+        const outputPattern =
+          manifest.outputPattern === "auto"
+            ? `${renderOutputBase(shot.shotCode, versionString)}.[#####].exr`
+            : manifest.outputPattern;
+
+        const resolvedManifest: RenderManifest = { ...manifest, outputDir, outputPattern };
+
+        // Supersede older non-terminal exports (T1/T17); cancel their open jobs.
+        const toSupersede = await tx.export.findMany({
+          where: {
+            shotId: shot.id,
+            status: { in: input.force ? [...SUPERSEDABLE_STATUSES, ...ACTIVE_STATUSES] : SUPERSEDABLE_STATUSES },
+          },
+          select: { id: true, status: true },
+        });
+
+        const created = await tx.export.create({
+          data: {
+            shotId: shot.id,
+            projectId: shot.project.id,
+            taskId: shot.tasks[0]?.id ?? null,
+            versionNumber,
+            versionString,
+            status: "QUEUED",
+            aepPath: manifest.aepPath,
+            compName: manifest.compName,
+            rendererType: manifest.rendererType,
+            outputDir,
+            outputPattern,
+            frameStart: manifest.frameStart,
+            frameEnd: manifest.frameEnd,
+            fps: manifest.fps,
+            width: manifest.width,
+            height: manifest.height,
+            colorspace: manifest.expected?.colorspace ?? null,
+            vfxScope,
+            submissionNote,
+            submittedById: submittedBy?.id ?? null,
+            submittedByName: submittedBy?.name ?? input.submittedByEmail ?? null,
+          },
+        });
+
+        for (const old of toSupersede) {
+          await tx.renderJob.updateMany({
+            where: { exportId: old.id, status: { in: [...OPEN_JOB_STATUSES] } },
+            data: { status: "CANCELLED", finishedAt: new Date(), errorMessage: `Superseded by export ${created.id} (${versionString})` },
+          });
+          await applyTransition(
+            tx,
+            old,
+            "SUPERSEDED",
+            { type: "SYSTEM", note: `Superseded by ${versionString}` },
+            { supersededById: created.id }
+          );
+        }
+
+        const renderJob = await tx.renderJob.create({
+          data: {
+            type: "AE_RENDER",
+            exportId: created.id,
+            attempt: 1,
+            maxAttempts,
+            priority: input.priority ?? 50,
+            manifest: resolvedManifest as unknown as Prisma.InputJsonValue,
+          },
+        });
+
+        await tx.exportEvent.create({
+          data: {
+            exportId: created.id,
+            fromStatus: null,
+            toStatus: "QUEUED",
+            actorType: "USER",
+            actorId: submittedBy?.id ?? null,
+            note: `Queue Export ${versionString}${input.force ? " (force)" : ""}`,
+          },
+        });
+
+        // Mirror the legacy panel PATCH: Shot stays the canonical "current version"
+        const updatedShot = await tx.shot.update({
+          where: { id: shot.id },
+          data: { shotVersion: versionString, exrOutput: exrBase },
+          select: { id: true, shotVersion: true, exrOutput: true },
+        });
+
+        return {
+          export: {
+            id: created.id,
+            shotCode: shot.shotCode,
+            versionNumber,
+            versionString,
+            status: created.status,
+            outputDir,
+            outputPattern,
+            vfxScope,
+            submissionNote,
+          },
+          renderJob: { id: renderJob.id, attempt: renderJob.attempt, priority: renderJob.priority },
+          shot: updatedShot,
+          superseded: toSupersede.map((s) => s.id),
+        };
+      });
+    } catch (err) {
+      if (err instanceof Prisma.PrismaClientKnownRequestError && err.code === "P2002" && tryNo < 2) {
+        continue; // concurrent E1 won the version — recompute
+      }
+      throw err;
+    }
+  }
+  throw new PipelineError(409, "Could not allocate a version number after retries");
+}
+
+function serializeExport<T extends { exrTotalBytes: bigint | null }>(e: T) {
+  return { ...e, exrTotalBytes: e.exrTotalBytes == null ? null : Number(e.exrTotalBytes) };
+}
+
+/** E2 — latest export + status for the AE panel header (§6.4). */
+export async function getLatestExport(shotCode: string, projectCode?: string | null) {
+  const shot = await db.shot.findFirst({
+    where: { shotCode, ...(projectCode ? { project: { code: projectCode } } : {}) },
+    select: { id: true },
+  });
+  if (!shot) throw new PipelineError(404, "Shot not found");
+
+  const exp = await db.export.findFirst({
+    where: { shotId: shot.id },
+    orderBy: { versionNumber: "desc" },
+    include: {
+      renderJobs: {
+        orderBy: { attempt: "desc" },
+        take: 1,
+        include: { machine: { select: { name: true } } },
+      },
+    },
+  });
+  if (!exp) return { export: null };
+
+  const job = exp.renderJobs[0] ?? null;
+  return {
+    export: {
+      id: exp.id,
+      versionNumber: exp.versionNumber,
+      versionString: exp.versionString,
+      status: exp.status,
+      statusChangedAt: exp.statusChangedAt,
+      outputDir: exp.outputDir,
+      outputPattern: exp.outputPattern,
+      // Panel prefills its slate fields from these
+      vfxScope: exp.vfxScope,
+      submissionNote: exp.submissionNote,
+      renderJob: job
+        ? {
+            id: job.id,
+            attempt: job.attempt,
+            status: job.status,
+            progress: job.progress,
+            currentFrame: job.currentFrame,
+            totalFrames: job.totalFrames,
+            etaSeconds: job.etaSeconds,
+            errorMessage: job.errorMessage,
+            machineName: job.machine?.name ?? null,
+          }
+        : null,
+      previewUrl: exp.previewMovKey ? `/api/files/${exp.previewMovKey}` : null,
+      thumbnailUrl: exp.thumbnailKey ? `/api/files/${exp.thumbnailKey}` : null,
+    },
+  };
+}
+
+/** E3 — export detail incl. attempts and audit events. */
+export async function getExportDetail(exportId: string) {
+  const exp = await db.export.findUnique({
+    where: { id: exportId },
+    include: {
+      shot: { select: { id: true, shotCode: true, episode: true, project: { select: { id: true, name: true, code: true } } } },
+      renderJobs: {
+        orderBy: { attempt: "asc" },
+        include: { machine: { select: { id: true, name: true } } },
+      },
+      events: { orderBy: { createdAt: "asc" } },
+    },
+  });
+  if (!exp) throw new PipelineError(404, "Export not found");
+  return { export: serializeExport(exp) };
+}
+
+/**
+ * E14 / E15 — retry a failed export: new RenderJob attempt, back to QUEUED.
+ * PREVIEW_FAILED is routed to the preview-only retry (T11) so validated EXRs
+ * are never re-rendered just to rebuild a slate.
+ */
+export async function retryExport(exportId: string, actor: TransitionActor) {
+  const current = await db.export.findUnique({ where: { id: exportId }, select: { status: true } });
+  if (!current) throw new PipelineError(404, "Export not found");
+  if (current.status === "PREVIEW_FAILED") {
+    const { retryPreview } = await import("./preview");
+    return retryPreview(exportId, actor.id ?? undefined);
+  }
+
+  return db.$transaction(async (tx) => {
+    const exp = await tx.export.findUnique({
+      where: { id: exportId },
+      include: { renderJobs: { orderBy: { attempt: "desc" }, take: 1 } },
+    });
+    if (!exp) throw new PipelineError(404, "Export not found");
+    if (!["RENDER_FAILED", "VALIDATION_FAILED"].includes(exp.status)) {
+      throw new PipelineError(409, `Export is ${exp.status} — only RENDER_FAILED, VALIDATION_FAILED or PREVIEW_FAILED exports can be retried`);
+    }
+    const last = exp.renderJobs[0];
+    if (!last) throw new PipelineError(500, "Export has no render jobs");
+
+    const job = await tx.renderJob.create({
+      data: {
+        type: last.type,
+        exportId: exp.id,
+        attempt: last.attempt + 1,
+        maxAttempts: last.maxAttempts,
+        priority: last.priority,
+        manifest: last.manifest as Prisma.InputJsonValue,
+      },
+    });
+    const updated = await applyTransition(tx, exp, "QUEUED", {
+      ...actor,
+      note: actor.note ?? `Retry (attempt ${job.attempt})`,
+    });
+    return { export: { id: updated.id, status: updated.status }, renderJob: { id: job.id, attempt: job.attempt } };
+  });
+}
+
+/** E15 / T18 — cancel a queued or rendering export. Worker learns via heartbeat/progress. */
+export async function cancelExport(exportId: string, actor: TransitionActor) {
+  return db.$transaction(async (tx) => {
+    const exp = await tx.export.findUnique({ where: { id: exportId }, select: { id: true, status: true } });
+    if (!exp) throw new PipelineError(404, "Export not found");
+    if (exp.status === "CANCELLED") return { export: exp, alreadyApplied: true };
+    if (!["QUEUED", "RENDERING"].includes(exp.status)) {
+      throw new PipelineError(409, `Export is ${exp.status} — only QUEUED or RENDERING exports can be cancelled`);
+    }
+    await tx.renderJob.updateMany({
+      where: { exportId: exp.id, status: "QUEUED" },
+      data: { status: "CANCELLED", finishedAt: new Date(), errorMessage: "Cancelled by user" },
+    });
+    // Claimed/running jobs stay marked CANCELLED but keep finishedAt null until
+    // the worker acknowledges (heartbeat CANCEL_JOB command / progress cancelRequested)
+    await tx.renderJob.updateMany({
+      where: { exportId: exp.id, status: { in: ["CLAIMED", "RUNNING"] } },
+      data: { status: "CANCELLED", errorMessage: "Cancelled by user" },
+    });
+    const updated = await applyTransition(tx, exp, "CANCELLED", actor);
+    return { export: { id: updated.id, status: updated.status }, alreadyApplied: false };
+  });
+}
+
+/**
+ * Phase 1 stopgap (§17.1): admin "mark as done manually" so the studio can
+ * keep using the old render path while the queue is validated. Removed once
+ * workers render for real.
+ */
+export async function markExportDoneManually(exportId: string, userId: string, note?: string) {
+  return db.$transaction(async (tx) => {
+    const exp = await tx.export.findUnique({ where: { id: exportId }, select: { id: true, status: true } });
+    if (!exp) throw new PipelineError(404, "Export not found");
+    if (!["QUEUED", "RENDERING"].includes(exp.status)) {
+      throw new PipelineError(409, `Export is ${exp.status} — only QUEUED or RENDERING exports can be marked done`);
+    }
+    await tx.renderJob.updateMany({
+      where: { exportId: exp.id, status: { in: [...OPEN_JOB_STATUSES] } },
+      data: { status: "CANCELLED", finishedAt: new Date(), errorMessage: "Manually marked done (legacy render path)" },
+    });
+    const updated = await applyTransition(tx, exp, "READY_FOR_QC", {
+      type: "USER",
+      id: userId,
+      note: note ?? "Manually marked done (rendered outside the pipeline)",
+    });
+    return { export: { id: updated.id, status: updated.status } };
+  });
+}
+
+export interface ListExportsFilter {
+  projectId?: string;
+  episode?: string;
+  status?: ExportStatus;
+  shotId?: string;
+  page?: number;
+  limit?: number;
+}
+
+/** E21 — paginated export list for the web queue/monitoring pages. */
+export async function listExports(filter: ListExportsFilter) {
+  const page = Math.max(1, filter.page ?? 1);
+  const limit = Math.min(200, Math.max(1, filter.limit ?? 50));
+  const where: Prisma.ExportWhereInput = {
+    ...(filter.projectId ? { projectId: filter.projectId } : {}),
+    ...(filter.status ? { status: filter.status } : {}),
+    ...(filter.shotId ? { shotId: filter.shotId } : {}),
+    ...(filter.episode ? { shot: { episode: filter.episode } } : {}),
+  };
+  const [total, rows] = await Promise.all([
+    db.export.count({ where }),
+    db.export.findMany({
+      where,
+      orderBy: { createdAt: "desc" },
+      skip: (page - 1) * limit,
+      take: limit,
+      include: {
+        shot: { select: { shotCode: true, episode: true, project: { select: { name: true, code: true } } } },
+        renderJobs: {
+          orderBy: { attempt: "desc" },
+          take: 1,
+          include: { machine: { select: { name: true } } },
+        },
+      },
+    }),
+  ]);
+  return {
+    pagination: { page, limit, total, pages: Math.ceil(total / limit) },
+    exports: rows.map((e) => ({
+      id: e.id,
+      shotCode: e.shot.shotCode,
+      episode: e.shot.episode,
+      projectName: e.shot.project.name,
+      projectCode: e.shot.project.code,
+      versionString: e.versionString,
+      status: e.status,
+      statusChangedAt: e.statusChangedAt,
+      createdAt: e.createdAt,
+      outputDir: e.outputDir,
+      job: e.renderJobs[0]
+        ? {
+            id: e.renderJobs[0].id,
+            attempt: e.renderJobs[0].attempt,
+            status: e.renderJobs[0].status,
+            progress: e.renderJobs[0].progress,
+            currentFrame: e.renderJobs[0].currentFrame,
+            totalFrames: e.renderJobs[0].totalFrames,
+            etaSeconds: e.renderJobs[0].etaSeconds,
+            priority: e.renderJobs[0].priority,
+            machineName: e.renderJobs[0].machine?.name ?? null,
+            errorMessage: e.renderJobs[0].errorMessage,
+          }
+        : null,
+    })),
+  };
+}