Continuum/continuum-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dispatch queued jobs, and derive their state from the printers

Browse Source 
`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.
This commit is contained in:
octoturge
2026-10-15 22:54:44 +00:00
parent 1db96caa91
commit 564d01816b
9 changed files with 723 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/dispatch/index.ts
+383
View File
@@ -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<number> {
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<void> {
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<typeof setInterval> | 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<void> {
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})`);
}
}