feat(xmtp_api_d14n): BidiConnection actor — own both wire halves, internal liveness

[tylerhawkes]

What

A bidirectional subscription connection actor (XIP-83), living at the transport layer in xmtp_api_d14n next to the subscribe_bidi transport. It owns a single bidi stream end-to-end and is the sole writer of the request half.

This supersedes the original per-subscription actor that lived in xmtp_mls (BidiSubscription): that design held an outbound sender purely to auto-pong, which decoupled the request half's lifetime from the stream's and let a post-close mutate/ping succeed silently (Macroscope flagged exactly this). Pushing the actor down to where the wire lives fixes it by ownership rather than with a side-signal.

Design

• Sole owner of both halves. Callers submit Command::{Mutate, Probe} over a channel; the actor multiplexes them with auto-pongs onto the one outbound FIFO. No second outbound sender exists anywhere.
• Liveness is internal. Server Ping → auto-Pong, never surfaced. probe() sends a client Ping and the actor correlates the returning Pong via a oneshot. BidiEvent has no Ping/Pong arm — consumers deal with zero keepalive.
• Teardown by ownership. Inbound end/error → the actor drops wire_out (request half closes, HTTP/2 torn down both ways), the command receiver (→ mutate/probe return Closed), the event sender (→ next() ends), and any pending probe acks. One place, every teardown — the "enqueue into a dead stream" gap can't exist.
• Bounded probe. probe() self-bounds at N × keepalive_interval_ms (N=3; 30s fallback until Started, whose cadence the actor captures). probe_within(Duration) lets a latency-sensitive caller — e.g. a push-notification handler — bound to seconds. A timeout returns the distinct ProbeTimedOut → drop + re-open; tight bounds are safe because resume-from-cursors dedupes.

Notes

