feat(xmtp_api_d14n): generalize XIP-83 bidi over a binding + d14n backend

[tylerhawkes]

Prerequisite #3789 has merged; this now targets main directly.

Generalizes the XIP-83 bidirectional subscription connection over a BidiBinding trait so the backend-agnostic control core lives once, and adds the d14n backend.

What this does

• Connection<B: BidiBinding> (queries/bidi.rs) — the control core (probe + timeout, the non-blocking select loop, the give-up cap, finish()/half-close + best-effort flush, and ownership teardown) now lives once, generic over a backend.
• BidiBinding — each backend supplies only its wire vocabulary: request/response/mutate/message types, frame construction, and a handle() that classifies an inbound response into an Inbound instruction.
• v3 binding (queries/v3/connection.rs) — thin: mls_v1 frames, messages surfaced as raw proto. Its public API (BidiConnection, BidiEvent) is preserved via type aliases, so the feat(xmtp_api_d14n): XIP-83 bidi half-close + live integration tests #3789 tests are unchanged.
• d14n binding (queries/d14n/connection.rs) — a stateless, pure extractor: it demultiplexes each delivered OriginatorEnvelope batch by topic kind and runs the matching extractor into the unified xmtp_proto::types::{GroupMessage, WelcomeMessage}, surfacing them in wire order. Symmetric with the v3 binding — no reordering happens at the transport.
• TrackedStatsClient::inner() — lets the d14n test reach the concrete D14nClient through the stats wrapper.

Ordering

The d14n binding deliberately does not reorder envelopes (no icebox / cursor pipeline). Rationale:

• Per XIP-49 §1.3, the broadcast network is not required to totally order. The payloads that do need strict ordering — commit and identity messages — are chain-anchored to a single fixed originator (Originators::MLS_COMMITS = 0), so MLS's epoch chain is the real dependency and gates processing natively: an out-of-order application message just waits for its epoch.
• XIP-49 §3.3.5 makes cross-originator causal ordering explicitly optional ("clients are free to define an ordering strategy… for additional trustlessness… at the cost of additional complexity").

A single per-process transport is the wrong layer for that. If a consumer wants trustless causal ordering it belongs as a durable client-level icebox — the server-streaming path's OrderedStream is left untouched for exactly that use. This keeps the d14n binding a dumb envelope shoveler, symmetric with v3.

Tests

• 4 d14n live integration tests round-tripping to local xmtpd (mirror the v3 ones): happy-path catch-up → live marker → live stream, history-only, and history-only half-close (bounded sync).
• The shared bidi.rs carries the same half-close behavior landed in feat(xmtp_api_d14n): XIP-83 bidi half-close + live integration tests #3789 — non-blocking try_reserve flush (a stalled transport can't wedge the half-close) and deterministic post-finish Closed — covered by unit tests in bidi.rs and v3/connection.rs.
• Malformed-TopicsLive topics are logged with their bytes (capped at 33) before being skipped; the hex preview is built only on the error path.

🤖 Generated with Claude Code

Note

Generalize XIP-83 bidi subscription over a pluggable binding and d14n backend

• Extracts the bidirectional subscription actor loop into a reusable Connection<B> generic in bidi.rs, with a BidiBinding trait defining wire vocabulary (frame constructors, inbound classifier) for each backend.
• Adds a d14n binding in d14n/connection.rs that opens a /xmtp.xmtpv4.message_api.QueryApi/Subscribe stream, handles ping/pong/started/catchup/topics-live/envelope responses, and emits GroupMessage/WelcomeMessage events.
• Rewrites the v3 binding in v3/connection.rs as a V3Binding impl delegating to the shared core, removing the old in-file actor and related types.
• Adds integration tests in d14n_bidi_tests.rs covering welcome delivery, catch-up ordering, live streaming, history-only behavior, and half-close semantics against a real d14n backend.
• Behavioral Change: malformed TopicsLive entries are now skipped with a warning instead of terminating the connection; envelopes that fail decoding are skipped per-envelope rather than aborting the batch.

