From 564d01816baf64a874cc6d9baba26328a53c0c72 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Thu, 15 Oct 2026 22:54:44 +0000 Subject: [PATCH] Dispatch queued jobs, and derive their state from the printers `POST /jobs` ended at a `queued` row and an upload URL; nothing moved it. This is the join, plus the two things that join turns out to need. ## The scheduler `src/dispatch` ticks every 5s: for each idle, claimed printer on a farm with a connected gateway, take the oldest unassigned queued job and send `start_print`. A timer rather than an event, because a job becomes dispatchable for two unrelated reasons - someone queued it, or a printer became free - and the second has no event this backend owns. It arrives as a telemetry snapshot, seconds after the fact. A scheduler that only ran on job creation would leave a queue stuck forever behind one busy printer. Claiming is an `UPDATE ... WHERE status = 'queued'` that returns the row, so two ticks cannot both win the same job. Selecting and then updating would let a tick and a nudge each believe they owned it and send two starts. It checks the bucket before dispatching. `POST /jobs` hands back a presigned PUT and returns, so the row exists before the file does; dispatching anyway would have the gateway download R2's 403 error document and give it to a printer as a project. A job whose file never arrives just stays queued, and a zero-byte upload fails outright because it will never become valid. The download URL is presigned at dispatch and not at creation - a queued job may wait days. ## Job state comes from telemetry, not from the gateway's ok This is the part that is not obvious and was learned the expensive way on 2026-10-15: a real print was accepted, uploaded byte-perfect, reached RUNNING, and then the printer paused itself on a physical broken-filament fault. A job marked `printing` on the gateway's reply would have claimed to be printing for the rest of the day. So `advanceJobsFrom` runs on every telemetry snapshot and the mapping lives in `job_status.ts`, apart from the database so every pair can be tested without a Postgres or a printer. The line that matters: an idle printer does *not* complete a job that is still `uploading`. A printer is idle for the entire transfer - minutes of it, at the 0.2 MiB/s measured over FTPS - so reading that as completion would mark every job successful seconds after dispatch, before the file reached the machine. 15 tests cover the table exhaustively. Because the gateway now acks a start immediately (it must - awaiting the transfer would stop its telemetry for minutes, and telemetry is what this depends on), a failure after the ack has no result message to travel on. Hence the reaper: a job `uploading` for 20 minutes is failed, not requeued, because the printer may actually be printing it and a requeue would put a second job on a working machine. ## Unclaimed printers are visible now `persistTelemetry` still refuses to create printer rows for gateway ids it does not recognise - a typo in a `printers.toml` becoming a permanent phantom printer is worse. But the refusal was silent, which made the worst first-run experience in the platform: gateway connects, reports happily, fleet stays empty, and the only evidence is in stderr. The observation is now kept and served by `GET /farms/:id/unclaimed-printers`, with `gatewayOnline` alongside it because an empty list otherwise cannot distinguish "all claimed" from "no gateway ever connected". Claiming is still `POST /printers`, still owner-only. 77 tests pass. --- src/dispatch/index.ts | 383 ++++++++++++++++++++++++++++++++++++ src/dispatch/job_status.ts | 38 ++++ src/index.ts | 6 + src/routes/farms.ts | 42 ++++ src/routes/jobs.ts | 17 +- src/routes/ws/edge.ts | 81 +++++++- src/s3/index.ts | 32 ++- tests/dispatch.test.ts | 72 +++++++ tests/edge_protocol.test.ts | 70 ++++++- 9 files changed, 723 insertions(+), 18 deletions(-) create mode 100644 src/dispatch/index.ts create mode 100644 src/dispatch/job_status.ts create mode 100644 tests/dispatch.test.ts diff --git a/src/dispatch/index.ts b/src/dispatch/index.ts new file mode 100644 index 0000000..f105a58 --- /dev/null +++ b/src/dispatch/index.ts @@ -0,0 +1,383 @@ +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"; + +/** + * 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 { + 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, + }); + + 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; + 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})`); + } +} diff --git a/src/dispatch/job_status.ts b/src/dispatch/job_status.ts new file mode 100644 index 0000000..5966768 --- /dev/null +++ b/src/dispatch/job_status.ts @@ -0,0 +1,38 @@ +/** + * The job-status decision, on its own and with no imports. + * + * Separate from `./index.ts` for one concrete reason: that module opens a + * database connection at import time, and this is the piece of logic most + * worth testing exhaustively without one. A decision that cannot be tested + * cheaply is a decision that does not get tested. + */ + +/** + * What a job's status becomes, given what its printer is reporting. + * + * Pulled out of the database path on purpose: this is the one piece of logic + * here where a mistake reports a print that never happened, and it is worth + * being able to test every pair exhaustively without a printer or a + * Postgres. `undefined` means "leave it alone". + */ +export function nextJobStatus( + current: "uploading" | "printing" | "paused", + printerStatus: "idle" | "printing" | "paused" | "error", +): "printing" | "paused" | "completed" | "failed" | undefined { + switch (printerStatus) { + case "printing": + return "printing"; + case "paused": + return "paused"; + case "error": + return "failed"; + case "idle": + // Only from a state that means the print had actually begun. A job + // still `uploading` on an idle printer has not started yet - the + // transfer takes minutes at the measured 0.2 MiB/s, and the printer is + // legitimately idle for all of it - so calling that "completed" would + // report a successful print that never ran. This is the single most + // important line in the file. + return current === "printing" || current === "paused" ? "completed" : undefined; + } +} diff --git a/src/index.ts b/src/index.ts index 73d4f22..d8284c9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -11,6 +11,7 @@ import { farmRoutes } from "./routes/farms"; import { printerRoutes } from "./routes/printers"; import { jobRoutes } from "./routes/jobs"; import { edgeWsRoute } from "./routes/ws/edge"; +import { startDispatcher } from "./dispatch"; const app = new Elysia() .use(cors()) @@ -65,4 +66,9 @@ const app = new Elysia() console.log(`continuum-backend listening on :${app.server?.port}`); +// After `listen`, so a scheduler tick cannot race the server coming up. It +// only ever sends work to farms with a connected gateway, so on a fresh boot +// it does nothing until one connects. +startDispatcher(); + export type App = typeof app; diff --git a/src/routes/farms.ts b/src/routes/farms.ts index dbd95d0..336644a 100644 --- a/src/routes/farms.ts +++ b/src/routes/farms.ts @@ -6,6 +6,7 @@ import { farms } from "../db/schema"; import { farmProjection } from "../db/projections"; import { requireFarm, requireOrg, requireUser, visibleFarms } from "../auth"; import { PaginationQuery, paginate } from "./pagination"; +import { isGatewayOnline, listUnclaimedPrinters } from "./ws/edge"; const IdParams = t.Object({ id: t.String({ format: "uuid" }) }); @@ -168,4 +169,45 @@ export const farmRoutes = new Elysia({ prefix: "/farms" }) summary: "Rotate and return this farm's edge-proxy token (owner only)", }, }, + ) + /** + * Printers this farm's gateway can see that no printer row claims. + * + * This exists because of the worst first-run experience in the platform. + * A gateway reports its fleet by the ids in its own `printers.toml`, and + * `persistTelemetry` refuses to create rows for ids it does not recognise - + * correctly, since a typo there would otherwise become a permanent phantom + * printer that nobody can explain. But the refusal used to be silent, so a + * new owner saw a connected gateway, an empty fleet, and no way to find out + * why. The answer was in the server's stderr. + * + * Now it is here. Claim one by creating it with `POST /printers`, passing + * the `gatewayPrinterId` from this list; the next telemetry batch fills in + * everything else and the entry disappears from here. + * + * `viewer`, not `owner`: this reports what hardware exists, which is the + * same class of fact as the printer list itself. Creating the row still + * needs `owner`. + * + * `gatewayOnline` is in the response because an empty list is ambiguous on + * its own - it means either "everything is claimed" or "no gateway has ever + * connected", and those need completely different things done about them. + */ + .get( + "/:id/unclaimed-printers", + async ({ user, params }) => { + await requireFarm(user.id, params.id, "viewer"); + return { + farmId: params.id, + gatewayOnline: isGatewayOnline(params.id), + unclaimed: listUnclaimedPrinters(params.id), + }; + }, + { + params: IdParams, + detail: { + tags: ["farms"], + summary: "Printers the gateway reports that no printer row claims yet", + }, + }, ); diff --git a/src/routes/jobs.ts b/src/routes/jobs.ts index acfa81f..c2aea02 100644 --- a/src/routes/jobs.ts +++ b/src/routes/jobs.ts @@ -45,11 +45,18 @@ export const jobRoutes = new Elysia({ prefix: "/jobs" }) * Creates the job row and hands back a presigned URL to PUT the file to. * * `operator`, matching printer commands: a queued job is a request to move - * a machine, even though nothing dispatches it yet. Which is the other - * thing to know about this endpoint - it ends at a `queued` row and an - * upload URL. Nothing transitions the status and nothing reaches a gateway, - * because the gateway has no start-print command yet (see the note in - * continuum-proxy's uplink protocol). Jobs accumulate until it does. + * a machine. + * + * The row lands as `queued` and the scheduler takes it from there - see + * `src/dispatch`. + * + * Note the ordering this endpoint cannot fix: the caller has not used the + * upload URL yet when this returns, so the row exists before the file + * does. The scheduler therefore checks the bucket before dispatching + * rather than assuming, and a job whose file never arrives simply stays + * queued. That also means there is nothing to nudge here - the file is + * guaranteed absent at this point - so a job starts on the first tick + * after its upload lands. */ .post( "/", diff --git a/src/routes/ws/edge.ts b/src/routes/ws/edge.ts index c9111e7..9461f98 100644 --- a/src/routes/ws/edge.ts +++ b/src/routes/ws/edge.ts @@ -9,6 +9,7 @@ import { } from "@continuum/ts-types"; import { db } from "../../db"; import { printers } from "../../db/schema"; +import { advanceJobsFrom } from "../../dispatch"; import { enqueueFrameIngestionBatch } from "../../queue"; import { edgeJwtSecret } from "../../auth/edge_secret"; @@ -59,6 +60,70 @@ const connections = new Map(); */ const gatewayByFarm = new Map(); +/** + * farmId -> the printers its gateway is reporting that nothing in the cloud + * claims, keyed by the gateway's own id for them. + * + * `persistTelemetry` deliberately refuses to insert a printer row for an id + * it does not recognise - a typo in someone's `printers.toml` would otherwise + * become a permanent phantom printer, and a gateway should not be able to + * create records just by mentioning them. But the previous behaviour was to + * log a warning and drop the snapshot, which made the single worst first-run + * experience in the platform: the gateway connects, reports happily, and the + * fleet stays empty with the only evidence in the server's stderr. + * + * So the refusal stands and the *observation* is kept. A new farm's owner can + * list what their gateway can see and claim it with `POST /printers`, which + * is the one place printer rows are created. + * + * In memory on purpose: this is live gateway state, exactly like + * `connections` and `gatewayByFarm` beside it. It is rebuilt within one + * telemetry interval of a reconnect, and persisting it would mean inventing + * a second, staler source of truth about what a gateway can see. + */ +const unclaimedByFarm = new Map>(); + +export interface UnclaimedPrinter { + /** The id in the gateway's `printers.toml` - what to put in `gatewayPrinterId`. */ + gatewayPrinterId: string; + /** The gateway's own display name for it, to make a claim prompt legible. */ + name: string; + /** Its vendor state, so an owner can tell a live printer from a stale entry. */ + vendorState: string | null; + firstSeenAt: string; + lastSeenAt: string; +} + +function noteUnclaimed(farmId: string, snapshot: PrinterTelemetry) { + let farm = unclaimedByFarm.get(farmId); + if (!farm) { + farm = new Map(); + unclaimedByFarm.set(farmId, farm); + } + const now = new Date().toISOString(); + const existing = farm.get(snapshot.printer_id); + farm.set(snapshot.printer_id, { + gatewayPrinterId: snapshot.printer_id, + name: snapshot.name, + vendorState: snapshot.state ?? null, + firstSeenAt: existing?.firstSeenAt ?? now, + lastSeenAt: now, + }); +} + +/** + * What this farm's gateway is reporting that no printer row claims. + * + * Empty is the healthy answer, and also the answer when no gateway has ever + * connected - a caller that wants to tell those apart should check + * `isGatewayOnline` too. + */ +export function listUnclaimedPrinters(farmId: string): UnclaimedPrinter[] { + return [...(unclaimedByFarm.get(farmId)?.values() ?? [])].sort((a, b) => + a.gatewayPrinterId.localeCompare(b.gatewayPrinterId), + ); +} + /** How long to wait for a gateway to answer a command before giving up. */ const COMMAND_TIMEOUT_MS = 10_000; @@ -118,16 +183,16 @@ async function persistTelemetry(conn: EdgeConn, batch: PrinterTelemetry[]) { }); if (!row) { - console.warn( - `[edge] gateway ${conn.gatewayId} reported unknown printer ` + - `'${snapshot.printer_id}'; no printer in farm ${conn.farmId} claims that ` + - `gateway_printer_id, skipping`, - ); + // Still not inserted - see `unclaimedByFarm` for why - but no longer + // only a line in stderr. An owner can now see it and claim it. + noteUnclaimed(conn.farmId, snapshot); continue; } printerId = row.id; conn.printerIds.set(snapshot.printer_id, printerId); + // It is claimed now, so it is no longer waiting to be. + unclaimedByFarm.get(conn.farmId)?.delete(snapshot.printer_id); } const status = mapStatus(snapshot.state, snapshot.print_error); @@ -156,6 +221,12 @@ async function persistTelemetry(conn: EdgeConn, batch: PrinterTelemetry[]) { updatedAt: new Date(), }) .where(eq(printers.id, printerId)); + + // The printer's row is the latest snapshot; the job's row is the history + // that outlives it. This is the only thing that moves a job out of + // `uploading` - see `advanceJobsFrom` for why the gateway's `ok` on a + // start command is explicitly not trusted to do it. + await advanceJobsFrom(printerId, status, snapshot.percent); } } diff --git a/src/s3/index.ts b/src/s3/index.ts index a3b1534..6b75739 100644 --- a/src/s3/index.ts +++ b/src/s3/index.ts @@ -1,4 +1,4 @@ -import { S3Client, PutObjectCommand, GetObjectCommand } from "@aws-sdk/client-s3"; +import { S3Client, PutObjectCommand, GetObjectCommand, HeadObjectCommand } from "@aws-sdk/client-s3"; import { getSignedUrl } from "@aws-sdk/s3-request-presigner"; const accountId = process.env.R2_ACCOUNT_ID ?? ""; @@ -25,6 +25,36 @@ export async function getDownloadUrl(key: string, expiresInSeconds = 900) { return getSignedUrl(s3, command, { expiresIn: expiresInSeconds }); } +/** + * Whether the object is actually in the bucket, and how big it is. + * + * The scheduler needs this because `POST /jobs` returns a presigned PUT and + * the client has not used it yet: the row exists before the file does. Rather + * than trust a client to tell us it finished - a client that crashes + * mid-upload would never say so, and one that lies would have a gateway + * download an error document - this asks the bucket. + * + * Returns `undefined` for a missing object rather than throwing, because for + * this caller "not uploaded yet" is the ordinary case on every tick between + * queueing a job and the upload finishing. + */ +export async function headObject(key: string): Promise<{ size: number } | undefined> { + try { + const head = await s3.send(new HeadObjectCommand({ Bucket: bucket, Key: key })); + return { size: head.ContentLength ?? 0 }; + } catch (err) { + // 404/NotFound is the expected answer for a job whose file has not + // arrived. Anything else - credentials, networking, a bucket that does + // not exist - is a real problem and must not be reported as "no file", + // or every job on a misconfigured deployment would sit queued forever + // with nothing saying why. + const name = (err as { name?: string }).name; + const status = (err as { $metadata?: { httpStatusCode?: number } }).$metadata?.httpStatusCode; + if (name === "NotFound" || name === "NoSuchKey" || status === 404) return undefined; + throw err; + } +} + export function jobFileKey(farmId: string, jobId: string, fileName: string) { return `farms/${farmId}/jobs/${jobId}/${fileName}`; } diff --git a/tests/dispatch.test.ts b/tests/dispatch.test.ts new file mode 100644 index 0000000..15041cb --- /dev/null +++ b/tests/dispatch.test.ts @@ -0,0 +1,72 @@ +import { describe, expect, test } from "bun:test"; +import { nextJobStatus } from "../src/dispatch/job_status"; + +/** + * The job-status mapping, tested exhaustively. + * + * This is the logic that decides whether a print happened. It exists because + * a gateway accepting `start_print` says nothing about the outcome: a real + * end-to-end run on a P1P (2026-10-15) was accepted, uploaded byte-perfect, + * reached RUNNING and then the printer paused itself on a physical + * broken-filament fault. Job state is therefore derived from the telemetry + * that follows, and getting this table wrong means the API reports prints + * that never ran - the one failure nobody would catch by looking. + */ +describe("nextJobStatus", () => { + const currents = ["uploading", "printing", "paused"] as const; + const printerStates = ["idle", "printing", "paused", "error"] as const; + + // Written out rather than computed: a table derived the same way the + // implementation derives it agrees with a wrong implementation. + const expected: Record = { + "uploading/idle": undefined, + "uploading/printing": "printing", + "uploading/paused": "paused", + "uploading/error": "failed", + "printing/idle": "completed", + "printing/printing": "printing", + "printing/paused": "paused", + "printing/error": "failed", + "paused/idle": "completed", + "paused/printing": "printing", + "paused/paused": "paused", + "paused/error": "failed", + }; + + for (const current of currents) { + for (const printerState of printerStates) { + const key = `${current}/${printerState}`; + test(`${key} -> ${expected[key] ?? "unchanged"}`, () => { + expect(nextJobStatus(current, printerState)).toBe(expected[key] as never); + }); + } + } + + test("an idle printer does NOT complete a job that was still uploading", () => { + // The case worth its own test. A printer is idle for the entire transfer + // - minutes of it, at the 0.2 MiB/s measured over FTPS - so "idle" there + // means "has not started", not "finished". Reading it as completion + // would mark every job successful seconds after dispatch, before the + // file had even reached the machine. + expect(nextJobStatus("uploading", "idle")).toBeUndefined(); + }); + + test("a printer error fails the job from any state, including mid-upload", () => { + // The P1P's self-pause arrived while the job was still `uploading` by + // the cloud's reckoning, because the printer faulted during its + // pre-print stages. If `uploading` were excluded here, that job would + // have sat claiming to upload until the reaper got it twenty minutes + // later. + for (const current of currents) { + expect(nextJobStatus(current, "error")).toBe("failed"); + } + }); + + test("completion is never reported from a state the print had not begun in", () => { + // Guards the invariant rather than the table: the only route to + // "completed" is through a state that means the printer had actually + // been printing. + const completable = currents.filter((c) => nextJobStatus(c, "idle") === "completed"); + expect(completable).toEqual(["printing", "paused"]); + }); +}); diff --git a/tests/edge_protocol.test.ts b/tests/edge_protocol.test.ts index 197ee1f..f853d58 100644 --- a/tests/edge_protocol.test.ts +++ b/tests/edge_protocol.test.ts @@ -140,6 +140,19 @@ describe("backend -> gateway frames", () => { printer_id: "p1p", command: { action: "control_print", control: "stop" }, }, + { + type: "command", + command_id: "9a7d2f15-0000-4000-8000-000000000000", + printer_id: "p1p", + command: { + action: "start_print", + job_id: "7c9e6679-0000-4000-8000-000000000000", + file_name: "cube.gcode.3mf", + // Exactly what src/dispatch sends: a presigned GET, minted at + // dispatch time rather than at job creation. + url: "https://example.r2.cloudflarestorage.com/farms/f/jobs/j/cube.gcode.3mf?X-Amz-Signature=x", + }, + }, ]; for (const frame of outbound) { @@ -165,18 +178,61 @@ describe("backend -> gateway frames", () => { } }); - it("has no start-print command to send", () => { - // Still deliberate, but no longer for the original reason: starting an - // uploaded file *has* now been confirmed on a real P1P (2026-10-11). - // What is missing is the path around it - nothing uploads a sliced file - // to a gateway and nothing decides which printer runs which job. Delete - // this test when there is something to start. + it("rejects a start_print missing any of its three required fields", () => { + // The variant exists now - the job scheduler in src/dispatch sends it. + // What this guards is that none of its fields is optional. Each one is + // load-bearing in a different way: without `url` the gateway has nothing + // to fetch, without `file_name` it has nowhere to put it, and without + // `job_id` nothing can line a gateway's logs up against a job. An + // optional field here would fail at the gateway, minutes later, on a + // machine nobody is watching. + const complete = { + action: "start_print", + job_id: "7c9e6679-0000-4000-8000-000000000000", + file_name: "plate_1.3mf", + url: "https://example.test/signed", + }; + check(EdgeServerMessageSchema, { + type: "command", + command_id: "x", + printer_id: "p1p", + command: complete, + }); + + for (const omit of ["job_id", "file_name", "url"] as const) { + const partial: Record = { ...complete }; + delete partial[omit]; + expect( + Value.Check(EdgeServerMessageSchema, { + type: "command", + command_id: "x", + printer_id: "p1p", + command: partial, + }), + ).toBe(false); + } + }); + + it("will not carry a filament or AMS mapping on a start_print", () => { + // Deliberately absent from the contract, and the reason is expensive: + // hardcoding the filament choice instead of reading it from the project + // archive made a real P1P heat to 250C and pause itself at 1%. The + // gateway derives it from the `.3mf`'s own slice_info.config, which is + // the only thing that knows. A field here would invite a caller to guess + // again - and because this schema is a closed union, an extra property + // makes the whole command fail to validate rather than being ignored. expect( Value.Check(EdgeServerMessageSchema, { type: "command", command_id: "x", printer_id: "p1p", - command: { action: "start_print", file_name: "plate_1.3mf" }, + command: { + action: "start_print", + job_id: "7c9e6679-0000-4000-8000-000000000000", + file_name: "plate_1.3mf", + url: "https://example.test/signed", + ams_mapping: [3, -1, -1, -1, -1], + }, }), ).toBe(false); });