diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index 255c34f114..2daf331898 100644 --- a/src/queue/job-dispatch.ts +++ b/src/queue/job-dispatch.ts @@ -343,5 +343,20 @@ export async function processJob(env: Env, message: JobMessage): Promise { // an empty table). Never throws. await retryFailedRelays(env); return; + default: + // An unrecognized job type (a stale queued message from a renamed/removed type, a producer/consumer skew + // during a rolling deploy, or a corrupted payload) would otherwise fall through and be acked with zero + // trace. Log it — matching the retired_review_job_ignored (src/index.ts) / dlq_message_dead_lettered + // (src/queue/dlq.ts) structured-warn precedents — then return normally so the caller's ack flow is + // unchanged. Observability only; never throws (#5836). message narrows to `never` here, so read the + // runtime type through a cast. + console.warn( + JSON.stringify({ + level: "warn", + event: "unknown_job_type_ignored", + jobType: (message as { type?: unknown }).type, + }), + ); + return; } } diff --git a/test/unit/job-dispatch.test.ts b/test/unit/job-dispatch.test.ts new file mode 100644 index 0000000000..fa3fafc0b2 --- /dev/null +++ b/test/unit/job-dispatch.test.ts @@ -0,0 +1,41 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { processJob } from "../../src/queue/job-dispatch"; +import { createTestEnv } from "../helpers/d1"; +import type { JobMessage } from "../../src/types"; + +describe("processJob unknown job type (#5836)", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("logs a structured unknown_job_type_ignored warning and does not throw for an unrecognized type", async () => { + const warnLogs: string[] = []; + vi.spyOn(console, "warn").mockImplementation((...args: unknown[]) => { + warnLogs.push(String(args[0])); + }); + + const env = createTestEnv(); + // A type outside the discriminated union — a stale/renamed job or a producer/consumer skew at runtime. + const message = { type: "totally-unknown-job-type" } as unknown as JobMessage; + + await expect(processJob(env, message)).resolves.toBeUndefined(); + + expect(warnLogs).toHaveLength(1); + const log = JSON.parse(warnLogs[0] ?? "{}") as Record; + expect(log).toMatchObject({ level: "warn", event: "unknown_job_type_ignored", jobType: "totally-unknown-job-type" }); + }); + + it("does not log the unknown-type warning for a recognized job type", async () => { + const warnLogs: string[] = []; + vi.spyOn(console, "warn").mockImplementation((...args: unknown[]) => { + warnLogs.push(String(args[0])); + }); + + const env = createTestEnv(); + // A recognized type that no-ops safely without external I/O: retryFailedRelays fails open on an empty table. + await processJob(env, { type: "retry-orb-relay" } as JobMessage); + + expect(warnLogs.some((line) => line.includes("unknown_job_type_ignored"))).toBe(false); + }); +});