Fail a job when the gateway reports it could not start it

A start_print is acknowledged as soon as the gateway accepts it, and the
download and transfer that follow can still fail. That late result was only
logged, so the job sat in uploading until the stall reaper gave up twenty
minutes later.

The dispatcher now remembers which job each start command belonged to. A
late failure for it fails the job with the gateway's own reason in
metadata.lastDispatchError - but only while the job is still uploading, so a
job the printer has since started is left to its telemetry.

scripts/e2e-vendors.sh gains a phase where the proxy refuses the download
and the job must fail at once with that reason.

scripts/e2e-vendors.sh

@@ -65,8 +65,12 @@ 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"
 
-for KIND in $KINDS; do
-  echo "==> $KIND"
+# run_kind <kind> <allow insecure downloads: true|false>
+# 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)"
   KEY="key-$(new_uuid | cut -c1-8)"
   bun scripts/fake-http-printer.ts "$KIND" "$FAKE_PORT" "$KEY" > "$WORK/fake.log" 2>&1 &
   FAKE_PID=$!
@@ -78,7 +82,7 @@ for KIND in $KINDS; do
   GW_TOKEN="$(curl -sf "${auth[@]}" -X POST "$BASE/farms/$FARM_ID/edge-proxy-token" | json 'd["edgeProxyToken"]')"
 
   env -C "$PROXY_DIR" \
-    CONTINUUM_ALLOW_INSECURE_LOCAL_DOWNLOADS=true CONTINUUM_GATEWAY_ID=e2e-vendors-gw CONTINUUM_FARM_ID="$FARM_ID" CONTINUUM_GATEWAY_TOKEN="$GW_TOKEN" \
+    CONTINUUM_ALLOW_INSECURE_LOCAL_DOWNLOADS="$ALLOW" CONTINUUM_GATEWAY_ID=e2e-vendors-gw CONTINUUM_FARM_ID="$FARM_ID" CONTINUUM_GATEWAY_TOKEN="$GW_TOKEN" \
     CONTINUUM_UPLINK_URL="ws://localhost:$PORT/ws/edge/v1" CONTINUUM_GO2RTC_BIN=/nonexistent-go2rtc CONTINUUM_GO2RTC_CONFIG="$WORK/go2rtc.yaml" \
     CONTINUUM_PRINTERS_PATH="$WORK/no-such-printers.toml" CONTINUUM_STORE_PATH="$WORK/store-$KIND/continuum.db" RUST_LOG=info \
     "$PROXY_DIR/target/debug/continuum-proxy" > "$WORK/proxy.log" 2>&1 &
@@ -117,15 +121,21 @@ for KIND in $KINDS; do
   [ "$CODE" = 200 ] || fail "job file upload returned $CODE"
 
   SEEN=""; START=$SECONDS; STATUS=""
-  while [ $((SECONDS - START)) -lt 90 ]; do
+  while [ $((SECONDS - START)) -lt 60 ]; do
     STATUS="$(psql_q "SELECT status FROM print_jobs WHERE id='$JOB_ID'")"
     case " $SEEN " in *" $STATUS "*) ;; *) SEEN="$SEEN $STATUS" ;; esac
     [ "$STATUS" = completed ] || [ "$STATUS" = failed ] && break
     sleep 1
   done
   echo "  job status history:$SEEN"
-  [ "$STATUS" = completed ] || fail "the $KIND job ended as '$STATUS'"
-  ok "$KIND: queued, scheduled, uploaded, printed and completed (fake printer)"
+  if [ "$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
+  fi
 
   bun -e "
 import { s3 } from './src/s3';
@@ -133,5 +143,7 @@ import { DeleteObjectCommand } from '@aws-sdk/client-s3';
 await s3.send(new DeleteObjectCommand({ Bucket: process.env.R2_BUCKET ?? 'continuum', Key: process.argv[1] }));
 " "$OBJECT_KEY" >/dev/null 2>&1 || true
   stop "$PROXY_PID" "$FAKE_PID"; PROXY_PID=""; FAKE_PID=""
-done
-echo "PASS - $KINDS: added from the API, scheduled and completed against fake printers."
+}
+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."

src/dispatch/index.ts

