XMTP Node stream error

[XHFkindergarten]

XMTP Node stream error PR handoff

Purpose

This handoff contains the complete patch and proposed upstream PR text for
xmtp/libxmtp.

The patch fixes one independently provable Node binding bug:

Conversations::stream_all_messages logs and discards every per-item
SubscribeError before it can reach JavaScript.

After this patch, the existing error-first callback receives
(error, undefined) and the underlying stream continues consuming later
items. The public N-API and TypeScript signatures do not change.

This is deliberately an observability/contract-parity fix. It does not
claim to fix every transport or MLS omission tracked by #2351/#3541, does not
change cursor advancement, and does not add automatic replay.

Base

• Repository: https://github.com/xmtp/libxmtp
• Base branch: main
• Base commit used to prepare and verify this patch:
  6943ad25ecdcf17e3723353ca655a4f5b860213c
• Base commit title:
  feat(xmtp_mls,bindings): route callback streams over bidi behind XMTP_BIDI_STREAMS_ENABLED (#3832)

If upstream main has moved, apply the patch to current main and resolve the
small context difference in bindings/node/src/conversations/streams.rs.

Changed files

1. bindings/node/src/conversations/streams.rs
   • removes the special Err => warn + return behavior;
   • preserves a warning, but changes it to state that the error is forwarded;
   • passes both Ok(Message) and Err(napi::Error) through the existing
     ThreadsafeFunction callback.
2. sdks/js/node-sdk/test/streams.test.ts
   • proves that a per-item callback error reaches onError;
   • proves that a subsequent value still reaches onValue;
   • proves the item error does not invoke the terminal onFail path.

Why this behavior is correct

• Conversations::stream, single-conversation message streams, consent
  streams, and preference streams already forward errors through their
  callbacks.
• Mobile and WASM streamAllMessages also expose the equivalent error.
• libxmtp #899 and
  PR #1182 introduced stream error
  forwarding so external SDKs could handle failures.
• libxmtp PR #1938 later swallowed
  aggregate message-stream errors because the old Node AsyncStream ended on
  an error.
• xmtp-js PR #1118 subsequently
  changed that contract: callback errors now go only to onError, and the
  stream remains active. Therefore Swallow stream all msg errors in node bindings #1938's original reason no longer applies.
• Sending an item error to on_close would be wrong: it is a terminal signal
  and can start a second subscription while the original per-item-error stream
  is still alive.

Complete patch

Save the following block as /tmp/xmtp-node-stream-errors.patch and run:

git apply --check /tmp/xmtp-node-stream-errors.patch
git apply /tmp/xmtp-node-stream-errors.patch

diff --git a/bindings/node/src/conversations/streams.rs b/bindings/node/src/conversations/streams.rs
index 3765d8e2..cfdb3e4d 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 456a0ae6..97a1fb72 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();
+  });
 });

Suggested commit

fix(node): forward streamAllMessages errors to callbacks

Suggested PR title

fix(node): forward streamAllMessages errors to JavaScript callbacks

Suggested PR body

## Summary

Forward per-item `SubscribeError` values emitted by
`Conversations::stream_all_messages` to the Node N-API callback instead of
logging and discarding them.

The stream continues consuming subsequent items after reporting the error.

## Context

#899 and #1182 introduced stream error propagation so external SDKs could
observe and handle failures.

#1938 later made `stream_all_messages` a special case: errors are logged and
discarded before crossing the N-API boundary. At that time the Node
`AsyncStream` ended when an error reached it. xmtp-js #1118 subsequently
changed that behavior: item errors are passed to `onError` and the stream
remains active.

As a result, the original reason for suppressing these errors no longer
applies. Keeping the suppression now makes the Node SDK's `onError` handler
unreachable for aggregate message-stream item errors, while conversation,
single-conversation message, consent, preference, Mobile, and WASM streams
still expose their errors.

This makes failures associated with reports such as #2351 and #3541 silent to
Node applications and prevents consumers from applying an explicit recovery
policy.

## Behavior

Before:

`Rust stream item Err -> warning log -> no JavaScript callback`

After:

`Rust stream item Err -> callback(error, undefined) -> stream continues`

There is no public API signature change.

## Scope

This PR restores observability at the Node binding boundary. It does not claim
to fix the underlying transport or MLS delivery omissions tracked by #2351 and
#3541, and it does not change cursor or retry behavior.

## Verification

- `cargo check --locked -p bindings_node --lib`
- `yarn workspace @xmtp/node-sdk test test/streams.test.ts`
- `eslint node-sdk/test/streams.test.ts`

Refs #899, #1182, #1938, #2351, #3541.

Local verification results

Completed successfully on the preparation machine:

• cargo fmt / direct rustfmt --check for the changed Rust file: passed.
• git diff --check: passed.
• cargo check --locked -p bindings_node --lib: passed.
  • It emitted one unrelated existing unused_mut warning in
    crates/xmtp_mls/src/identity.rs:842.
• yarn workspace @xmtp/node-sdk test test/streams.test.ts: passed,
  1 file / 2 tests.
• eslint node-sdk/test/streams.test.ts: passed.

Not completed locally:

• A standalone cargo test -p bindings_node --lib cannot link on this machine
  because a Rust test binary does not provide the Node N-API symbols. The Rust
  unit-test attempt was removed from the final patch.
• Full Node SDK type-check requires first building the current Node binding
  portal. Without that build, TypeScript resolved stale binding declarations
  outside the worktree and produced unrelated missing-export errors.
• Full integration tests require XMTP's local backend.

Run these on the submission machine before marking the PR ready:

cargo fmt --check
cargo check --locked -p bindings_node
just node check
just backend up
just node test
just js test-node-sdk-ci

Review notes and scope boundary

1. This PR does not recover the item that produced an error. It only makes the
   error visible and preserves subsequent stream delivery.
2. The message: None plus cursor-advancement path remains a separate design
   issue. Changing that in this PR could cause duplicate delivery or a stuck
   cursor because None also represents legitimate MLS commits with no
   application message.
3. Do not write Fixes #2351 or Fixes #3541; use Refs only.
4. XMTP's CONTRIBUTING.md requires meaningful human review and understanding
   of AI-assisted changes. Review this patch and PR reasoning before submitting
   or marking a draft ready.

Supporting evidence from our reproduction

• Full upstream report:
  dist/xmtp-live-probe/XMTP_UPSTREAM_REPORT.md
• Machine-readable native-boundary result:
  dist/xmtp-live-probe/results/native-boundary-20260708-round1-loss.json
• The reproduction showed 1,000 successful sends, 997/997 native callbacks,
  997/997 handlers, zero native-callback errors, and three common omissions
  before the JavaScript-facing callback.

Those results demonstrate why restoring the missing error signal is necessary,
but they do not prove that a swallowed Rust error caused each omitted message;
server-side non-delivery from #517 remains another possible root cause.