import { and, asc, eq, inArray, isNotNull, isNull, lt } from "drizzle-orm"; import { db } from "../db"; import { printers, printJobs } from "../db/schema"; import { getDownloadUrl, headObject } from "../s3"; import { isGatewayOnline, sendPrinterCommand } from "../routes/ws/edge"; import { nextJobStatus } from "./job_status"; import { dispatchEnabled, dispatchSwitchWarning } from "./enabled"; /** * The job scheduler: picks up `queued` jobs and puts them on idle printers. * * Until now `POST /jobs` ended at a `queued` row and an upload URL, and * nothing ever moved it. This is the join - deliberately the simplest thing * that is correct, because every interesting failure here is a physical one. * * ## Why a timer rather than an event * * A job becomes dispatchable for two unrelated reasons: someone queued it, or * a printer became free. The second has no event the backend owns - it * arrives as a telemetry snapshot from a gateway, five seconds after the fact * at best. A timer catches both without needing a nudge from either, and a * scheduler that only ran on job creation would leave a queue stuck forever * behind one busy printer. * * `POST /jobs` nudges it anyway, so a queued job on an idle farm starts in * well under a second rather than waiting out the tick. * * ## What it will not do * * It sends to a printer whose *last reported* status is `idle`, which is a * claim about several seconds ago. Two things follow. A printer someone * started by hand at the machine can still be sent a job, and the gateway's * own guard is what refuses it - `start_uploaded_file` checks the printer is * in a terminal state first. And a job is only ever dispatched once, because * `claimNextJob` moves the row out of `queued` in the same statement that * selects it; two ticks racing cannot both win. */ /** How often to look for dispatchable work. */ const TICK_MS = 5_000; /** * How long the gateway gets to download the file from R2 and transfer it to * the printer before the presigned URL stops working. * * Generous on purpose. The FTPS transfer to the printer alone was measured at * 0.2 MiB/s on a real P1P, so a 62 MB project - which is the size of the * largest file sitting on that printer's card - is over five minutes of * upload before a byte of printing happens. A URL that expires mid-transfer * would fail the job for a reason that looks like a storage outage. */ const DOWNLOAD_URL_TTL_S = 1800; /** * Printer states that may be sent work. * * `idle` only. `offline` is the trap worth naming: a printer the gateway has * not heard from keeps whatever status it last had, so `offline` can mean * "powered off mid-print", and a printer in `error` may be sitting in a * paused state with filament loaded. Neither is a machine to put a job on. */ const DISPATCHABLE_PRINTER_STATUS = ["idle"] as const; /** * How long a job may sit in `uploading` before it is given up on. * * `uploading` means the gateway has accepted the start and is downloading the * file and transferring it to the printer. That reply is deliberately an * "accepted, working on it" - the gateway answers immediately because * awaiting the transfer inline would stop its telemetry for minutes, and * telemetry is what every job's state is derived from. * * The consequence is that a failure *after* the ack has no `CommandResult` to * carry it, so a job whose download 403s on an expired URL, or whose archive * the printer refuses, would otherwise sit in `uploading` forever claiming to * be nearly printing. * * Twenty minutes: the transfer alone is over five for the largest project * seen (62 MB at the measured 0.2 MiB/s), and the printer then spends minutes * in bed levelling before it reports `RUNNING` - a P1P sat at layer 0 for * several while `percent` climbed to 14. Anything tighter would reap jobs * that are working perfectly. */ const UPLOADING_TIMEOUT_MS = 20 * 60_000; /** * Takes one queued job for one printer, atomically. * * The `where` repeats `status = 'queued'`, which is what makes this safe: an * `UPDATE ... WHERE status = 'queued'` that returns a row is proof this * caller is the one that moved it. Selecting and then updating would let two * ticks - or a tick and a nudge - both believe they owned the same job and * send two `start_print` commands for it. */ async function claimNextJob(farmId: string, printerId: string) { const next = await db.query.printJobs.findFirst({ columns: { id: true }, where: and( eq(printJobs.farmId, farmId), eq(printJobs.status, "queued"), // A job that already names a printer was aimed at one deliberately; // this scheduler only places the unassigned ones. Honouring a request // for a *specific* printer is a separate feature and wants its own // rule for what happens when that printer is busy. isNull(printJobs.printerId), ), // Oldest first. Not a priority system - just the absence of one, made // explicit so the order is not whatever Postgres felt like returning. orderBy: [asc(printJobs.createdAt)], }); if (!next) return undefined; const [claimed] = await db .update(printJobs) .set({ printerId, status: "uploading", startedAt: new Date(), updatedAt: new Date() }) .where(and(eq(printJobs.id, next.id), eq(printJobs.status, "queued"))) .returning(); return claimed; } /** Puts a job back for another tick, or fails it outright. */ async function releaseJob(jobId: string, outcome: "requeue" | "fail", reason: string) { const metadataNote = { lastDispatchError: reason, lastDispatchErrorAt: new Date().toISOString() }; if (outcome === "requeue") { await db .update(printJobs) .set({ status: "queued", printerId: null, startedAt: null, updatedAt: new Date(), metadata: metadataNote, }) .where(eq(printJobs.id, jobId)); return; } await db .update(printJobs) .set({ status: "failed", completedAt: new Date(), updatedAt: new Date(), metadata: metadataNote }) .where(eq(printJobs.id, jobId)); } /** * One pass: for every farm with a connected gateway and an idle printer, try * to place a job. * * Returns how many jobs it dispatched, which is what the tests assert on - * a scheduler that silently does nothing is the failure mode to catch. */ export { nextJobStatus } from "./job_status"; export async function dispatchTick(): Promise { // Also covers the nudge from `POST /jobs`, which calls this directly. if (!dispatchEnabled()) return 0; await reapStalledUploads(); let dispatched = 0; // Idle, claimed printers. The `gateway_printer_id` filter is not redundant // with the status check: an unclaimed printer has no id to send a command // to, and `sendPrinterCommand` would throw for every one of them. const candidates = await db.query.printers.findMany({ columns: { id: true, farmId: true, name: true }, where: and( inArray(printers.status, [...DISPATCHABLE_PRINTER_STATUS]), // Only printers a gateway has claimed. Not redundant with the status // check: an unclaimed printer has no gateway id to address, and // `sendPrinterCommand` throws for every one of them. isNotNull(printers.gatewayPrinterId), ), }); for (const printer of candidates) { // Cheap and in-memory, so it goes before the queue query rather than // after: a farm whose gateway is offline has nothing dispatchable no // matter what is queued, and asking the database first would be a // round trip per printer to find that out. if (!isGatewayOnline(printer.farmId)) continue; const job = await claimNextJob(printer.farmId, printer.id); if (!job) continue; try { // Is the file even there? `POST /jobs` hands the client a presigned PUT // and returns, so the row exists before the object does. Asking the // bucket is the only honest answer: a client that crashed mid-upload // would never tell us, and dispatching anyway would have the gateway // download R2's 403 XML error document and hand it to a printer as a // project. A job whose file never arrives just stays queued. const object = await headObject(job.r2ObjectKey); if (!object) { await releaseJob(job.id, "requeue", "the job file is not in object storage yet"); continue; } if (object.size === 0) { // Zero bytes is a finished upload of nothing, not an upload in // progress, and it will never become valid on its own. await releaseJob(job.id, "fail", "the job file is empty"); console.warn(`[dispatch] job ${job.id} failed: uploaded file is 0 bytes`); continue; } // Presigned here, at dispatch, and never at job creation: a queued job // may wait days, and a URL minted when it was queued would be long // expired by the time anything sent it. const url = await getDownloadUrl(job.r2ObjectKey, DOWNLOAD_URL_TTL_S); const result = await sendPrinterCommand(printer.id, { action: "start_print", job_id: job.id, file_name: job.fileName, url, // Chosen by whoever queued the job; absent means "by the project's filament". ...(typeof job.metadata?.amsSlot === "number" ? { ams_slot: job.metadata.amsSlot } : {}), }); if (!result.ok) { // Requeued, not failed. The gateway refusing a start is usually // transient from the cloud's point of view - the printer was busier // than its last snapshot said, or the gateway dropped mid-command - // and a job that fails permanently because of a five-second-stale // status reading would be indefensible. await releaseJob(job.id, "requeue", result.error ?? "gateway reported the start failed"); console.warn(`[dispatch] job ${job.id} requeued: ${result.error}`); continue; } // Note what this does *not* do: it does not mark the job `printing`. // The gateway accepting a start is three steps away from a running // print - confirmed the expensive way on 2026-10-15, when a real print // was accepted, uploaded byte-perfect, reached RUNNING and then paused // itself on a physical filament fault. The job stays `uploading` until // telemetry from the printer says otherwise. See `advanceJobsFrom`. dispatched += 1; console.log(`[dispatch] job ${job.id} sent to printer ${printer.name} (${printer.id})`); } catch (err) { // `sendPrinterCommand` throws only for things that were true before // anything was sent: no such printer, unclaimed, or no gateway. All // three are transient or fixable, so the job goes back in the queue. const reason = err instanceof Error ? err.message : String(err); await releaseJob(job.id, "requeue", reason); console.warn(`[dispatch] job ${job.id} requeued before sending: ${reason}`); } } return dispatched; } /** * Fails jobs that have been `uploading` too long. * * Marked `failed` rather than requeued, deliberately. A requeue would send * the same file to the same printer and, if the cause was the file rather * than a hiccup, do it again every twenty minutes forever. More importantly * the printer may by now be *printing* it - the gateway could have succeeded * and only its telemetry be late - and requeueing would then dispatch a * second job onto a working machine. Failing is the recoverable mistake * here; a human can requeue, and the printer's own telemetry will still * report what it is really doing. */ async function reapStalledUploads(): Promise { const cutoff = new Date(Date.now() - UPLOADING_TIMEOUT_MS); const stalled = await db .update(printJobs) .set({ status: "failed", completedAt: new Date(), updatedAt: new Date(), metadata: { lastDispatchError: `the gateway accepted the start but the printer never reported printing within ` + `${Math.round(UPLOADING_TIMEOUT_MS / 60_000)} minutes - check the gateway's log for ` + `the download or transfer error, which has no path back to the cloud yet`, lastDispatchErrorAt: new Date().toISOString(), }, }) .where(and(eq(printJobs.status, "uploading"), lt(printJobs.startedAt, cutoff))) .returning({ id: printJobs.id }); for (const job of stalled) { console.warn(`[dispatch] job ${job.id} failed: stalled in uploading`); } } let timer: ReturnType | undefined; /** Starts the scheduler. Idempotent, so a second call does not double-tick. */ export function startDispatcher() { if (timer) return; const warning = dispatchSwitchWarning(); if (warning) console.warn(`[dispatch] ${warning}`); if (!dispatchEnabled()) { console.log("[dispatch] DISABLED (DISPATCH_ENABLED=false): queued jobs will not be sent to printers"); return; } timer = setInterval(() => { dispatchTick().catch((err) => console.error("[dispatch] tick failed", err)); }, TICK_MS); // Node/Bun keep the process alive for an interval; this one should not be // the reason a server refuses to exit. timer.unref?.(); console.log(`[dispatch] scheduler started, every ${TICK_MS}ms`); } export function stopDispatcher() { if (timer) clearInterval(timer); timer = undefined; } /** Runs a tick now, without waiting for the next one. */ export function nudgeDispatcher() { dispatchTick().catch((err) => console.error("[dispatch] nudge failed", err)); } /** * Moves job rows along from what the printers are actually reporting. * * This is the half that matters, and the reason the dispatcher above refuses * to mark anything `printing`. A gateway accepting `start_print` means it * accepted the command - no more. Confirmed the expensive way on a real P1P * (2026-10-15): the command was accepted, the file uploaded byte-perfect, the * printer reached `RUNNING`, and then it paused itself on a physical * broken-filament fault. A job marked `printing` on the gateway's `ok` would * have sat there claiming to print for the rest of the day. * * So job state is derived from the telemetry that follows, and the mapping is * narrow on purpose: * * - `printing` while the printer reports it is printing. * - `paused` likewise - including the printer pausing *itself* on a fault, * which is what that P1P did and which an operator needs to see. * - `completed` when the printer goes idle having been printing. A Bambu * reports a finished print as `FINISH`, which `mapStatus` narrows to * `idle`, so "was printing, now idle" is the completion signal. * - `failed` when the printer reports an error. Note a Bambu reports an * operator-cancelled print as `FAILED` too, so this cannot tell a * cancellation from a fault; `cancelled` is set by whoever sends the stop, * and is left alone here precisely so it is not overwritten by the `FAILED` * that follows it. * * Progress is copied across whenever it is reported, because a job row that * says 0% through a two-hour print is worse than useless to an operator. */ export async function advanceJobsFrom( printerId: string, status: "idle" | "printing" | "paused" | "error" | undefined, percent: number | null | undefined, ): Promise { if (!status) return; // The one job this printer could be running. `uploading` is included // because that is the state the dispatcher leaves a job in, and the first // telemetry that says `printing` is what promotes it. const job = await db.query.printJobs.findFirst({ columns: { id: true, status: true }, where: and( eq(printJobs.printerId, printerId), inArray(printJobs.status, ["uploading", "printing", "paused"]), ), orderBy: [asc(printJobs.createdAt)], }); if (!job) return; // Narrowed explicitly rather than cast: the query's `where` already limits // this to the three, so a value outside them means the query and this call // have drifted apart - which is exactly the bug a cast would hide. if (job.status !== "uploading" && job.status !== "printing" && job.status !== "paused") return; const next = nextJobStatus(job.status, status); const terminal = next === "completed" || next === "failed"; // Progress alone still deserves a write, so an unchanged status does not // freeze the percentage. if (!next && percent == null) return; await db .update(printJobs) .set({ ...(next ? { status: next } : {}), ...(percent != null ? { progressPct: percent } : {}), ...(terminal ? { completedAt: new Date(), progressPct: next === "completed" ? 100 : undefined } : {}), updatedAt: new Date(), }) .where(eq(printJobs.id, job.id)); if (next && next !== job.status) { console.log(`[dispatch] job ${job.id} ${job.status} -> ${next} (printer reported ${status})`); } }