@@ -1,9 +1,10 @@
-import { and, asc, eq, inArray, isNotNull, isNull, lt } from "drizzle-orm";
+import { and, asc, eq, inArray, isNotNull, isNull, lt, sql } 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 { rememberStart, takeStart } from "./late_failure";
 import { dispatchEnabled, dispatchSwitchWarning } from "./enabled";
 
 /**
@@ -217,6 +218,9 @@ export async function dispatchTick(): Promise<number> {
         ...(typeof job.metadata?.amsSlot === "number" ? { ams_slot: job.metadata.amsSlot } : {}),
       });
 
+      // Accepted or not, a late result for this command can name the job.
+      rememberStart(result.commandId, job.id);
+
       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
@@ -249,6 +253,35 @@ export async function dispatchTick(): Promise<number> {
   return dispatched;
 }
 
+/**
+ * A gateway that had accepted a `start_print` reports that the download or
+ * transfer then failed. Fails the job with that reason - but only while it is
+ * still `uploading`: one that has since reached `printing` is being driven by
+ * the printer's own telemetry, and a late message must not overrule it.
+ *
+ * Failed rather than requeued, for the reason `reapStalledUploads` gives.
+ */
+export async function failJobFromGateway(commandId: string, error: string): Promise<void> {
+  const jobId = takeStart(commandId);
+  if (!jobId) return;
+
+  const failed = await db
+    .update(printJobs)
+    .set({
+      status: "failed",
+      completedAt: new Date(),
+      updatedAt: new Date(),
+      metadata: sql`coalesce(${printJobs.metadata}, '{}'::jsonb) || ${JSON.stringify({
+        lastDispatchError: `the gateway could not start it: ${error}`,
+        lastDispatchErrorAt: new Date().toISOString(),
+      })}::jsonb`,
+    })
+    .where(and(eq(printJobs.id, jobId), eq(printJobs.status, "uploading")))
+    .returning({ id: printJobs.id });
+
+  if (failed.length > 0) console.warn(`[dispatch] job ${jobId} failed: the gateway could not start it: ${error}`);
+}
+
 /**
  * Fails jobs that have been `uploading` too long.
  *
@@ -274,7 +307,7 @@ async function reapStalledUploads(): Promise<void> {
         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`,
+          `the download or transfer error (a failure it reports promptly fails the job sooner)`,
         lastDispatchErrorAt: new Date().toISOString(),
       },
     })

src/dispatch/late_failure.ts

@@ -0,0 +1,33 @@
+/**
+ * A `start_print` is answered the moment the gateway accepts it, because the
+ * download and transfer take minutes and this side stops waiting after ten
+ * seconds. If that work then fails, the gateway sends a second `command_result`
+ * for the same command id - by which time nothing here is waiting for it.
+ *
+ * This remembers which job each start belongs to, so that second result can
+ * fail the job with the gateway's own reason instead of the job sitting in
+ * `uploading` until the stall reaper gives up on it twenty minutes later.
+ *
+ * In memory on purpose: after a restart the reaper is still there to catch
+ * what this forgot.
+ */
+const MAX_REMEMBERED = 1000;
+const REMEMBER_MS = 60 * 60 * 1000;
+
+const starts = new Map<string, { jobId: string; at: number }>();
+
+export function rememberStart(commandId: string, jobId: string, now = Date.now()): void {
+  starts.set(commandId, { jobId, at: now });
+  // Maps iterate in insertion order, so the oldest entries come first.
+  for (const [id, entry] of starts) {
+    if (starts.size <= MAX_REMEMBERED && now - entry.at < REMEMBER_MS) break;
+    starts.delete(id);
+  }
+}
+
+/** The job a start command was for, forgetting it: a command is only answered late once. */
+export function takeStart(commandId: string): string | undefined {
+  const entry = starts.get(commandId);
+  starts.delete(commandId);
+  return entry?.jobId;
+}

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 } from "../../dispatch";
+import { advanceJobsFrom, failJobFromGateway } from "../../dispatch";
 import { enqueueFrameIngestionBatch } from "../../queue";
 import { edgeJwtSecret } from "../../auth/edge_secret";
 import { farmFromEdgeUpgrade } from "../../auth/edge_upgrade";
@@ -867,17 +867,19 @@ export const edgeWsRoute = new Elysia()
             // gives up after ten seconds - so its real outcome can only ever
             // arrive late.
             //
-            // The reason is logged rather than dropped. It is the difference
-            // between an operator seeing "job failed" and seeing which half
-            // of a download-then-upload failed and what the printer said.
-            // It does not change the job's fate: the job never reaching
-            // `printing` is what fails it, and that already works. Recording
-            // it against the job needs a column print_jobs does not have.
+            // The reason is logged, and when the command was a start we
+            // remember (`rememberStart`), it also fails that job with the
+            // gateway's own words rather than leaving it in `uploading`.
             const reason = message.ok ? "reported success" : `failed: ${message.error ?? "no reason given"}`;
             console.warn(
               `[edge] gateway ${conn.gatewayId} sent a result for unknown, ` +
                 `expired or already-answered command ${message.command_id} - ${reason}`,
             );
+            if (!message.ok) {
+              failJobFromGateway(message.command_id, message.error ?? "no reason given").catch((err) =>
+                console.error("[edge] could not record a late start failure", err),
+              );
+            }
             return;
           }
           clearTimeout(pending.timer);

tests/late_failure.test.ts

@@ -0,0 +1,28 @@
+import { describe, expect, test } from "bun:test";
+import { rememberStart, takeStart } from "../src/dispatch/late_failure";
+
+describe("late start failures", () => {
+  test("a remembered start is returned once, then forgotten", () => {
+    rememberStart("c1", "job-1");
+    expect(takeStart("c1")).toBe("job-1");
+    expect(takeStart("c1")).toBeUndefined();
+  });
+
+  test("a command nobody started is unknown", () => {
+    expect(takeStart("never-sent")).toBeUndefined();
+  });
+
+  test("an hour-old start is dropped rather than failing a job long since resolved", () => {
+    const longAgo = Date.now() - 2 * 60 * 60 * 1000;
+    rememberStart("old", "job-old", longAgo);
+    rememberStart("new", "job-new");
+    expect(takeStart("old")).toBeUndefined();
+    expect(takeStart("new")).toBe("job-new");
+  });
+
+  test("the table is bounded", () => {
+    for (let i = 0; i < 1500; i++) rememberStart(`bulk-${i}`, `job-${i}`);
+    expect(takeStart("bulk-0")).toBeUndefined();
+    expect(takeStart("bulk-1499")).toBe("job-1499");
+  });
+});