Apply job outcomes a gateway kept while the cloud was away

A job_event from a gateway moves a job that is still in flight (uploading,
printing, paused) to printing, completed or failed, and is acknowledged
whether or not it changed anything so the gateway stops sending it. A job
the cloud has cancelled, completed or put back in the queue is left alone,
and a gateway can only touch jobs of its own farm. If the database is
unreachable it is not acknowledged, so it is sent again.

scripts/e2e-vendors.sh gains a phase that takes the backend away for the
whole of a fake print and checks the job completes on reconnect from the
proxy's kept record.

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 &
+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 && break; sleep 0.5; done
+  for _ in $(seq 1 60); do curl -sf "$BASE/health" >/dev/null 2>&1 && return 0; sleep 0.5; done
-curl -sf "$BASE/health" >/dev/null || fail "backend did not become healthy"
+  fail "backend did not become healthy"
+}
+start_backend
 
-# run_kind <kind> <allow insecure downloads: true|false>
+# run_kind <kind> <allow insecure downloads: true|false> [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"
+  local KIND="$1" ALLOW="$2" OUTAGE="${3:-}"
-  echo "==> $KIND (insecure local downloads: $ALLOW)"
+  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."

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.
  *

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";
+  }
+}

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

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();
+      }
+    }
+  });
+});