^{Macroscope summarized 00a7bad.}

[claude]

Claude finished @tylerhawkes's task —— View job

---

Code Review Complete

Review Date: July 1, 2026, 7:00 PM UTC

---

I've reviewed this PR focusing on areas not already covered by existing review comments. Overall, the architecture is well-designed with clear separation of concerns via the BidiBinding trait. Here are the key findings:

✅ Strengths

1. Strong abstraction design: The BidiBinding trait effectively decouples backend-specific wire formats from the core control logic, enabling code reuse between v3 and d14n backends.

2. Comprehensive test coverage: The d14n integration tests in d14n_bidi_tests.rs cover the critical paths well:

   • Live welcome delivery
   • Catch-up ordering with concurrent messages
   • History-only mode
   • Bounded sync with half-close

3. Clear ownership model: The actor pattern with single ownership of both wire halves ensures clean teardown semantics.

4. Good error handling: The binding's per-envelope extraction (lines 144-172 in d14n/connection.rs) correctly isolates failures so one bad envelope doesn't sink the entire batch.

🟡 Issues to Address

1. Unresolved Macroscope comments warrant attention

The following medium-severity issues flagged by Macroscope appear to still be unresolved and deserve human review:

• bidi.rs:628 (drain_after_finish): The flush uses try_reserve() which drops accepted frames under transient backpressure. This could silently discard a previously-accepted mutate() that returned Ok. Consider using reserve().await with a timeout budget instead.

• d14n/connection.rs:183 (extract_messages): The _group_failures and _welcome_failures from try_consume are silently discarded. When extraction fails, messages are lost with no visibility. Adding tracing::warn! for non-empty failure vectors would surface these drops.

2. Test coverage gap: mid-stream mutate()

The integration tests only exercise the initial subscription in open_group_history_sub(). There's no test for calling mutate() mid-stream to add/remove topics, which is a supported feature (line 232 in bidi.rs). Consider adding a test that:

• Opens a subscription to topic A
• Receives some messages
• Calls mutate() to add topic B
• Verifies delivery from both topics

3. Minor: missing documentation on inner()

In api_stats.rs:42, the inner() method has good inline comments explaining why it's test-gated, but the public API lacks a doc comment. Consider adding:

