diff --git a/scripts/e2e-vendors.sh b/scripts/e2e-vendors.sh index 064c6e8..ab7f894 100755 --- a/scripts/e2e-vendors.sh +++ b/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 +# 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." diff --git a/src/dispatch/index.ts b/src/dispatch/index.ts index 6295ab9..0c0fb9c 100644 --- a/src/dispatch/index.ts +++ b/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 { ...(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 { 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 { + 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 { 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(), }, }) diff --git a/src/dispatch/late_failure.ts b/src/dispatch/late_failure.ts new file mode 100644 index 0000000..c7a63b7 --- /dev/null +++ b/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(); + +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; +} diff --git a/src/routes/ws/edge.ts b/src/routes/ws/edge.ts index c74c2cd..80f4de0 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 } 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); diff --git a/tests/late_failure.test.ts b/tests/late_failure.test.ts new file mode 100644 index 0000000..f0e3164 --- /dev/null +++ b/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"); + }); +});