From 42853bd8df3cfd3e6e8a578e92a6342b4be3dbf7 Mon Sep 17 00:00:00 2001
From: shin-core <153108882+shin-core@users.noreply.github.com>
Date: Wed, 15 Jul 2026 16:27:31 +1000
Subject: [PATCH] fix(queue): log unrecognized job types in processJob instead
 of dropping silently (#5836)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

processJob's `switch (message.type)` in src/queue/job-dispatch.ts fanned out to
dozens of handlers with no `default:` case. An unrecognized message.type — a
stale queued message from a renamed/removed job type, a producer/consumer skew
during a rolling deploy, or a corrupted payload — fell through and was acked by
the caller with zero trace: no log line, no metric, no audit event. This was the
only ack-and-drop path in the queue pipeline with no observability, unlike the
`retired_review_job_ignored` (src/index.ts) and `dlq_message_dead_lettered`
(src/queue/dlq.ts) precedents.

Add a `default:` case that emits a structured `unknown_job_type_ignored` warning
(level/event/jobType, mirroring those precedents' JSON shape) via console.warn,
then returns normally. It never throws, so the existing ack-after-processJob
flow is unaffected — this is purely an observability addition. No existing case
branch is changed.

The regression test imports processJob directly from ./job-dispatch (the file
changed) and asserts the default branch logs the event without throwing, plus a
negative case confirming a recognized type does not emit the warning.
---
 src/queue/job-dispatch.ts      | 15 +++++++++++++
 test/unit/job-dispatch.test.ts | 41 ++++++++++++++++++++++++++++++++++
 2 files changed, 56 insertions(+)
 create mode 100644 test/unit/job-dispatch.test.ts

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<void> {
       // 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<string, unknown>;
+    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);
+  });
+});