diff --git a/scripts/e2e-vendors.sh b/scripts/e2e-vendors.sh index ab7f894..f558732 100755 --- a/scripts/e2e-vendors.sh +++ b/scripts/e2e-vendors.sh @@ -44,6 +44,7 @@ cleanup() { echo "--- proxy log (tail) ---"; tail -30 "$WORK/proxy.log" 2>/dev/null || true echo "--- fake printer log (tail) ---"; tail -15 "$WORK/fake.log" 2>/dev/null || true fi + [ -n "${KEEP_LOGS:-}" ] && cp "$WORK"/*.log "$KEEP_LOGS"/ 2>/dev/null || true rm -rf "$WORK"; exit $rc } trap cleanup EXIT @@ -60,19 +61,22 @@ USER_ID="$(new_uuid)" TOKEN="$(curl -sf "$AUTH_URL/token?sub=$USER_ID" | json 'd["access_token"]')" auth=(-H "authorization: Bearer $TOKEN") -DISPATCH_ENABLED=true bun run src/index.ts > "$WORK/backend.log" 2>&1 & -BACKEND_PID=$! -for _ in $(seq 1 60); do curl -sf "$BASE/health" >/dev/null 2>&1 && break; sleep 0.5; done -curl -sf "$BASE/health" >/dev/null || fail "backend did not become healthy" +start_backend() { + DISPATCH_ENABLED=true bun run src/index.ts >> "$WORK/backend.log" 2>&1 & + BACKEND_PID=$! + for _ in $(seq 1 60); do curl -sf "$BASE/health" >/dev/null 2>&1 && return 0; sleep 0.5; done + fail "backend did not become healthy" +} +start_backend -# run_kind +# run_kind [outage] # With downloads refused the proxy accepts the start and then fails it, which # is the late-failure path: the job must fail with the proxy's reason. run_kind() { - local KIND="$1" ALLOW="$2" - echo "==> $KIND (insecure local downloads: $ALLOW)" + local KIND="$1" ALLOW="$2" OUTAGE="${3:-}" + echo "==> $KIND (insecure local downloads: $ALLOW${OUTAGE:+, cloud outage mid-job})" KEY="key-$(new_uuid | cut -c1-8)" - bun scripts/fake-http-printer.ts "$KIND" "$FAKE_PORT" "$KEY" > "$WORK/fake.log" 2>&1 & + env ${OUTAGE:+RUN_S=14} bun scripts/fake-http-printer.ts "$KIND" "$FAKE_PORT" "$KEY" > "$WORK/fake.log" 2>&1 & FAKE_PID=$! for _ in $(seq 1 40); do grep -q listening "$WORK/fake.log" && break; sleep 0.25; done @@ -120,6 +124,19 @@ run_kind() { CODE="$(curl -s -o /dev/null -w '%{http_code}' -X PUT -H 'Content-Type: application/octet-stream' --data-binary @"$WORK/part.gcode" "$UPLOAD_URL")" [ "$CODE" = 200 ] || fail "job file upload returned $CODE" + if [ -n "$OUTAGE" ]; then + # Take the cloud away the moment the job is dispatched - before any + # telemetry can have told it the print began - and bring it back only + # after the fake print is over. Telemetry alone would leave the job + # `uploading` (idle printer, never seen printing); only the gateway's + # own kept record can complete it. + for _ in $(seq 1 300); do [ "$(psql_q "SELECT status FROM print_jobs WHERE id='$JOB_ID'")" = uploading ] && break; sleep 0.1; done + [ "$(psql_q "SELECT status FROM print_jobs WHERE id='$JOB_ID'")" = uploading ] || fail "the job was not dispatched" + stop "$BACKEND_PID"; BACKEND_PID="" + echo " (cloud down; the fake print runs 14s without it)" + sleep 20 + start_backend + fi SEEN=""; START=$SECONDS; STATUS="" while [ $((SECONDS - START)) -lt 60 ]; do STATUS="$(psql_q "SELECT status FROM print_jobs WHERE id='$JOB_ID'")" @@ -128,13 +145,17 @@ run_kind() { sleep 1 done echo " job status history:$SEEN" - if [ "$ALLOW" = true ]; then + if [ -n "$OUTAGE" ]; then + [ "$STATUS" = completed ] || fail "after the outage the $KIND job is '$STATUS', not completed" + grep -Eq "gateway reported (started|finished)" "$WORK/backend.log" || fail "it completed, but not from the gateway's kept record" + ok "the cloud was away for the whole print; the proxy's kept record completed the job on reconnect" + elif [ "$ALLOW" = true ]; then [ "$STATUS" = completed ] || fail "the $KIND job ended as '$STATUS'" ok "$KIND: queued, scheduled, uploaded, printed and completed (fake printer)" else [ "$STATUS" = failed ] || fail "the refused download left the job '$STATUS', not failed" REASON="$(psql_q "SELECT metadata->>'lastDispatchError' FROM print_jobs WHERE id='$JOB_ID'")" - case "$REASON" in *"could not start it"*"non-HTTPS"*) ok "the job failed at once with the proxy's reason: $REASON" ;; *) fail "the job failed without the proxy's reason: '$REASON'" ;; esac + case "$REASON" in *"non-HTTPS"*) ok "the job failed at once with the proxy's reason: $REASON" ;; *) fail "the job failed without the proxy's reason: '$REASON'" ;; esac fi bun -e " @@ -146,4 +167,5 @@ await s3.send(new DeleteObjectCommand({ Bucket: process.env.R2_BUCKET ?? 'contin } for KIND in $KINDS; do run_kind "$KIND" true; done run_kind klipper false -echo "PASS - $KINDS: added from the API, scheduled and completed against fake printers; a refused download fails the job with its reason." +run_kind klipper true outage +echo "PASS - $KINDS: added from the API, scheduled and completed against fake printers; a refused download fails the job with its reason; a job finished while the cloud was away completes on reconnect." diff --git a/src/dispatch/index.ts b/src/dispatch/index.ts index 0c0fb9c..5e4e90e 100644 --- a/src/dispatch/index.ts +++ b/src/dispatch/index.ts @@ -3,7 +3,7 @@ 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 { jobStatusFromGatewayEvent, nextJobStatus } from "./job_status"; import { rememberStart, takeStart } from "./late_failure"; import { dispatchEnabled, dispatchSwitchWarning } from "./enabled"; @@ -282,6 +282,55 @@ export async function failJobFromGateway(commandId: string, error: string): Prom if (failed.length > 0) console.warn(`[dispatch] job ${jobId} failed: the gateway could not start it: ${error}`); } +/** + * Applies an outcome a gateway kept for us. Throws only when the database + * could not be reached - the one case where the gateway must NOT be told it + * was handled, so that it sends it again. A job that is unknown, belongs to + * another farm, or is no longer in flight is not an error: there is nothing + * to change, and the gateway should stop sending it. + */ +export async function applyGatewayJobEvent( + farmId: string, + event: { job_id: string; event: "started" | "finished" | "failed"; detail?: string | null; at_ms: number }, +): Promise<"applied" | "ignored"> { + // The farm in the `where` is the authorisation: a gateway can only move + // jobs of the farm it is connected for. + const job = await db.query.printJobs.findFirst({ + columns: { id: true, status: true }, + where: and(eq(printJobs.id, event.job_id), eq(printJobs.farmId, farmId)), + }); + if (!job) return "ignored"; + + const next = jobStatusFromGatewayEvent(job.status, event.event); + if (!next) return "ignored"; + + const at = new Date(event.at_ms); + const terminal = next === "completed" || next === "failed"; + const updated = await db + .update(printJobs) + .set({ + status: next, + updatedAt: new Date(), + ...(next === "completed" ? { progressPct: 100 } : {}), + ...(terminal ? { completedAt: at } : {}), + ...(next === "failed" + ? { + metadata: sql`coalesce(${printJobs.metadata}, '{}'::jsonb) || ${JSON.stringify({ + lastDispatchError: event.detail ?? "the gateway reported the job failed", + lastDispatchErrorAt: at.toISOString(), + })}::jsonb`, + } + : {}), + }) + // Re-checked in the write: the status may have moved since the read. + .where(and(eq(printJobs.id, job.id), eq(printJobs.status, job.status))) + .returning({ id: printJobs.id }); + + if (updated.length === 0) return "ignored"; + console.log(`[dispatch] job ${job.id} ${job.status} -> ${next} (gateway reported ${event.event})`); + return "applied"; +} + /** * Fails jobs that have been `uploading` too long. * diff --git a/src/dispatch/job_status.ts b/src/dispatch/job_status.ts index 5966768..4f56356 100644 --- a/src/dispatch/job_status.ts +++ b/src/dispatch/job_status.ts @@ -36,3 +36,30 @@ export function nextJobStatus( return current === "printing" || current === "paused" ? "completed" : undefined; } } + +/** + * What a job's status becomes when its gateway reports an outcome it kept + * for us (`job_event`) - possibly long after it happened, because the + * gateway holds each one until the cloud acknowledges it. + * + * Only jobs still in flight move. A job the cloud has cancelled, failed or + * completed meanwhile is left exactly as it is, which is what makes a stale + * report from a gateway harmless; so is `queued`, a job the cloud took back + * and may already have given to another printer. `undefined` means leave it. + */ +export function jobStatusFromGatewayEvent( + current: string, + event: "started" | "finished" | "failed", +): "printing" | "completed" | "failed" | undefined { + const inFlight = current === "uploading" || current === "printing" || current === "paused"; + if (!inFlight) return undefined; + switch (event) { + case "started": + // Only the first step; a job already printing or paused is further on. + return current === "uploading" ? "printing" : undefined; + case "finished": + return "completed"; + case "failed": + return "failed"; + } +} diff --git a/src/routes/ws/edge.ts b/src/routes/ws/edge.ts index 80f4de0..7a143b7 100644 --- a/src/routes/ws/edge.ts +++ b/src/routes/ws/edge.ts @@ -11,7 +11,7 @@ import { import { db } from "../../db"; import { printerAlerts, printers } from "../../db/schema"; import { deriveAlert, type Status } from "../../alerts/derive"; -import { advanceJobsFrom, failJobFromGateway } from "../../dispatch"; +import { advanceJobsFrom, applyGatewayJobEvent, failJobFromGateway } from "../../dispatch"; import { enqueueFrameIngestionBatch } from "../../queue"; import { edgeJwtSecret } from "../../auth/edge_secret"; import { farmFromEdgeUpgrade } from "../../auth/edge_upgrade"; @@ -891,6 +891,20 @@ export const edgeWsRoute = new Elysia() break; } + case "job_event": { + if (!conn.farmId) return; + try { + await applyGatewayJobEvent(conn.farmId, message); + // Acknowledged whether or not it changed anything: "nothing to + // change" is an answer, and the gateway stops sending it. + conn.send({ type: "job_event_ack", event_id: message.event_id }); + } catch (err) { + // No ack, so the gateway keeps it and sends it again. + console.error(`[edge] could not record job event ${message.event_id} from gateway ${conn.gatewayId}`, err); + } + break; + } + case "discovered_printers": { if (!conn.farmId) return; // No gateway id until `hello`, and the handshake timer closes a diff --git a/tests/late_failure.test.ts b/tests/late_failure.test.ts index f0e3164..5d9127b 100644 --- a/tests/late_failure.test.ts +++ b/tests/late_failure.test.ts @@ -26,3 +26,26 @@ describe("late start failures", () => { expect(takeStart("bulk-1499")).toBe("job-1499"); }); }); + +import { jobStatusFromGatewayEvent } from "../src/dispatch/job_status"; + +describe("jobStatusFromGatewayEvent", () => { + test("a finish is believed for any job still in flight, even one never seen printing", () => { + for (const current of ["uploading", "printing", "paused"]) { + expect(jobStatusFromGatewayEvent(current, "finished")).toBe("completed"); + expect(jobStatusFromGatewayEvent(current, "failed")).toBe("failed"); + } + }); + test("started only promotes a job that was still uploading", () => { + expect(jobStatusFromGatewayEvent("uploading", "started")).toBe("printing"); + expect(jobStatusFromGatewayEvent("printing", "started")).toBeUndefined(); + expect(jobStatusFromGatewayEvent("paused", "started")).toBeUndefined(); + }); + test("a job the cloud has already settled or taken back is left alone", () => { + for (const current of ["queued", "completed", "failed", "cancelled"]) { + for (const event of ["started", "finished", "failed"] as const) { + expect(jobStatusFromGatewayEvent(current, event)).toBeUndefined(); + } + } + }); +});