vfxreview/lib/render-pipeline/reaper.ts

import { Prisma } from "@prisma/client";
import { pruneHighResSweep } from "./high-res";
import { db } from "@/lib/db";
import { applyTransition } from "./transitions";
import { getConfigNumber } from "./config";

/**
 * Server-side reaper (RenderPipeline2 §7.7) — the only background job the web
 * app gains. Covers worker power loss, BSOD and network partition with no
 * worker cooperation:
 *   1. expired leases → job EXPIRED; requeue (T5) or Export → RENDER_FAILED (T4)
 *   2. silent machines → OFFLINE
 *   3. prune heartbeats older than 7 days
 */
export async function reapExpiredLeases() {
  const expired = await db.renderJob.findMany({
    where: { status: { in: ["CLAIMED", "RUNNING"] }, leaseExpiresAt: { lt: new Date() } },
    include: { export: { select: { id: true, status: true } } },
  });

  for (const job of expired) {
    try {
      await db.$transaction(async (tx) => {
        // Re-check inside the transaction — the worker may have reported in between
        const fresh = await tx.renderJob.findUnique({ where: { id: job.id }, select: { status: true, leaseExpiresAt: true } });
        if (!fresh || !["CLAIMED", "RUNNING"].includes(fresh.status) || !fresh.leaseExpiresAt || fresh.leaseExpiresAt >= new Date()) {
          return;
        }
        await tx.renderJob.update({
          where: { id: job.id },
          data: { status: "EXPIRED", finishedAt: new Date(), errorMessage: "Lease expired (worker unreachable)" },
        });

        // A delivery build that died leaves a half-written folder, so it is
        // failed rather than silently retried: someone should look at what
        // landed on the delivery volume before it is rebuilt.
        if (job.deliveryId) {
          await tx.deliveryPackage.updateMany({
            where: { id: job.deliveryId, status: "BUILDING" },
            data: {
              status: "FAILED",
              errorMessage: "Worker became unreachable mid-build — check the package folder before rebuilding",
            },
          });
          return;
        }

        if (!job.export) return;

        if (job.attempt < job.maxAttempts) {
          const next = await tx.renderJob.create({
            data: {
              type: job.type,
              exportId: job.exportId,
              attempt: job.attempt + 1,
              maxAttempts: job.maxAttempts,
              priority: job.priority,
              manifest: job.manifest as Prisma.InputJsonValue,
            },
          });
          // VALIDATING counts too: the worker holds the same job through the
          // validation stage, so dying there means the render must be redone —
          // nothing was established about the EXRs either way.
          if (["RENDERING", "VALIDATING"].includes(job.export.status)) {
            await applyTransition(tx, job.export, "QUEUED", {
              type: "SYSTEM",
              note: `Lease expired on attempt ${job.attempt} during ${job.export.status} — requeued as attempt ${next.attempt}`,
            });
          }
        } else if (["RENDERING", "VALIDATING"].includes(job.export.status)) {
          await applyTransition(tx, job.export, "RENDER_FAILED", {
            type: "SYSTEM",
            note: `Lease expired on attempt ${job.attempt}/${job.maxAttempts} during ${job.export.status} — no retries remain`,
          });
        }
      });
    } catch (err) {
      console.error(`[render-pipeline] reaper failed for job ${job.id}`, err);
    }
  }
  return expired.length;
}

export async function reaperSweep() {
  const heartbeatSeconds = await getConfigNumber("render.heartbeatSeconds");

  await reapExpiredLeases();

  await db.machine.updateMany({
    where: {
      status: "ONLINE",
      OR: [{ lastSeenAt: null }, { lastSeenAt: { lt: new Date(Date.now() - heartbeatSeconds * 3 * 1000) } }],
    },
    data: { status: "OFFLINE" },
  });

  await db.workerHeartbeat.deleteMany({
    where: { createdAt: { lt: new Date(Date.now() - 7 * 24 * 3600 * 1000) } },
  });

  // High-quality media housekeeping: hourly is plenty, and it can touch
  // object storage, so it stays off the once-a-minute lease path.
  if (Date.now() - (globalForReaper.lastHighResSweep ?? 0) >= HIGH_RES_SWEEP_MS) {
    globalForReaper.lastHighResSweep = Date.now();
    await pruneHighResSweep().catch((err) => console.error("[high-res] housekeeping sweep failed", err));
  }
}

const HIGH_RES_SWEEP_MS = 60 * 60 * 1000;

const globalForReaper = globalThis as unknown as {
  renderPipelineReaper?: ReturnType<typeof setInterval>;
  lastHighResSweep?: number;
};

/** Started once from instrumentation.ts (Next.js server startup hook). */
export function startReaper(intervalMs = 60_000) {
  if (globalForReaper.renderPipelineReaper) return;
  globalForReaper.renderPipelineReaper = setInterval(() => {
    reaperSweep().catch((err) => console.error("[render-pipeline] reaper sweep failed", err));
  }, intervalMs);
  // Don't hold the process open on shutdown
  globalForReaper.renderPipelineReaper.unref?.();
  console.log("[render-pipeline] reaper started (interval", intervalMs, "ms)");
}