• Native-only (#[cfg(not(target_arch = "wasm32"))]), generic over XmtpMlsBidiStreams — mock-tested, no deployed node required.
• This is the Phase-1 primitive the herald connection-multiplexer (Phase 2) drives per shard; durability / reconnection stays at the mux / consumer layer, not coupled into the actor.
• Production Topic parsing uses the validating TryFrom (not the test-only from_bytes); malformed TopicsLive topics are skipped with a warn.
• 11 unit tests: ownership-teardown, auto-pong-not-surfaced, probe round-trip / timeout / keepalive-derived default, wire-order, unknown-version tolerance.

Note

Add BidiConnection actor for bidirectional subscription streams in xmtp_api_d14n

• Adds connection.rs, a native-only module exposing BidiConnection, a handle that owns both inbound and outbound halves of a gRPC bidi stream.
• The actor sends an initial Mutate first, auto-responds to server Ping frames, and correlates client probe()/probe_within() calls to Pong responses for liveness checking.
• Outbound frames are buffered in a VecDeque when the transport is not draining; if the backlog exceeds MAX_PENDING_FRAMES, the actor tears down the connection.
• BidiEvent surfaces structured events (Started, CatchUpComplete, TopicsLive, GroupMessages, WelcomeMessages) while hiding keepalive frames from callers.
• Dropping BidiConnection aborts the actor and closes the stream; the module is excluded from wasm32 targets.

^{Macroscope summarized 893718b.}

[claude]

Claude finished @tylerhawkes's task —— View job

---

Code Review — June 26, 2026, 19:35 UTC

• Read and analyze the new connection.rs file
• Review code quality, potential bugs, performance, security, and test coverage
• Post review feedback

---

Review Feedback

Overall, this is excellent code — well-documented, properly abstracted, and thoroughly tested. The ownership-based teardown design is sound. A few observations:

Minor Issues

1. Potential atomicity gap in probe_nonce (connection.rs:194-197)

   • fetch_add(1, Ordering::Relaxed).wrapping_add(1) results in nonces starting at 1 (since 0.fetch_add(1) returns 0, then +1 → 1). If this is intentional to avoid nonce 0, consider a comment. If not, just use fetch_add(1, ...) directly.
   • Using Relaxed ordering is fine here since nonce uniqueness doesn't require synchronization, but consider AcqRel if there are subtle ordering dependencies I missed.

2. Event buffer sizing rationale (connection.rs:37)

   • EVENT_BUFFER: usize = 1024 is 16× larger than WIRE_BUFFER. The comment says "large enough that a brief consumer stall doesn't stall wire reads," which is good. Consider whether 1024 is empirically derived or could be tuned down (e.g., 256) if memory pressure becomes a concern at scale. Not urgent, just worth documenting if there's a specific reason for this ratio.

3. Test coverage for malformed TopicsLive (connection.rs:438-448)

   • The code correctly validates topics via Topic::try_from and skips invalid ones with a warn. However, I don't see a test for this path (e.g., a server sending an empty topic byte array or invalid kind byte). Consider adding a regression test like skips_malformed_topics_in_topics_live to verify the filter_map logic and ensure the warning fires.

4. Drop documentation (connection.rs:226-234)

   • The Drop impl correctly aborts the actor to prevent zombie keepalives. Minor: the comment "abort the actor so it cannot keep auto-ponging" could clarify that this also closes both wire directions immediately, which is critical for connection pool management in a multi-shard setup (Phase 2). Readers might miss that detail.

Code Quality

• Excellent: The actor's ownership-based design eliminates the "send into closed stream" class of bugs that Macroscope caught in earlier revisions.
• Excellent: Test suite is comprehensive (11 tests covering teardown, concurrency, backpressure, timeouts, wire-order preservation).
• Good: Inline comments explain non-obvious choices (e.g., why try_send for the initial mutate, why reserve() branch prevents blocking).

Performance

• The reserve() pattern (connection.rs:324-341) correctly prevents inbound stalls during wire backpressure. This is critical for keepalive correctness.
• Bounded buffers are appropriately sized. The MAX_PENDING_FRAMES cap (connection.rs:52) provides backpressure without unbounded growth.

Security

• No secrets are logged or exposed.
• The actor aborts on Drop, preventing resource leaks (server-side subscription holds).
• Malformed server frames (unknown versions, invalid topics) are handled defensively without panicking.

Test Coverage

• 95.67% patch coverage is strong. The 21 missing lines (per Codecov) are likely edge cases in error paths or defensive branches (e.g., the None arm at line 471).
• All critical paths (ownership teardown, probe correlation, auto-pong, wire-order) are tested.
• Mock-based tests are clean and don't require a deployed node.

---

Recommendation: Approve after addressing the minor test gap (malformed TopicsLive). The nonce atomicity and buffer sizing are minor polish items, not blockers.

---

[codecov]

Codecov Report

❌ Patch coverage is 94.46367% with 32 lines in your changes missing coverage. Please review.
✅ Project coverage is 84.54%. Comparing base (82a7c3b) to head (893718b).

Files with missing lines	Patch %	Lines
crates/xmtp_api_d14n/src/queries/v3/connection.rs	94.46%	32 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3772      +/-   ##
==========================================
+ Coverage   84.40%   84.54%   +0.13%     
==========================================
  Files         409      410       +1     
  Lines       60138    60716     +578     
==========================================
+ Hits        50759    51331     +572     
- Misses       9379     9385       +6     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[macroscopeapp]

Approvability

Verdict: Needs human review

3 blocking correctness issues found. This PR introduces a substantial new BidiConnection actor (~1000 lines) with complex async networking logic for bidirectional stream management, liveness probing, and backpressure handling. New features of this scope introducing significant runtime behavior warrant human review, and there's an unresolved comment about potential blocking behavior in the actor loop.

^{You can customize Macroscope's approvability policy. Learn more.}

[insipx]

one thing we'll want to figure out with BiDi is how we handle any potential difference in V3/d14n APIs while also handling MLS logic. Mostly v3 and d14n differ in how they handle their proto types/what they do with those types handled by the Extractor system in d14n. That's where we define common Stream types for each version.

It could make the delineation between MLS and Bidi stream handling easier than if raw Bidi stream handling is implemented in xmtp_api_d14n or even its own stream crate and xmtp_mls is given only a small interface surface to interact with it, then we can test both the mls and stream parts separately, and also compose the interface xmtp_mls receives with the xmtp_proto::types unified GroupMessage/Group etc types to avoid dealing with raw protobuf types in the more easily complicated MLS code, and also avoid worrying about manually reconciling cursor schemes or other differences between v3/d14n (which would then be handled by the extractors)

[insipx]

these message types are where we will need to be careful about the v3/d14n differences. based on client type we can re-use the d14n or v3 extractor schemes here. namely, d14n message return from the D14nClientStreams include logic for the iceboxing & dependency resolution based on depends_on that v3 does not include. If we are going to use this bidi stream across many clients, we will need a new cursor store implementation that is decoupled from MLS (and potentially includes its own on-disk storage solution or sqlite db?) in order to keep the icebox consistent and avoid depending on any single client store

but maybe i'm speaking too soon and need more info/design on how individual clients will interact with the bidi actor, maybe we can pass along the proto and do the per-version processing later on?

[tylerhawkes]

Agree — taking your "pass the proto along, process per-version later" path, with the extractor scheme + MLS-decoupled cursor store designed into the d14n follow-up after this lands.

[tylerhawkes]

@insipx +1 — pushing bidi handling down behind a small unified-types interface to xmtp_mls is the right shape, and I'll do it as part of the d14n lane rather than this PR.

[macroscopeapp]

🟡 Medium subscriptions/bidi.rs:186

When the server closes the bidi stream first, run_actor stops and next() returns None, but BidiSubscription.outbound is still alive so later mutate()/ping() calls can still return Ok(()) and enqueue frames into the dead request stream. Those requests are silently lost even though the API implies send failure means the stream is dead. Consider detecting when the actor stops and failing subsequent sends, or document that callers must stop using the subscription after next() returns None.

-async fn run_actor<S, E>(
+use tokio::sync::mpsc::error::SendError;
+
+async fn run_actor<S, E>(
     mut inbound: S,
     outbound: mpsc::Sender<SubscribeRequest>,
     events: mpsc::Sender<BidiEvent>,
+    outbound_closed: tokio::sync::watch::Sender<bool>,
 ) where
     S: futures::Stream<Item = Result<SubscribeResponse, E>> + Unpin,
     E: std::fmt::Display,
 {

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @crates/xmtp_mls/src/subscriptions/bidi.rs around lines 186-196:

When the server closes the bidi stream first, `run_actor` stops and `next()` returns `None`, but `BidiSubscription.outbound` is still alive so later `mutate()`/`ping()` calls can still return `Ok(())` and enqueue frames into the dead request stream. Those requests are silently lost even though the API implies send failure means the stream is dead. Consider detecting when the actor stops and failing subsequent sends, or document that callers must stop using the subscription after `next()` returns `None`.

[macroscopeapp]

🟡 Medium subscriptions/bidi.rs:257

TopicsLive at line 259 uses Topic::from_bytes without validation, which its docs state can cause undefined behavior on invalid layouts. A server can send an empty topic or unknown kind byte, and this code will emit it as a BidiEvent::TopicsLive. Any later use of that Topic (e.g., Debug, Display, or kind()) will panic because those paths call self.inner[0] or TopicKind::try_from(..).expect(..) on unvalidated data. Consider filtering out or erroring on invalid topics instead of passing them through.

-            Some(Response::TopicsLive(live)) => {
-                let event = BidiEvent::TopicsLive {
-                    topics: live.topics.into_iter().map(Topic::from_bytes).collect(),
-                };
+            Some(Response::TopicsLive(live)) => {
+                let topics: Vec<Topic> = live
+                    .topics
+                    .into_iter()
+                    .filter_map(|bytes| Topic::try_from_bytes(&bytes).ok())
+                    .collect();
+                let event = BidiEvent::TopicsLive { topics };
                 if emit(&events, event).await {
                     break;
                 }
             }

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @crates/xmtp_mls/src/subscriptions/bidi.rs around lines 257-264:

`TopicsLive` at line 259 uses `Topic::from_bytes` without validation, which its docs state can cause undefined behavior on invalid layouts. A server can send an empty topic or unknown kind byte, and this code will emit it as a `BidiEvent::TopicsLive`. Any later use of that `Topic` (e.g., `Debug`, `Display`, or `kind()`) will panic because those paths call `self.inner[0]` or `TopicKind::try_from(..).expect(..)` on unvalidated data. Consider filtering out or erroring on invalid topics instead of passing them through.

[macroscopeapp]

🟡 Medium v3/connection.rs:486

emit() awaits events.send(), which blocks the entire actor loop when EVENT_BUFFER fills. While blocked, the actor stops processing commands.recv(), so a Command::Probe times out even on a healthy connection—particularly after a resume that queues many events before the consumer drains them. Consider decoupling event emission from command processing, e.g. via a select! that concurrently handles commands while awaiting event sends.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @crates/xmtp_api_d14n/src/queries/v3/connection.rs around line 486:

`emit()` awaits `events.send()`, which blocks the entire actor loop when `EVENT_BUFFER` fills. While blocked, the actor stops processing `commands.recv()`, so a `Command::Probe` times out even on a healthy connection—particularly after a resume that queues many events before the consumer drains them. Consider decoupling event emission from command processing, e.g. via a `select!` that concurrently handles commands while awaiting event sends.

[insipx]

TIL abt the reserve() api

[insipx]

small nit: shouldn't "inbound" be "wire_in" for consistency -- afaiu the inbound = messages we're getting and wire_out is messages we're sending off into bidi, where wire terminology is differentiating it from the in-mem message channels