From 3b42412d987aaea08aa5170a19db0e748d2c791b Mon Sep 17 00:00:00 2001
From: Adam Domurad <adamdomurad@gmail.com>
Date: Tue, 18 Aug 2026 17:01:15 -0400
Subject: [PATCH] message-bus: rethrow handler errors, skip poison messages
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The eachMessage wrapper swallowed handler exceptions, so kafkajs
resolved the offset even when the handler's database write had failed —
silently losing the event. Meanwhile a message that failed BSON
deserialization threw outside the catch, crash-looping the partition on
a permanently bad message. Both are inverted to the correct behavior:

- handler errors are rethrown so the offset is not resolved and kafkajs
  redelivers (at-least-once; handlers are id-keyed upserts, safe to
  replay)
- undeserializable messages are logged and skipped: retrying a poison
  pill can never succeed

Identified in the AztecScan production-readiness audit (failed-message
acknowledgement, immediate work order #1).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015mXnawSBPUySPtPwjLp5g1
---
 packages/message-bus/src/class.ts | 33 ++++++++++++++++++++++++-------
 1 file changed, 26 insertions(+), 7 deletions(-)

diff --git a/packages/message-bus/src/class.ts b/packages/message-bus/src/class.ts
index 7f888aab..369ca137 100644
--- a/packages/message-bus/src/class.ts
+++ b/packages/message-bus/src/class.ts
@@ -232,13 +232,26 @@ export class MessageBus {
             : typeof messageValue === "string"
               ? new TextEncoder().encode(messageValue)
               : new Uint8Array(0);
-        const deserializedObj = BSON.deserialize(valueAsUint8Array);
-        if (typeof deserializedObj?.data !== "object") {
-          throw new Error(
-            `Deserialized message does not contain a valid data object`,
+        // A message that cannot be deserialized is a poison pill: retrying
+        // it can never succeed, so log it and move on rather than
+        // crash-looping the whole partition.
+        let data: object;
+        try {
+          const deserializedObj = BSON.deserialize(valueAsUint8Array);
+          if (typeof deserializedObj?.data !== "object") {
+            throw new Error(
+              `Deserialized message does not contain a valid data object`,
+            );
+          }
+          data = deserializedObj.data as object;
+        } catch (e) {
+          this.logger.error(
+            `Skipping undeserializable message on topic ${topic}: ${
+              e instanceof Error ? e.stack : String(e)
+            }`,
           );
+          return;
         }
-        const data = deserializedObj.data as object;
         const cb = this.#consumers[groupId]?.topicCallbacks[topic];
         if (cb) {
           // Send a heartbeat before invoking the handler so the broker knows
@@ -251,16 +264,22 @@ export class MessageBus {
             // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
             await cb(data);
           } catch (e) {
+            // Rethrow so kafkajs does NOT resolve this offset. Swallowing
+            // here acknowledged messages whose handler (typically a database
+            // write) had failed, silently losing the event; rethrowing makes
+            // kafkajs redeliver it, giving at-least-once semantics. Handlers
+            // are id-keyed upserts, so redelivery is safe.
             if (e instanceof Error) {
               this.logger.error(
-                `Provided callback for topic ${topic} failed: ${e.stack}`,
+                `Provided callback for topic ${topic} failed (message will be redelivered): ${e.stack}`,
               );
             } else {
               this.logger.warn(
                 // eslint-disable-next-line @typescript-eslint/restrict-template-expressions
-                `Provided callback for topic ${topic} failed with non-Error: ${e}`,
+                `Provided callback for topic ${topic} failed with non-Error (message will be redelivered): ${e}`,
               );
             }
+            throw e;
           }
         }
       },