/// Returns a reference to the wrapped client.
///
/// **Test-only**: This bypasses stat tracking and exists solely to let
/// integration tests reach concrete client types for bidi connection setup.
#[cfg(any(test, feature = "test-utils"))]
pub fn inner(&self) -> &C {

📊 Performance Considerations

• The WIRE_BUFFER (64), COMMAND_BUFFER (64), and EVENT_BUFFER (1024) constants seem reasonable for the expected workload.
• The non-blocking try_send/try_reserve pattern in the actor loop correctly prevents wire backpressure from stalling liveness (auto-pong), which is critical for keepalive.
• The MAX_PENDING_FRAMES cap (128) provides a safety valve against unbounded memory growth.

🔒 Security

No new security concerns beyond what's already been reviewed. The binding is stateless and doesn't introduce new attack surfaces.

Summary

This is solid work with a well-thought-out architecture. The two unresolved Macroscope comments (silent frame dropping in drain_after_finish and unreported extraction failures) are the main items I'd prioritize addressing. The test coverage gap for mid-stream mutate() is a nice-to-have but not blocking.

---

[macroscopeapp]

🟡 Medium queries/bidi.rs:367

When the wire is wedged, pending is allowed to grow past WIRE_BUFFER (up to MAX_PENDING_FRAMES), but the commands.recv() gate in run_actor only blocks new commands once pending.len() >= WIRE_BUFFER. So once pending crosses WIRE_BUFFER, the actor stops reading commands entirely while pending still holds frames below the MAX_PENDING_FRAMES cap — a queued Command::Finish is never read, so the request half never closes and next() hangs indefinitely instead of terminating. Consider capping pending at WIRE_BUFFER (or lowering MAX_PENDING_FRAMES to match) so the command gate and the backlog limit stay aligned.

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

In file @crates/xmtp_api_d14n/src/queries/bidi.rs around line 367:

When the wire is wedged, `pending` is allowed to grow past `WIRE_BUFFER` (up to `MAX_PENDING_FRAMES`), but the `commands.recv()` gate in `run_actor` only blocks new commands once `pending.len() >= WIRE_BUFFER`. So once `pending` crosses `WIRE_BUFFER`, the actor stops reading commands entirely while `pending` still holds frames below the `MAX_PENDING_FRAMES` cap — a queued `Command::Finish` is never read, so the request half never closes and `next()` hangs indefinitely instead of terminating. Consider capping `pending` at `WIRE_BUFFER` (or lowering `MAX_PENDING_FRAMES` to match) so the command gate and the backlog limit stay aligned.

[macroscopeapp]

Approvability

Verdict: Needs human review

3 blocking correctness issues found. New feature introducing d14n bidirectional subscription capability with substantial new logic. Multiple unresolved medium-severity comments identify potential issues: command starvation causing hangs, silent frame drops during drain, and silently discarded extraction failures. Human review recommended to address these concerns.

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

[codecov]

Codecov Report

❌ Patch coverage is 90.56974% with 48 lines in your changes missing coverage. Please review.
✅ Project coverage is 84.56%. Comparing base (58a583e) to head (00a7bad).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
crates/xmtp_api_d14n/src/queries/bidi.rs	86.97%	37 Missing ⚠️
...rates/xmtp_api_d14n/src/queries/d14n/connection.rs	93.93%	10 Missing ⚠️
crates/xmtp_api_d14n/src/queries/v3/connection.rs	98.24%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3804      +/-   ##
==========================================
+ Coverage   84.54%   84.56%   +0.01%     
==========================================
  Files         410      412       +2     
  Lines       60843    61209     +366     
==========================================
+ Hits        51440    51759     +319     
- Misses       9403     9450      +47     

☔ 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.

[insipx]

small nit/non-blocker: one small improvement could be to make the bidi into an Endpoint which keeps path constants in one spot. Endpoints have less meaning since the HTTP1 shim crate was removed but it does remain minimally useful to separate some builder arguments and clean up client callsites/paths

[macroscopeapp]

🟡 Medium d14n/connection.rs:183

extract_messages discards the _group_failures and _welcome_failures returned by try_consume, so when an individual envelope fails extraction (e.g. malformed MLS payload bytes) that delivered message is silently dropped and the function still returns Ok((group, welcome)) containing only the successful items. The consumer receives a seemingly complete delivery with no indication that messages were lost. Consider logging the failures (as extract does for unreadable topics) so dropped envelopes are surfaced rather than silently swallowed.

Suggested change

	let (group, _group_failures) = group_envs.try_consume::<GroupMessageExtractor>()?;
	let (welcome, _welcome_failures) = welcome_envs.try_consume::<WelcomeMessageExtractor>()?;
	Ok((group, welcome))
	}
	let (group, group_failures) = group_envs.try_consume::<GroupMessageExtractor>()?;
	let (welcome, welcome_failures) = welcome_envs.try_consume::<WelcomeMessageExtractor>()?;
	if !group_failures.is_empty() {
	tracing::warn!("d14n bidi: {} group envelope(s) failed extraction", group_failures.len());
	}
	if !welcome_failures.is_empty() {
	tracing::warn!("d14n bidi: {} welcome envelope(s) failed extraction", welcome_failures.len());
	}
	Ok((group, welcome))

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

In file @crates/xmtp_api_d14n/src/queries/d14n/connection.rs around lines 183-186:

`extract_messages` discards the `_group_failures` and `_welcome_failures` returned by `try_consume`, so when an individual envelope fails extraction (e.g. malformed MLS payload bytes) that delivered message is silently dropped and the function still returns `Ok((group, welcome))` containing only the successful items. The consumer receives a seemingly complete delivery with no indication that messages were lost. Consider logging the failures (as `extract` does for unreadable topics) so dropped envelopes are surfaced rather than silently swallowed.