From 34a1b14b5e6dc68e31be123f9f3bb50429442f2e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E9=A2=A8=E9=AD=94=E5=B0=8F=E6=AC=A1=E9=83=8E?=
 <xhfkindergarten@gmail.com>
Date: Fri, 10 Jul 2026 15:44:18 +0800
Subject: [PATCH] fix(node): forward streamAllMessages errors to callbacks

---
 bindings/node/src/conversations/streams.rs | 59 ++++++++--------------
 sdks/js/node-sdk/test/streams.test.ts      | 38 +++++++++++++-
 2 files changed, 57 insertions(+), 40 deletions(-)

diff --git a/bindings/node/src/conversations/streams.rs b/bindings/node/src/conversations/streams.rs
index 3765d8e29d..cfdb3e4d45 100644
--- a/bindings/node/src/conversations/streams.rs
+++ b/bindings/node/src/conversations/streams.rs
@@ -99,49 +99,30 @@ impl Conversations {
         .collect()
     });
 
-    let on_message =
-      move |message: std::result::Result<_, xmtp_mls::subscriptions::SubscribeError>| {
-        tracing::trace!(
-            inbox_id,
-            conversation_type = ?conversation_type,
-            "[received] message result"
-        );
+    let on_message = move |message| {
+      tracing::trace!(
+          inbox_id,
+          conversation_type = ?conversation_type,
+          "[received] message result"
+      );
 
-        // Skip any messages that are errors
-        if let Err(err) = &message {
-          tracing::warn!(
-            inbox_id,
-            error = ?err,
-            "[received] message error, swallowing to continue stream"
-          );
-          return; // Skip this message entirely
-        }
+      if let Err(error) = &message {
+        tracing::warn!(
+          inbox_id,
+          error = ?error,
+          "[received] forwarding message error to callback"
+        );
+      }
 
-        // For successful messages, try to transform and pass to JS
-        // otherwise log error and continue stream
-        match message
+      let status = callback.call(
+        message
           .map(Into::into)
           .map_err(ErrorWrapper::from)
-          .map_err(Error::from)
-        {
-          Ok(transformed_msg) => {
-            tracing::trace!(
-              inbox_id,
-              "[received] calling tsfn callback with successful message"
-            );
-            let status = callback.call(Ok(transformed_msg), ThreadsafeFunctionCallMode::Blocking);
-            tracing::info!("Stream status: {:?}", status);
-          }
-          Err(err) => {
-            // Just in case the transformation itself fails
-            tracing::error!(
-              inbox_id,
-              error = ?err,
-              "[received] error during message transformation, swallowing to continue stream"
-            );
-          }
-        }
-      };
+          .map_err(Error::from),
+        ThreadsafeFunctionCallMode::Blocking,
+      );
+      tracing::info!("Stream status: {:?}", status);
+    };
     let on_close = move || {
       on_close.call(Ok(()), ThreadsafeFunctionCallMode::Blocking);
     };
diff --git a/sdks/js/node-sdk/test/streams.test.ts b/sdks/js/node-sdk/test/streams.test.ts
index 456a0ae6d3..97a1fb720f 100644
--- a/sdks/js/node-sdk/test/streams.test.ts
+++ b/sdks/js/node-sdk/test/streams.test.ts
@@ -1,6 +1,6 @@
 import { describe, expect, it, vi } from "vitest";
 import { StreamFailedError } from "@/utils/errors";
-import { createStream } from "@/utils/streams";
+import { createStream, type StreamCallback } from "@/utils/streams";
 
 describe("createStream", () => {
   it("should forward StreamFailedError to onError", async () => {
@@ -35,4 +35,40 @@ describe("createStream", () => {
 
     expect(onErrorSpy).toHaveBeenCalledWith(expect.any(StreamFailedError));
   });
+
+  it("should report item errors without ending the stream", async () => {
+    const itemError = new Error("message processing failed");
+    const onErrorSpy = vi.fn();
+    const onFailSpy = vi.fn();
+    const onValueSpy = vi.fn();
+    let emit!: StreamCallback<number>;
+
+    const mockStreamFunction = vi.fn(
+      async (callback: StreamCallback<number>) => {
+        emit = callback;
+        return Promise.resolve({
+          end: vi.fn(),
+          endAndWait: vi.fn().mockResolvedValue(undefined),
+          isClosed: vi.fn().mockReturnValue(false),
+          waitForReady: vi.fn().mockResolvedValue(undefined),
+        });
+      },
+    );
+
+    const stream = await createStream<number>(mockStreamFunction, undefined, {
+      onError: onErrorSpy,
+      onFail: onFailSpy,
+      onValue: onValueSpy,
+    });
+
+    emit(itemError, undefined);
+    emit(null, 42);
+
+    expect(onErrorSpy).toHaveBeenCalledOnce();
+    expect(onErrorSpy).toHaveBeenCalledWith(itemError);
+    expect(onValueSpy).toHaveBeenCalledWith(42);
+    expect(onFailSpy).not.toHaveBeenCalled();
+
+    void stream.end();
+  });
 });