vfxreview/lib/render-pipeline/reaper.ts

import { Prisma } from "@prisma/client";
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)" },
        });
        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,
            },
          });
          if (job.export.status === "RENDERING") {
            await applyTransition(tx, job.export, "QUEUED", {
              type: "SYSTEM",
              note: `Lease expired on attempt ${job.attempt} — requeued as attempt ${next.attempt}`,
            });
          }
        } else if (job.export.status === "RENDERING") {
          await applyTransition(tx, job.export, "RENDER_FAILED", {
            type: "SYSTEM",
            note: `Lease expired on attempt ${job.attempt}/${job.maxAttempts} — 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) } },
  });
}

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

/** 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)");
}