From 00a7bad8accd58e4937851bcc19a2930f3a7fb71 Mon Sep 17 00:00:00 2001 From: Tyler Hawkes Date: Fri, 26 Jun 2026 18:44:50 -0600 Subject: [PATCH] feat(xmtp_api_d14n): generic bidi Connection + d14n binding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Generalize the XIP-83 bidi connection over a backend BidiBinding so the control loop (probe/timeout, non-blocking select, give-up cap, finish/half-close, ownership teardown) lives once: - queries/bidi.rs: the backend-agnostic Connection, Event, Inbound, BidiBinding trait, and the actor. - v3 becomes a thin V3Binding; BidiConnection/BidiEvent are type aliases over the generic core, so the merged v3 actor's public API and behavior are unchanged (13 unit + 4 live integration tests still green). - d14n binding (queries/d14n/connection.rs): transport over the xmtpv4 QueryApi/Subscribe RPC; OriginatorEnvelope delivery decoded into unified xmtp_proto::types::{GroupMessage, WelcomeMessage} via the existing extractor + dependency-ordering (resolve/icebox) + per-topic vector-cursor pipeline. Uses a fresh connection-local cursor store (decoupled from the MLS-coupled durable store). open() takes the D14nClient via a new TrackedStatsClient::inner(). - 4 d14n live integration tests round-trip to the local xmtpd (xmtpd #2020): welcome, catch-up/marker/live ordering, history_only, bounded-sync half-close — validating the full envelope-extraction pipeline end to end. --- crates/xmtp_api_d14n/src/queries/api_stats.rs | 11 + crates/xmtp_api_d14n/src/queries/bidi.rs | 778 +++++++++++++++++ crates/xmtp_api_d14n/src/queries/d14n.rs | 5 + .../src/queries/d14n/connection.rs | 338 ++++++++ crates/xmtp_api_d14n/src/queries/mod.rs | 5 + .../src/queries/v3/connection.rs | 790 +++--------------- .../src/subscriptions/d14n_bidi_tests.rs | 402 +++++++++ crates/xmtp_mls/src/subscriptions/mod.rs | 9 +- 8 files changed, 1661 insertions(+), 677 deletions(-) create mode 100644 crates/xmtp_api_d14n/src/queries/bidi.rs create mode 100644 crates/xmtp_api_d14n/src/queries/d14n/connection.rs create mode 100644 crates/xmtp_mls/src/subscriptions/d14n_bidi_tests.rs diff --git a/crates/xmtp_api_d14n/src/queries/api_stats.rs b/crates/xmtp_api_d14n/src/queries/api_stats.rs index d4d2455402..31eaf37a93 100644 --- a/crates/xmtp_api_d14n/src/queries/api_stats.rs +++ b/crates/xmtp_api_d14n/src/queries/api_stats.rs @@ -31,6 +31,17 @@ impl TrackedStatsClient { identity_stats: Default::default(), } } + + /// The wrapped client — a test-stage escape hatch so the bidi integration + /// tests can reach the concrete `D14nClient` and open a `Connection`. + /// Test-gated on purpose: it bypasses everything this wrapper tracks, and it + /// cannot serve production anyway — the production stack holds a type-erased + /// `Arc`, which has no bidi surface. The client-integration + /// phase adds the real production route instead of widening this hole. + #[cfg(any(test, feature = "test-utils"))] + pub fn inner(&self) -> &C { + &self.inner + } } #[xmtp_common::async_trait] diff --git a/crates/xmtp_api_d14n/src/queries/bidi.rs b/crates/xmtp_api_d14n/src/queries/bidi.rs new file mode 100644 index 0000000000..a5bf3f598d --- /dev/null +++ b/crates/xmtp_api_d14n/src/queries/bidi.rs @@ -0,0 +1,778 @@ +//! Backend-agnostic XIP-83 bidirectional subscription *connection* (native-only). +//! +//! This is the control core shared by the v3 and d14n backends: it owns a single +//! bidi stream end-to-end and is the **sole writer** of the request half. It +//! auto-answers server `Ping`s, correlates client liveness probes, multiplexes +//! caller commands onto the wire, and surfaces only real subscription events — +//! keepalive never reaches the consumer. +//! +//! Everything backend-specific (wire types, frame construction, and turning an +//! inbound frame into consumer events — including d14n's `OriginatorEnvelope` +//! extraction) lives behind the [`BidiBinding`] trait. +//! The control logic — probe + timeout, the non-blocking select loop, the +//! give-up cap, `finish()`/half-close, and ownership teardown — lives here, once. +//! +//! Owning *both* halves is the point. The actor holds the wire-outbound sender +//! and the inbound stream; when inbound dies it drops both, so the request half +//! tears down with it and any later `mutate`/`probe` fails with `Closed` by +//! channel ownership — never by silently enqueueing into a stream nothing reads. + +use std::collections::{HashMap, VecDeque}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Duration; + +use futures::StreamExt; +use futures::stream::BoxStream; +use tokio::sync::{mpsc, oneshot}; +use xmtp_common::{AbortHandle, StreamHandle}; +use xmtp_proto::types::{Topic, TopicKind}; + +/// Wire-outbound depth. The actor is the sole writer; a transport that stops +/// draining backs frames up here first, then in the actor's `pending` queue, +/// until [`MAX_PENDING_FRAMES`] declares the wire wedged and the actor gives up. +pub(crate) const WIRE_BUFFER: usize = 64; +/// Caller→actor command depth. +const COMMAND_BUFFER: usize = 64; +/// Actor→caller event depth; large enough that a brief consumer stall doesn't +/// stall wire reads (and thus pong liveness). +const EVENT_BUFFER: usize = 1024; +/// XIP-83 client req 2 fallback keepalive, used until the server's `Started` +/// frame advertises its own cadence. +pub(crate) const DEFAULT_KEEPALIVE_MS: u32 = 30_000; +/// `N` from XIP-83 client req 2 (recommended 2–3): the *default* probe deadline +/// is this many keepalive intervals — generous enough not to false-positive a +/// slow-but-live link. Latency-sensitive callers (e.g. a notification handler) +/// pass a much smaller bound to [`Connection::probe_within`]. +pub(crate) const PROBE_TIMEOUT_MULTIPLIER: u32 = 3; +/// Hard cap on the outbound backlog. Past this, the wire has been wedged long +/// enough that buffering more is pointless — the transport isn't draining the +/// request half, so the link is effectively dead — and the actor gives up, +/// tearing down so the consumer re-opens from cursors on a fresh stream. +/// Reached only under a sustained stall while commands or auto-pongs keep +/// arriving; tearing down here (rather than parking callers forever) is also +/// what keeps a queued `Finish` from starving behind a wedged wire. +pub(crate) const MAX_PENDING_FRAMES: usize = WIRE_BUFFER * 2; +/// Total time the post-`finish` drain will wait for wire capacity while flushing +/// already-accepted frames. Generous for a transient stall (a draining transport +/// frees a slot in milliseconds) yet bounded, so a wedged transport can't hold +/// the half-close hostage — see [`drain_after_finish`]. +const DRAIN_FLUSH_BUDGET: Duration = Duration::from_secs(1); + +/// Backend-specific wire vocabulary for a bidi subscription. The control core is +/// generic over this: a binding names the wire request/response types and the +/// per-backend `Mutate` and message types, builds outbound frames, and — via +/// [`BidiBinding::handle`] — classifies an inbound response into an [`Inbound`] +/// instruction (which is where d14n runs its per-envelope extractors). +pub trait BidiBinding: Send + 'static { + /// Outbound wire frame (client → server). + type Request: Send + 'static; + /// Inbound wire frame (server → client). + type Response: Send + 'static; + /// The backend's `Mutate` payload (v3: single `id_cursor`; d14n: vector cursor). + type Mutate: Send; + /// Consumer-facing group message (v3: raw proto; d14n: unified, post-extract). + type GroupMessage: Send; + /// Consumer-facing welcome message (v3: raw proto; d14n: unified, post-extract). + type WelcomeMessage: Send; + + /// Wrap a `Mutate` as an outbound request frame. + fn mutate_frame(mutate: Self::Mutate) -> Self::Request; + /// A client `Ping` request frame (liveness probe). + fn ping_frame(nonce: u64) -> Self::Request; + /// A `Pong` request frame answering a server `Ping`. + fn pong_frame(nonce: u64) -> Self::Request; + + /// Classify one inbound response into an actor instruction. An associated + /// fn, like the frame constructors: bindings are stateless by design — any + /// stateful ordering or cursor tracking belongs to the consumer, not a + /// single per-process transport (see the d14n binding's module docs). + fn handle(response: Self::Response) -> Inbound; +} + +/// What an inbound frame means to the control core, after the binding classifies +/// it. Liveness (`Ping`/`Pong`) is handled by the actor and never surfaced; +/// everything else becomes a consumer [`Event`]. +pub enum Inbound { + /// Server ping — the actor auto-pongs this nonce. + Ping(u64), + /// Server pong — the actor resolves the matching client probe. + Pong(u64), + /// A single consumer event to surface (handshake / markers). + Emit(Event), + /// A delivery batch — the actor emits `GroupMessages(group)` then + /// `WelcomeMessages(welcome)`, each only if non-empty. Kept distinct from + /// `Emit` so a frame carrying both kinds needs no extra allocation. + Messages { group: Vec, welcome: Vec }, + /// Nothing to do — unknown version or an informational/undecodable frame. + Skip, +} + +/// Events surfaced to the consumer, in wire order. `Ping`/`Pong` never appear — +/// liveness lives entirely inside the actor. Generic over the backend's group +/// and welcome message types. +#[derive(Debug, Clone, PartialEq)] +pub enum Event { + /// First frame on every stream; carries the server's keepalive cadence. + Started { + keepalive_interval_ms: u32, + capabilities: Vec, + }, + /// A `Mutate`'s adds are fully caught up; echoes the Mutate's `mutate_id`. + CatchUpComplete { + mutate_id: u64, + }, + /// These topics just crossed from catch-up to live. + TopicsLive { + topics: Vec, + }, + GroupMessages(Vec), + WelcomeMessages(Vec), +} + +#[derive(Debug, thiserror::Error)] +pub enum BidiError { + #[error("the bidi connection is closed; re-open and resume from durable cursors")] + Closed, + #[error("liveness probe timed out; treat the link as dead, drop it, and re-open")] + ProbeTimedOut, +} + +/// Submitted by the handle, performed by the actor (the sole wire writer). +enum Command { + Mutate(B::Mutate), + /// A client liveness probe. The actor sends the `Ping` and fires `ack` when + /// the matching `Pong` returns; if the actor exits first, `ack` is dropped + /// and the waiting [`Connection::probe`] resolves to `Closed`. + Probe { + nonce: u64, + ack: oneshot::Sender<()>, + }, + /// Half-close the request half. The actor drops its wire sender so the + /// outbound stream ends (the server sees the half-close), then drains inbound + /// to completion. No further outbound frames are sent after this. + Finish, +} + +/// A handle to one open bidirectional subscription. Writing to the wire is the +/// actor's job; this only submits commands and reads events. Generic over the +/// backend [`BidiBinding`]; the v3 and d14n modules provide concrete aliases. +pub struct Connection { + commands: mpsc::Sender>, + events: mpsc::Receiver>, + probe_nonce: AtomicU64, + /// Latched `true` once a `finish` has *delivered* `Command::Finish` to the + /// actor; monotonic — set once on delivery, never cleared. Lets `mutate`/ + /// `probe` observe `Closed` without racing the actor's teardown of the + /// command channel: after `finish` returns, `Command::Finish` is still in + /// flight and the receiver lives until the actor drains it, so a + /// FIFO-following `mutate` would otherwise be accepted into the buffer and + /// then silently dropped. Latching only on delivery (not before the send) is + /// what makes a cancelled `finish` harmless and concurrent `finish` calls + /// race-free — there is no reset to race. + finished: AtomicBool, + /// The server's advertised keepalive cadence (ms), recorded when + /// [`Self::next`] surfaces `Started`; `0` until then. Drives the default + /// probe deadline so `probe` can self-bound without the caller re-deriving + /// it. A probe issued before `Started` has been consumed falls back to the + /// 30s-derived default, which the probe docs already sanction as safe. + keepalive_ms: u32, + actor: Box, +} + +impl Connection { + /// Open the stream: seed `initial` as the first request frame (it names the + /// initial topic set with per-topic resume cursors; XIP-83 client req 3), + /// then hand the outbound frame stream to `transport` to obtain the inbound + /// frame stream, and spawn the actor. The backend modules wrap this with an + /// ergonomic `open(api, initial)`. + pub(crate) async fn start(initial: B::Mutate, transport: T) -> Result + where + T: FnOnce(BoxStream<'static, B::Request>) -> Fut, + Fut: Future>, + S: futures::Stream> + Send + 'static, + E: std::fmt::Display + Send + 'static, + { + let (wire_out, mut wire_out_rx) = mpsc::channel(WIRE_BUFFER); + let (commands_tx, commands_rx) = mpsc::channel(COMMAND_BUFFER); + let (event_tx, events) = mpsc::channel(EVENT_BUFFER); + + // The first wire frame MUST be this Mutate (XIP-83 req 3). Seed it into + // the fresh, empty wire channel before the transport or the actor can + // write anything else; the receiver is still held here, so a fresh, + // empty, bounded channel can neither be full nor closed — `try_send` + // makes that "cannot block" invariant structural. + wire_out + .try_send(B::mutate_frame(initial)) + .unwrap_or_else(|_| { + unreachable!("send into a fresh, empty, owned channel cannot fail") + }); + + // Wrap the receiver as a `Stream` without pulling in `tokio-stream`: + // `poll_recv` is exactly the poll fn a `Stream` needs. + let outbound = futures::stream::poll_fn(move |cx| wire_out_rx.poll_recv(cx)); + let inbound = transport(Box::pin(outbound)).await?; + + let actor = xmtp_common::spawn( + None, + // Box::pin makes the inbound stream `Unpin` for the select loop without + // requiring the transport's stream type to be `Unpin` itself. + run_actor::(Box::pin(inbound), wire_out, commands_rx, event_tx), + ); + + Ok(Self { + commands: commands_tx, + events, + probe_nonce: AtomicU64::new(0), + finished: AtomicBool::new(false), + keepalive_ms: 0, + actor: actor.abort_handle(), + }) + } + + /// Add/remove subscriptions in place. Awaits a free command slot + /// (backpressure is right for a state change); returns `Closed` once the + /// actor has stopped — the command receiver dies with it. + pub async fn mutate(&self, mutate: B::Mutate) -> Result<(), BidiError> { + if self.finished.load(Ordering::Acquire) { + return Err(BidiError::Closed); + } + self.commands + .send(Command::Mutate(mutate)) + .await + .map_err(|_| BidiError::Closed) + } + + /// Half-close the request half — signal that we are done sending. The + /// outbound stream ends, so `mutate` and `probe` thereafter return `Closed`; + /// any live delivery already in flight keeps arriving until the server closes + /// its side. Opened with a `history_only` Mutate, this is the bounded-sync + /// trigger (XIP-83): the server finishes the in-flight catch-up wave, emits + /// its `TopicsLive` / `CatchUpComplete` markers, and closes the stream, so the + /// consumer drains [`Self::next`] to `None` and stops. Returns `Closed` if the + /// actor has already stopped. + /// + /// Draining to `None` relies on the server actually closing its side. After + /// `finish` there is no `probe` escape hatch (it returns `Closed`), so a + /// consumer that doesn't trust the peer to honor the half-close should bound + /// its post-`finish` [`Self::next`] with a timeout rather than awaiting `None` + /// indefinitely. + /// + /// Not meant to race a concurrent `mutate`/`probe` from another task on the + /// same handle: ordering between two tasks' channel sends is undefined, so a + /// `mutate` racing `finish` may still report `Ok` and then be dropped. The + /// guarantee is for the sequential caller — once `finish` *returns*, a later + /// `mutate`/`probe` from that caller sees `Closed`. + pub async fn finish(&self) -> Result<(), BidiError> { + // Latch closed only *after* the send is delivered. A cancelled `finish` + // (future dropped mid-send) never delivered `Finish`, so it must not + // wedge the handle — leaving the latch untouched is exactly right. And + // because the latch is monotonic (set on delivery, never cleared), two + // concurrent `finish` calls can't race a reset. The same-caller contract + // still holds: a FIFO-following `mutate`/`probe` runs after this store. + self.commands + .send(Command::Finish) + .await + .map_err(|_| BidiError::Closed)?; + self.finished.store(true, Ordering::Release); + Ok(()) + } + + /// Probe the link (e.g. right after the process resumes) with the default + /// deadline: `N ×` the server's advertised keepalive interval (a 30s-derived + /// fallback until `Started` arrives). Resolves `Ok` on the matching `Pong`, + /// `Closed` if the link is already torn down, or `ProbeTimedOut` if no pong + /// arrives in time — the half-open case this whole mechanism exists to catch. + pub async fn probe(&self) -> Result<(), BidiError> { + self.probe_within(self.default_probe_timeout()).await + } + + /// Probe with an explicit deadline. A latency-sensitive caller — say a push + /// notification handler that must decide in a couple of seconds whether to + /// reuse the connection or re-open — passes a bound far below the default. + /// A tight bound trades occasional false `ProbeTimedOut`s for speed, which is + /// safe: re-opening replays from durable cursors and discards duplicates. + /// + /// The timeout covers the whole probe — submitting the ping (so a stalled + /// wire backpressuring the command queue can't park us) and awaiting the + /// pong. We deliberately don't fail-fast on a full command queue: a transient + /// burst of `mutate`s is "busy," not "dead." + pub async fn probe_within(&self, timeout: Duration) -> Result<(), BidiError> { + match tokio::time::timeout(timeout, self.probe_inner()).await { + Ok(result) => result, + Err(_elapsed) => Err(BidiError::ProbeTimedOut), + } + } + + async fn probe_inner(&self) -> Result<(), BidiError> { + if self.finished.load(Ordering::Acquire) { + return Err(BidiError::Closed); + } + let nonce = self + .probe_nonce + .fetch_add(1, Ordering::Relaxed) + .wrapping_add(1); + let (ack, ack_rx) = oneshot::channel(); + self.commands + .send(Command::Probe { nonce, ack }) + .await + .map_err(|_| BidiError::Closed)?; + ack_rx.await.map_err(|_| BidiError::Closed) + } + + pub(crate) fn default_probe_timeout(&self) -> Duration { + let keepalive = match self.keepalive_ms { + 0 => DEFAULT_KEEPALIVE_MS, + ms => ms, + }; + Duration::from_millis(u64::from(keepalive) * u64::from(PROBE_TIMEOUT_MULTIPLIER)) + } + + /// Next event, in wire order. `None` means the connection ended (server + /// close, network death, or reap) — resume from durable cursors on a fresh + /// connection. + /// + /// Only resolves on an event or end-of-stream: a server that goes quiet + /// without closing (a half-open link) leaves this pending. Detect that gap + /// out of band with [`Self::probe`] plus a timeout, never by waiting here. + pub async fn next(&mut self) -> Option> { + let event = self.events.recv().await; + // Record the server's cadence as `Started` passes through, so a probe + // issued in reaction to it already gets the keepalive-derived deadline. + if let Some(Event::Started { + keepalive_interval_ms, + .. + }) = &event + { + self.keepalive_ms = *keepalive_interval_ms; + } + event + } +} + +impl Drop for Connection { + fn drop(&mut self) { + // Abort the actor so it cannot keep auto-ponging — a zombie keepalive + // would hold the server-side subscription open forever. The abort drops + // the actor's wire-outbound and inbound, cancelling the underlying + // request and tearing the stream down in both directions. + self.actor.end(); + } +} + +/// Hand `frame` to the wire if it has room right now, else queue it for the +/// reserve branch to flush. Returns `false` when the actor should give up and tear +/// down: either the wire is already closed (this frame can't be delivered), or the +/// backlog has grown past [`MAX_PENDING_FRAMES`] — a wedged wire we won't buffer +/// behind forever. +/// +/// Reaching the queue means the transport isn't draining the request half — +/// which on a healthy link should essentially never happen — so we warn the +/// first time we fall back to it (not on every frame: an existing backlog skips +/// straight to the queue without re-probing the wire). +#[must_use] +fn enqueue(wire_out: &mpsc::Sender, pending: &mut VecDeque, frame: R) -> bool { + if pending.is_empty() { + match wire_out.try_send(frame) { + Ok(()) => return true, + Err(mpsc::error::TrySendError::Full(frame)) => { + tracing::warn!( + "bidi wire is backpressured (transport not draining the request half); \ + queuing outbound frames" + ); + pending.push_back(frame); + } + // Transport gone. Signal teardown now rather than queue an undeliverable + // frame and lean on the reserve branch to notice on a later `select!` + // iteration — which could accept a few more doomed frames first, since + // `select!` picks ready branches at random. The frame is dropped; a wire + // this dead will never send it, and the consumer re-syncs from durable + // cursors on re-open. + Err(mpsc::error::TrySendError::Closed(_frame)) => return false, + } + } else { + pending.push_back(frame); + } + if pending.len() > MAX_PENDING_FRAMES { + tracing::warn!( + backlog = pending.len(), + "bidi wire wedged past the outbound backlog cap; giving up so the consumer re-opens" + ); + return false; + } + true +} + +/// The single owner of the wire: sole reader of inbound, sole writer of +/// outbound, sole producer of events. Caller commands and server frames are +/// multiplexed onto one outbound FIFO via an internal `pending` queue, drained +/// to the wire by a `reserve()` branch — so no send ever blocks the loop and a +/// busy wire can't delay an auto-pong. (Event sends to the *consumer* do still +/// await: a hopelessly slow consumer backpressures here, the intended path to +/// reap-then-resume — distinct from outbound/wire pressure, which the queue +/// handles.) It ends when the wire ends/errors or the handle goes away; ending +/// drops `wire_out` (the request half closes, tearing the stream down), the +/// `commands` receiver (so `mutate`/`probe` see `Closed`), the `events` sender +/// (so `next` ends), and any outstanding probe acks (so pending `probe`s see +/// `Closed`). One place, every teardown. It also gives up the same way if the +/// outbound backlog blows past [`MAX_PENDING_FRAMES`] — a wedged wire that will +/// never drain, so we tear down and let the consumer re-open. +async fn run_actor( + mut inbound: S, + wire_out: mpsc::Sender, + mut commands: mpsc::Receiver>, + events: mpsc::Sender>, +) where + B: BidiBinding, + S: futures::Stream> + Unpin, + E: std::fmt::Display, +{ + // Outstanding client probes awaiting their `Pong`, keyed by nonce. + let mut probes: HashMap> = HashMap::new(); + // Outbound frames awaiting wire capacity. We drain these through a `reserve()` + // branch below rather than `send().await` in a branch body — an `.await` in a + // `select!` arm blocks the whole loop, so a backed-up wire would stall inbound + // reads and delay auto-pongs (getting us reaped on an otherwise healthy link). + let mut pending: VecDeque = VecDeque::new(); + // Set by `Command::Finish`: leave the main loop and drain inbound to close, + // rather than tear everything down. Distinguishes a deliberate half-close + // from the wire/handle-gone breaks, which fall straight through to teardown. + let mut finished = false; + + loop { + tokio::select! { + // Hand one queued frame to the wire the moment it has capacity, + // concurrently with everything else. This is what keeps a busy wire + // from blocking the loop. Gated on a non-empty queue so we only + // contend for a permit when there's something to flush. + permit = wire_out.reserve(), if !pending.is_empty() => { + match permit { + Ok(permit) => { + permit.send(pending.pop_front().expect("queue is non-empty")); + // Drain as much of the backlog as the wire will take right + // now, so a cleared stall flushes promptly. + while !pending.is_empty() { + match wire_out.try_reserve() { + Ok(permit) => { + permit.send(pending.pop_front().expect("queue is non-empty")); + } + Err(_) => break, // wire full again, or closed + } + } + } + Err(_) => break, // transport gone + } + } + // A command from the handle: mutate, probe, or finish. Never gated: + // a backlog-room gate would starve a queued `Finish` forever on a + // wedged wire (the give-up cap is only checked when a frame is + // enqueued, and a closed gate means nothing is). Backlog growth from + // accepted mutates/probes is bounded by `enqueue`'s + // [`MAX_PENDING_FRAMES`] give-up instead — a wedged wire tears down + // rather than parking callers forever. + cmd = commands.recv() => { + let Some(cmd) = cmd else { + // Every handle dropped — nothing more will be sent. + break; + }; + let queued = match cmd { + Command::Mutate(mutate) => { + enqueue(&wire_out, &mut pending, B::mutate_frame(mutate)) + } + Command::Probe { nonce, ack } => { + // Sweep entries whose probe already timed out client-side + // (their receiver is gone), so a scheduled prober against + // a pong-less peer can't grow the map without bound. + probes.retain(|_, ack| !ack.is_closed()); + probes.insert(nonce, ack); + enqueue(&wire_out, &mut pending, B::ping_frame(nonce)) + } + Command::Finish => { + // Half-close: stop accepting new commands, then flush the + // backlog and drain inbound to close (in `drain_after_finish`). + finished = true; + break; + } + }; + if !queued { + break; // wire wedged past the backlog cap — give up + } + } + // A frame from the server. Always read, so liveness never stalls behind + // outbound pressure. + frame = inbound.next() => { + let Some(frame) = frame else { + break; // inbound ended — the stream is closed + }; + let response = match frame { + Ok(r) => r, + Err(e) => { + tracing::warn!("bidi subscription stream errored: {e}"); + break; + } + }; + match B::handle(response) { + // Liveness is internal: auto-pong / probe-correlate here, never + // surface it to the consumer. + Inbound::Ping(nonce) => { + // Auto-pong: hand it to the wire (or queue it FIFO behind a + // backlog). A busy wire delays but never drops the pong, and + // never blocks us from reading the next frame — but if the + // backlog has blown past the cap, the wire is wedged and we + // give up rather than keep piling pongs into a dead stream. + if !enqueue(&wire_out, &mut pending, B::pong_frame(nonce)) { + break; + } + } + Inbound::Pong(nonce) => { + // Resolve the matching client probe. Unmatched pongs are + // ignored — never surfaced to the consumer. + if let Some(ack) = probes.remove(&nonce) { + let _ = ack.send(()); + } + } + // Consumer-facing frames. The same emit path feeds the live loop + // and the post-`finish` drain, so their event semantics match. + instruction => { + if emit_instruction(&events, instruction).await { + break; + } + } + } + } + } + } + // Resolve any in-flight probes now: dropping the probe-ack senders makes each + // waiting `probe()` see `Closed`. Must happen *before* `drain_after_finish`, + // which can block on `inbound` for a while on a half-open link — a probe must + // never hang on a connection that has already left the live loop. + drop(probes); + // A deliberate half-close drains inbound to close rather than tearing down; + // every other exit reason falls straight through to teardown. + if finished { + drain_after_finish::(inbound, wire_out, pending, commands, &events).await; + } + // One exit log for every reason (wire ended/errored, handle gone, or finished + // draining). The error a caller sees is just `Closed`; the "why" lives here. + tracing::debug!("bidi subscription actor stopped"); + // `wire_out`, `commands`, `events`, and the `pending` queue drop here — that + // *is* the teardown in both directions. +} + +/// Surface the consumer events for one classified inbound frame, returning true +/// if the consumer is gone (the actor should stop). `Ping`/`Pong` must already +/// have been handled by the caller — passing them here is a no-op. Shared by the +/// live loop and the post-`finish` drain so both deliver identical events. +async fn emit_instruction( + events: &mpsc::Sender>, + instruction: Inbound, +) -> bool { + match instruction { + Inbound::Emit(event) => emit(events, event).await, + Inbound::Messages { group, welcome } => { + if !group.is_empty() && emit(events, Event::GroupMessages(group)).await { + return true; + } + if !welcome.is_empty() && emit(events, Event::WelcomeMessages(welcome)).await { + return true; + } + false + } + // Ping/Pong are the caller's job; Skip is a no-op. + Inbound::Ping(_) | Inbound::Pong(_) | Inbound::Skip => false, + } +} + +/// Drain inbound to completion after a half-close. First flush any frames the +/// caller already had accepted (a `mutate` it got `Ok` for, an auto-pong, a +/// probe ping queued under backpressure) so half-close doesn't silently discard +/// them. Then dropping `wire_out` ends the outbound stream (the server sees the +/// half-close, finishes the wave, and closes its side); dropping `commands` +/// makes any late `mutate`/`probe` see `Closed`. We keep surfacing events until +/// the server closes inbound (`next` -> `None`) or the consumer goes away. +/// Liveness frames can't be answered (the wire is gone), so they are ignored; +/// outstanding probes drop with the actor and resolve to `Closed`. +async fn drain_after_finish( + mut inbound: S, + wire_out: mpsc::Sender, + mut pending: VecDeque, + commands: mpsc::Receiver>, + events: &mpsc::Sender>, +) where + B: BidiBinding, + S: futures::Stream> + Unpin, + E: std::fmt::Display, +{ + // Flush the accepted-but-unsent backlog before closing the request half, + // waiting for wire capacity under one shared budget. Bounded on purpose: an + // unbounded `reserve().await` on a stalled transport (wire full, server no + // longer reading the request stream) would block forever, so `drop(wire_out)` + // would never run and the half-close would never reach the server — but a + // purely non-blocking flush drops accepted frames on a *transient* stall a + // healthy transport clears in milliseconds. Frames still unplaced when the + // budget runs out are lost; the consumer re-syncs from durable cursors on + // re-open, and a wedged actor is the strictly worse outcome. + let flush_deadline = tokio::time::Instant::now() + DRAIN_FLUSH_BUDGET; + while let Some(frame) = pending.pop_front() { + match tokio::time::timeout_at(flush_deadline, wire_out.reserve()).await { + Ok(Ok(permit)) => permit.send(frame), + Ok(Err(_)) => break, // wire gone — close rather than block + Err(_elapsed) => { + tracing::warn!( + dropped = pending.len() + 1, + "bidi half-close flush budget exhausted on a stalled wire; \ + dropping the remaining backlog (re-open re-syncs from cursors)" + ); + break; + } + } + } + drop(wire_out); + drop(commands); + while let Some(frame) = inbound.next().await { + let response = match frame { + Ok(r) => r, + Err(e) => { + tracing::warn!("bidi stream errored during finish drain: {e}"); + break; + } + }; + if emit_instruction(events, B::handle(response)).await { + break; // consumer gone + } + } +} + +/// Send one event to the consumer, awaiting a free slot (the backpressure that +/// stalls wire reads once [`EVENT_BUFFER`] fills). Returns true when the +/// consumer is gone and the actor should shut down. (The server's keepalive +/// cadence is recorded handle-side, when [`Connection::next`] surfaces +/// `Started` — every event reaches the consumer through there.) +async fn emit(events: &mpsc::Sender>, event: Event) -> bool { + events.send(event).await.is_err() +} + +/// Parse kind-prefixed wire bytes into typed `Topic`s, validating each kind byte. +/// A malformed topic should never reach us — it means a server or wire-format bug +/// — so log the offending bytes (a topic is at most 33 bytes: a 1-byte kind + a +/// 32-byte id) and skip it rather than kill the connection. Shared by both +/// backends' `TopicsLive` handling. +pub(crate) fn parse_topics(topics: Vec>) -> Vec { + topics + .into_iter() + .filter_map(|bytes| { + // Validate the kind byte against a borrow first, so the hex preview — + // the only allocation — is built solely on the malformed path; the + // common all-valid case does no extra work. + match bytes.first().map(|&b| TopicKind::try_from(b)) { + Some(Ok(_)) => Topic::try_from(bytes).ok(), + outcome => { + let preview = hex::encode(&bytes[..bytes.len().min(33)]); + let reason = match outcome { + Some(Err(e)) => e.to_string(), + _ => "empty topic".to_string(), + }; + tracing::warn!( + topic = %preview, + "skipping malformed TopicsLive topic (server/wire-format bug): {reason}" + ); + None + } + } + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A trivial binding: wire frames are bare `u64`s and nothing is ever + /// surfaced. Enough to exercise the backend-agnostic control logic directly. + struct TestBinding; + + impl BidiBinding for TestBinding { + type Request = u64; + type Response = u64; + type Mutate = u64; + type GroupMessage = (); + type WelcomeMessage = (); + + fn mutate_frame(mutate: u64) -> u64 { + mutate + } + fn ping_frame(nonce: u64) -> u64 { + nonce + } + fn pong_frame(nonce: u64) -> u64 { + nonce + } + fn handle(_response: u64) -> Inbound<(), ()> { + Inbound::Skip + } + } + + /// `finish` is a half-close, not an abort: frames the caller already had + /// accepted under wire backpressure (parked in the actor's `pending` queue, + /// not yet on the wire) must be flushed before the request half closes. Drive + /// `drain_after_finish` directly with a pre-seeded `pending` and an + /// already-closed inbound, so the flush is the only behaviour under test. + #[xmtp_common::test(unwrap_try = true)] + async fn drain_after_finish_flushes_pending_before_closing() { + let (wire_out, mut wire_rx) = mpsc::channel::(WIRE_BUFFER); + let (_commands_tx, commands_rx) = mpsc::channel::>(COMMAND_BUFFER); + let (events, _events_rx) = mpsc::channel::>(EVENT_BUFFER); + + // A distinctively-tagged frame queued as if wire backpressure had parked + // it before the caller half-closed. + let mut pending = VecDeque::new(); + pending.push_back(7_777_u64); + + // Inbound is already closed, so the drain has nothing to read and returns + // as soon as the pending flush completes. + let inbound = futures::stream::empty::>(); + drain_after_finish::(inbound, wire_out, pending, commands_rx, &events) + .await; + + // The queued frame reached the wire instead of being dropped on close, + // and the wire then closes (the actor dropped its sender). + assert_eq!( + wire_rx.recv().await, + Some(7_777), + "the pending frame must be flushed to the wire" + ); + assert_eq!( + wire_rx.recv().await, + None, + "the wire closes once the flush is done" + ); + } + + /// A *wedged* wire must not hold the half-close hostage: the flush waits out + /// [`DRAIN_FLUSH_BUDGET`] for capacity, then drops the backlog and still + /// closes the request half so the server sees the half-close. + #[xmtp_common::test(unwrap_try = true)] + async fn drain_after_finish_bounds_the_flush_on_a_wedged_wire() { + // Capacity 1, pre-filled, receiver never polled: reserve() can never + // succeed, so only the budget can end the flush. + let (wire_out, mut wire_rx) = mpsc::channel::(1); + let (_commands_tx, commands_rx) = mpsc::channel::>(COMMAND_BUFFER); + let (events, _events_rx) = mpsc::channel::>(EVENT_BUFFER); + wire_out.try_send(1_u64).unwrap(); + + let mut pending = VecDeque::new(); + pending.push_back(7_777_u64); + + let inbound = futures::stream::empty::>(); + drain_after_finish::(inbound, wire_out, pending, commands_rx, &events) + .await; + + // The pre-wedged frame is still there; the un-flushable backlog was + // dropped; and crucially the request half closed anyway. + assert_eq!(wire_rx.recv().await, Some(1)); + assert_eq!( + wire_rx.recv().await, + None, + "the request half must close even when the flush budget expires" + ); + } +} diff --git a/crates/xmtp_api_d14n/src/queries/d14n.rs b/crates/xmtp_api_d14n/src/queries/d14n.rs index 9508000c6b..7eb553e2ce 100644 --- a/crates/xmtp_api_d14n/src/queries/d14n.rs +++ b/crates/xmtp_api_d14n/src/queries/d14n.rs @@ -1,5 +1,8 @@ //! Compatibility layer for d14n and previous xmtp_api crate mod client; +// XIP-83 bidi binding (native-only — full-duplex HTTP/2). +#[cfg(not(target_arch = "wasm32"))] +mod connection; mod identity; mod mls; mod streams; @@ -9,6 +12,8 @@ xmtp_common::if_test! { mod test_client; } pub use client::*; +#[cfg(not(target_arch = "wasm32"))] +pub use connection::*; #[cfg(test)] mod test; diff --git a/crates/xmtp_api_d14n/src/queries/d14n/connection.rs b/crates/xmtp_api_d14n/src/queries/d14n/connection.rs new file mode 100644 index 0000000000..e5c48f1e3b --- /dev/null +++ b/crates/xmtp_api_d14n/src/queries/d14n/connection.rs @@ -0,0 +1,338 @@ +//! d14n (xmtpv4 `QueryApi`) binding for the XIP-83 bidirectional subscription +//! connection. +//! +//! The control core lives in [`crate::queries::bidi`]; this supplies the d14n +//! wire vocabulary (`xmtpv4.message_api` frames over the `QueryApi/Subscribe` +//! RPC) and turns each delivered `OriginatorEnvelope` batch into unified +//! `xmtp_proto::types::{GroupMessage, WelcomeMessage}` with one fallible +//! extraction pass per envelope. Native-only. +//! +//! It deliberately does **not** reorder. Per XIP-49 §1.3 the broadcast network is +//! not required to totally order, and the payloads that do need strict ordering +//! (commits, identity updates) are chain-anchored to a single fixed originator +//! (`Originators::MLS_COMMITS`), 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"), so it belongs — if a +//! consumer wants it — as a durable *client-level* icebox, not bolted onto a +//! single per-process transport. This binding just extracts and fans out, exactly +//! like the v3 binding hands its messages straight through. + +use crate::protocol::Envelope; +use crate::queries::bidi::{BidiBinding, Connection, Event, Inbound, parse_topics}; +use futures::StreamExt; +use futures::stream::BoxStream; +use prost::Message; +use prost::bytes::Bytes; +use xmtp_proto::ApiEndpoint; +use xmtp_proto::api::{ApiClientError, Client, XmtpStream}; +use xmtp_proto::types::{GroupMessage, WelcomeMessage}; +use xmtp_proto::xmtp::xmtpv4::envelopes::OriginatorEnvelope; +use xmtp_proto::xmtp::xmtpv4::message_api::{ + Ping, Pong, SubscribeRequest, SubscribeResponse, subscribe_request, subscribe_response, +}; + +use super::D14nClient; + +const SUBSCRIBE_PATH: &str = "/xmtp.xmtpv4.message_api.QueryApi/Subscribe"; + +/// The d14n (xmtpv4 `QueryApi`) wire binding for a bidi subscription. Stateless: +/// it extracts each delivered `OriginatorEnvelope` batch into unified messages, +/// per envelope, and surfaces them in wire order. No cross-originator reordering +/// happens here — see the module docs for why (XIP-49 makes it optional and MLS +/// epoch-gates natively). Symmetric with the v3 binding, which likewise passes +/// its messages straight through. +pub struct D14nBinding; + +/// A d14n bidirectional subscription connection (XIP-83). See [`Connection`]. +pub type D14nBidiConnection = Connection; +/// Events surfaced by a d14n [`D14nBidiConnection`], in wire order. Messages are +/// the unified `xmtp_proto::types` shapes, not raw protobuf. +pub type D14nBidiEvent = Event; + +fn request_frame(request: subscribe_request::v1::Request) -> SubscribeRequest { + SubscribeRequest { + version: Some(subscribe_request::Version::V1(subscribe_request::V1 { + request: Some(request), + })), + } +} + +/// Open the d14n bidi `Subscribe` transport: encode outbound frames and hand the +/// stream to the inner transport `Client`, returning the decoded inbound stream. +// Spans the open handshake (not the stream's lifetime) as `rpc.subscribe_bidi`, +// mirroring the v3 transport — this free fn is the d14n RPC boundary, the same +// layer the other `rpc_span`s live at for unary calls. +#[xmtp_common::rpc_span] +async fn subscribe_bidi( + client: &C, + requests: BoxStream<'static, SubscribeRequest>, +) -> Result, ApiClientError> { + tracing::debug!("opening d14n bidirectional subscription"); + let outbound = requests.map(|frame| Bytes::from(frame.encode_to_vec())); + let response = client + .bidi_stream( + http::Request::builder(), + http::uri::PathAndQuery::from_static(SUBSCRIBE_PATH), + Box::pin(outbound), + ) + .await + .map_err(|e| e.endpoint(SUBSCRIBE_PATH.to_string()))?; + Ok(XmtpStream::new( + response.into_body(), + ApiEndpoint::Path(SUBSCRIBE_PATH.to_string()), + )) +} + +impl BidiBinding for D14nBinding { + type Request = SubscribeRequest; + type Response = SubscribeResponse; + type Mutate = subscribe_request::v1::Mutate; + type GroupMessage = GroupMessage; + type WelcomeMessage = WelcomeMessage; + + fn mutate_frame(mutate: subscribe_request::v1::Mutate) -> SubscribeRequest { + request_frame(subscribe_request::v1::Request::Mutate(mutate)) + } + + fn ping_frame(nonce: u64) -> SubscribeRequest { + request_frame(subscribe_request::v1::Request::Ping(Ping { nonce })) + } + + fn pong_frame(nonce: u64) -> SubscribeRequest { + request_frame(subscribe_request::v1::Request::Pong(Pong { nonce })) + } + + fn handle(response: SubscribeResponse) -> Inbound { + let Some(subscribe_response::Version::V1(v1)) = response.version else { + // A version we did not speak; XIP-83 pins responses to the request + // version, so this is a server bug — skip, don't die. + tracing::warn!("d14n bidi subscription received unknown response version"); + return Inbound::Skip; + }; + use subscribe_response::v1::Response; + match v1.response { + // Liveness is internal; the core auto-pongs and correlates probes. + Some(Response::Ping(ping)) => Inbound::Ping(ping.nonce), + Some(Response::Pong(pong)) => Inbound::Pong(pong.nonce), + Some(Response::Started(started)) => Inbound::Emit(Event::Started { + keepalive_interval_ms: started.keepalive_interval_ms, + capabilities: started.capabilities, + }), + Some(Response::CatchupComplete(complete)) => Inbound::Emit(Event::CatchUpComplete { + mutate_id: complete.mutate_id, + }), + Some(Response::TopicsLive(live)) => Inbound::Emit(Event::TopicsLive { + topics: parse_topics(live.topics), + }), + // Delivery: extract the envelopes into unified messages. + Some(Response::Envelopes(envelopes)) => extract(envelopes.envelopes), + // A future-revision arm: informational frames are safe to skip. + None => Inbound::Skip, + } + } +} + +/// Turn a delivered `OriginatorEnvelope` batch into unified messages, surfaced in +/// wire order (no reordering — see the module docs). One fallible extraction pass +/// per envelope via [`Envelope::group_message`]/[`Envelope::welcome_message`], so +/// a bad envelope — unreadable nesting, or a payload its extractor rejects — is +/// logged and skipped *alone*. It must never sink the valid messages delivered +/// alongside it: the consumer's cursors advance past a dropped batch, so anything +/// discarded here is never re-fetched. +fn extract(envelopes: Vec) -> Inbound { + let mut group = Vec::new(); + let mut welcome = Vec::new(); + for envelope in envelopes { + match envelope.group_message() { + Ok(Some(message)) => { + group.push(message); + continue; + } + // Not a group payload — try the welcome extractor below. + Ok(None) => {} + Err(e) => { + tracing::warn!("d14n bidi: skipping undecodable envelope: {e}"); + continue; + } + } + match envelope.welcome_message() { + Ok(Some(message)) => welcome.push(message), + // Identity-update / key-package delivery is not implemented yet + // (adopted later via XIP-83 `Started` capabilities); warn so a + // subscription to such a topic fails loudly instead of looking + // healthy while its data is dropped. + Ok(None) => tracing::warn!( + "d14n bidi: dropping delivered envelope of an unhandled payload kind" + ), + Err(e) => tracing::warn!("d14n bidi: skipping undecodable welcome envelope: {e}"), + } + } + Inbound::Messages { group, welcome } +} + +impl D14nBidiConnection { + /// Open the d14n stream and send `initial` as the first Mutate (it names the + /// initial topic set with per-topic resume *vector* cursors; XIP-83 client + /// req 3). Uses only the client's transport; the binding is stateless. + pub async fn open( + client: &D14nClient, + initial: subscribe_request::v1::Mutate, + ) -> Result + where + C: Client, + { + Self::start(initial, |outbound| subscribe_bidi(&client.client, outbound)).await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::protocol::extractors::test_utils::TestEnvelopeBuilder; + use xmtp_proto::types::{Topic, TopicKind}; + + fn response(r: subscribe_response::v1::Response) -> SubscribeResponse { + SubscribeResponse { + version: Some(subscribe_response::Version::V1(subscribe_response::V1 { + response: Some(r), + })), + } + } + + fn wire_topic(kind: TopicKind, identifier: &[u8]) -> Vec { + // The production constructor, so these tests can't drift from the real + // kind-prefixed wire layout. + kind.create(identifier).into() + } + + #[xmtp_common::test(unwrap_try = true)] + fn classifies_control_frames() { + let started = D14nBinding::handle(response(subscribe_response::v1::Response::Started( + subscribe_response::v1::Started { + keepalive_interval_ms: 30_000, + capabilities: vec![7], + }, + ))); + assert!(matches!( + started, + Inbound::Emit(Event::Started { + keepalive_interval_ms: 30_000, + .. + }) + )); + + let ping = D14nBinding::handle(response(subscribe_response::v1::Response::Ping(Ping { + nonce: 9, + }))); + assert!(matches!(ping, Inbound::Ping(9))); + + let pong = D14nBinding::handle(response(subscribe_response::v1::Response::Pong(Pong { + nonce: 9, + }))); + assert!(matches!(pong, Inbound::Pong(9))); + + let complete = + D14nBinding::handle(response(subscribe_response::v1::Response::CatchupComplete( + subscribe_response::v1::CatchupComplete { mutate_id: 42 }, + ))); + assert!(matches!( + complete, + Inbound::Emit(Event::CatchUpComplete { mutate_id: 42 }) + )); + } + + #[xmtp_common::test(unwrap_try = true)] + fn parses_topics_live_and_skips_malformed() { + let good = wire_topic(TopicKind::GroupMessagesV1, b"group"); + let bad = vec![0xFF, 1, 2, 3]; // unknown kind byte + let live = D14nBinding::handle(response(subscribe_response::v1::Response::TopicsLive( + subscribe_response::v1::TopicsLive { + topics: vec![good.clone(), bad], + }, + ))); + let Inbound::Emit(Event::TopicsLive { topics }) = live else { + panic!("expected TopicsLive"); + }; + // The malformed topic is dropped; the valid one survives. + assert_eq!(topics, vec![Topic::try_from(good).unwrap()]); + } + + #[xmtp_common::test(unwrap_try = true)] + fn skips_unknown_response_version() { + assert!(matches!( + D14nBinding::handle(SubscribeResponse { version: None }), + Inbound::Skip + )); + } + + #[xmtp_common::test(unwrap_try = true)] + fn empty_envelope_batch_yields_empty_messages() { + let out = D14nBinding::handle(response(subscribe_response::v1::Response::Envelopes( + subscribe_response::v1::Envelopes { envelopes: vec![] }, + ))); + match out { + Inbound::Messages { group, welcome } => { + assert!(group.is_empty()); + assert!(welcome.is_empty()); + } + _ => panic!("expected Messages"), + } + } + + /// One envelope that fails its *extractor* (nesting decodes, payload is + /// rejected — here a 3-byte installation key that can't be an + /// `InstallationId`) must be skipped alone: the valid welcome delivered in + /// the same frame survives. A batch-level short-circuit would drop both. + #[xmtp_common::test(unwrap_try = true)] + fn bad_payload_is_skipped_without_dropping_the_batch() { + let bad = TestEnvelopeBuilder::new() + .with_welcome_message(vec![1, 2, 3]) + .build(); + let good = TestEnvelopeBuilder::new() + .with_welcome_message(vec![7u8; 32]) + .build(); + let out = D14nBinding::handle(response(subscribe_response::v1::Response::Envelopes( + subscribe_response::v1::Envelopes { + envelopes: vec![bad, good], + }, + ))); + match out { + Inbound::Messages { group, welcome } => { + assert!(group.is_empty()); + assert_eq!( + welcome.len(), + 1, + "the valid welcome must survive its bad neighbor" + ); + } + _ => panic!("expected Messages, not a dropped batch"), + } + } + + #[xmtp_common::test(unwrap_try = true)] + fn malformed_envelope_is_skipped_without_dropping_the_batch() { + // Undecodable unsigned bytes fail the very first extraction pass. A + // batch-level short-circuit would turn this into `Skip`, silently + // dropping any valid messages delivered alongside it; the per-envelope + // pass skips only the bad one, so the delivery still resolves to + // `Messages`. + let bad = OriginatorEnvelope { + unsigned_originator_envelope: vec![0xFF; 8], + proof: None, + }; + let out = D14nBinding::handle(response(subscribe_response::v1::Response::Envelopes( + subscribe_response::v1::Envelopes { + envelopes: vec![bad], + }, + ))); + match out { + Inbound::Messages { group, welcome } => { + assert!(group.is_empty()); + assert!(welcome.is_empty()); + } + _ => panic!("expected Messages, not a dropped batch"), + } + } +} diff --git a/crates/xmtp_api_d14n/src/queries/mod.rs b/crates/xmtp_api_d14n/src/queries/mod.rs index bf8491b482..b52bb32897 100644 --- a/crates/xmtp_api_d14n/src/queries/mod.rs +++ b/crates/xmtp_api_d14n/src/queries/mod.rs @@ -1,4 +1,7 @@ mod api_stats; +// Backend-agnostic XIP-83 bidi connection core (native-only — full-duplex HTTP/2). +#[cfg(not(target_arch = "wasm32"))] +mod bidi; mod boxed_streams; mod builder; mod client_bundle; @@ -9,6 +12,8 @@ pub mod stream; mod v3; pub use api_stats::*; +#[cfg(not(target_arch = "wasm32"))] +pub use bidi::*; pub use boxed_streams::*; pub use builder::*; pub use client_bundle::*; diff --git a/crates/xmtp_api_d14n/src/queries/v3/connection.rs b/crates/xmtp_api_d14n/src/queries/v3/connection.rs index 387b2f781a..d1c03fcd72 100644 --- a/crates/xmtp_api_d14n/src/queries/v3/connection.rs +++ b/crates/xmtp_api_d14n/src/queries/v3/connection.rs @@ -1,293 +1,25 @@ -//! XIP-83 bidirectional subscription *connection* (native-only). +//! v3 (MLS API) binding for the XIP-83 bidirectional subscription connection. //! -//! This sits one layer below the application: it owns a single bidi stream -//! end-to-end and is the **sole writer** of the request half. It auto-answers -//! server `Ping`s, correlates client liveness probes, and forwards only real -//! subscription events upward — keepalive never reaches the consumer. -//! -//! Owning *both* halves is the point. The actor holds the wire-outbound sender -//! and the inbound stream; when inbound dies it drops both, so the request half -//! tears down with it and any later `mutate`/`probe` fails with `Closed` by -//! channel ownership — never by silently enqueueing into a stream nothing reads. - -use std::collections::{HashMap, VecDeque}; -use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; -use std::time::Duration; - -use futures::StreamExt; -use tokio::sync::{mpsc, oneshot}; -use xmtp_common::{AbortHandle, StreamHandle}; +//! The control core lives in [`crate::queries::bidi`]; this supplies the v3 wire +//! vocabulary (`mls_v1` frames) and surfaces messages as the raw proto +//! `GroupMessage`/`WelcomeMessage` (v3 carries a single id cursor and the +//! consumer decodes), via the [`BidiBinding`] trait. Native-only. + +use crate::queries::bidi::{BidiBinding, Connection, Event, Inbound, parse_topics}; use xmtp_proto::api_client::XmtpMlsBidiStreams; use xmtp_proto::mls_v1::subscribe_request::v1::Mutate; use xmtp_proto::mls_v1::{ GroupMessage, Ping, Pong, SubscribeRequest, SubscribeResponse, WelcomeMessage, subscribe_request, subscribe_response, }; -use xmtp_proto::types::Topic; - -/// Wire-outbound depth. The actor is the sole writer; if the transport stalls, -/// `send().await` here backpressures the actor, which backpressures the command -/// queue — exactly the signal `probe` turns into a fast `Closed`. -const WIRE_BUFFER: usize = 64; -/// Caller→actor command depth. -const COMMAND_BUFFER: usize = 64; -/// Actor→caller event depth; large enough that a brief consumer stall doesn't -/// stall wire reads (and thus pong liveness). -const EVENT_BUFFER: usize = 1024; -/// XIP-83 client req 2 fallback keepalive, used until the server's `Started` -/// frame advertises its own cadence. -const DEFAULT_KEEPALIVE_MS: u32 = 30_000; -/// `N` from XIP-83 client req 2 (recommended 2–3): the *default* probe deadline -/// is this many keepalive intervals — generous enough not to false-positive a -/// slow-but-live link. Latency-sensitive callers (e.g. a notification handler) -/// pass a much smaller bound to [`BidiConnection::probe_within`]. -const PROBE_TIMEOUT_MULTIPLIER: u32 = 3; -/// Hard cap on the outbound backlog. Past this, the wire has been wedged long -/// enough that buffering more is pointless — the transport isn't draining the -/// request half, so the link is effectively dead — and the actor gives up, -/// tearing down so the consumer re-opens from cursors on a fresh stream. Sits -/// well above the command-gated backlog (`WIRE_BUFFER`), so only a sustained -/// stall (pongs piling up on a half-open link) ever reaches it. -const MAX_PENDING_FRAMES: usize = WIRE_BUFFER * 2; - -/// Events surfaced to the consumer, in wire order. `Ping`/`Pong` never appear — -/// liveness lives entirely inside the actor. -#[derive(Debug, Clone, PartialEq)] -pub enum BidiEvent { - /// First frame on every stream; carries the server's keepalive cadence. - Started { - keepalive_interval_ms: u32, - capabilities: Vec, - }, - /// A `Mutate`'s adds are fully caught up; echoes the Mutate's `mutate_id`. - CatchUpComplete { - mutate_id: u64, - }, - /// These topics just crossed from catch-up to live. - TopicsLive { - topics: Vec, - }, - GroupMessages(Vec), - WelcomeMessages(Vec), -} - -#[derive(Debug, thiserror::Error)] -pub enum BidiError { - #[error("the bidi connection is closed; re-open and resume from durable cursors")] - Closed, - #[error("liveness probe timed out; treat the link as dead, drop it, and re-open")] - ProbeTimedOut, -} - -/// Submitted by the handle, performed by the actor (the sole wire writer). -enum Command { - Mutate(Mutate), - /// A client liveness probe. The actor sends the `Ping` and fires `ack` when - /// the matching `Pong` returns; if the actor exits first, `ack` is dropped - /// and the waiting [`BidiConnection::probe`] resolves to `Closed`. - Probe { - nonce: u64, - ack: oneshot::Sender<()>, - }, - /// Half-close the request half. The actor drops its wire sender so the - /// outbound stream ends (the server sees the half-close), then drains inbound - /// to completion. No further outbound frames are sent after this. - Finish, -} - -/// A handle to one open bidirectional subscription. Writing to the wire is the -/// actor's job; this only submits commands and reads events. -pub struct BidiConnection { - commands: mpsc::Sender, - events: mpsc::Receiver, - probe_nonce: AtomicU64, - /// Latched `true` once a `finish` has *delivered* `Command::Finish` to the - /// actor; monotonic — set once on delivery, never cleared. Lets `mutate`/ - /// `probe` observe `Closed` without racing the actor's teardown of the - /// command channel: after `finish` returns, `Command::Finish` is still in - /// flight and the receiver lives until the actor drains it, so a - /// FIFO-following `mutate` would otherwise be accepted into the buffer and - /// then silently dropped. Latching only on delivery (not before the send) is - /// what makes a cancelled `finish` harmless and concurrent `finish` calls - /// race-free — there is no reset to race. - finished: AtomicBool, - /// The server's advertised keepalive cadence (ms), learned from `Started`; - /// `0` until then. Shared with the actor, which sets it. Drives the default - /// probe deadline so `probe` can self-bound without the caller re-deriving it. - keepalive_ms: Arc, - actor: Box, -} - -impl BidiConnection { - /// Open the stream and send `initial` as the first Mutate (it names the - /// initial topic set with per-topic resume cursors; XIP-83 client req 3). - pub async fn open(api: &A, initial: Mutate) -> Result - where - A: XmtpMlsBidiStreams, - A::SubscribeStream: 'static, - { - let (wire_out, mut wire_out_rx) = mpsc::channel(WIRE_BUFFER); - let (commands_tx, commands_rx) = mpsc::channel(COMMAND_BUFFER); - let (event_tx, events) = mpsc::channel(EVENT_BUFFER); - - // The first wire frame MUST be this Mutate (XIP-83 req 3). Seed it into - // the fresh, empty wire channel before the transport or the actor can - // write anything else; the receiver is still held here, so a fresh, - // empty, bounded channel can neither be full nor closed — `try_send` - // makes that "cannot block" invariant structural. - wire_out - .try_send(request_frame(subscribe_request::v1::Request::Mutate( - initial, - ))) - .expect("send into a fresh, empty channel whose receiver we still hold cannot fail"); - - // Wrap the receiver as a `Stream` without pulling in `tokio-stream`: - // `poll_recv` is exactly the poll fn a `Stream` needs. - let outbound = futures::stream::poll_fn(move |cx| wire_out_rx.poll_recv(cx)); - let inbound = api.subscribe_bidi(Box::pin(outbound)).await?; - - let keepalive_ms = Arc::new(AtomicU32::new(0)); - let actor = xmtp_common::spawn( - None, - run_actor( - Box::pin(inbound), - wire_out, - commands_rx, - event_tx, - keepalive_ms.clone(), - ), - ); - Ok(Self { - commands: commands_tx, - events, - probe_nonce: AtomicU64::new(0), - finished: AtomicBool::new(false), - keepalive_ms, - actor: actor.abort_handle(), - }) - } +/// The v3 (MLS API) wire binding for a bidi subscription. +pub struct V3Binding; - /// Add/remove subscriptions in place. Awaits a free command slot - /// (backpressure is right for a state change); returns `Closed` once the - /// actor has stopped — the command receiver dies with it. - pub async fn mutate(&self, mutate: Mutate) -> Result<(), BidiError> { - if self.finished.load(Ordering::Acquire) { - return Err(BidiError::Closed); - } - self.commands - .send(Command::Mutate(mutate)) - .await - .map_err(|_| BidiError::Closed) - } - - /// Half-close the request half — signal that we are done sending. The - /// outbound stream ends, so `mutate` and `probe` thereafter return `Closed`; - /// any live delivery already in flight keeps arriving until the server closes - /// its side. Opened with `Mutate::history_only`, this is the bounded-sync - /// trigger (XIP-83): the server finishes the in-flight catch-up wave, emits - /// its `TopicsLive` / `CatchUpComplete` markers, and closes the stream, so the - /// consumer drains [`Self::next`] to `None` and stops. Returns `Closed` if the - /// actor has already stopped. - /// - /// Draining to `None` relies on the server actually closing its side. After - /// `finish` there is no `probe` escape hatch (it returns `Closed`), so a - /// consumer that doesn't trust the peer to honor the half-close should bound - /// its post-`finish` [`Self::next`] with a timeout rather than awaiting `None` - /// indefinitely. - /// - /// Not meant to race a concurrent `mutate`/`probe` from another task on the - /// same handle: ordering between two tasks' channel sends is undefined, so a - /// `mutate` racing `finish` may still report `Ok` and then be dropped. The - /// guarantee is for the sequential caller — once `finish` *returns*, a later - /// `mutate`/`probe` from that caller sees `Closed`. - pub async fn finish(&self) -> Result<(), BidiError> { - // Latch closed only *after* the send is delivered. A cancelled `finish` - // (future dropped mid-send) never delivered `Finish`, so it must not - // wedge the handle — leaving the latch untouched is exactly right. And - // because the latch is monotonic (set on delivery, never cleared), two - // concurrent `finish` calls can't race a reset. The same-caller contract - // still holds: a FIFO-following `mutate`/`probe` runs after this store. - self.commands - .send(Command::Finish) - .await - .map_err(|_| BidiError::Closed)?; - self.finished.store(true, Ordering::Release); - Ok(()) - } - - /// Probe the link (e.g. right after the process resumes) with the default - /// deadline: `N ×` the server's advertised keepalive interval (a 30s-derived - /// fallback until `Started` arrives). Resolves `Ok` on the matching `Pong`, - /// `Closed` if the link is already torn down, or `ProbeTimedOut` if no pong - /// arrives in time — the half-open case this whole mechanism exists to catch. - pub async fn probe(&self) -> Result<(), BidiError> { - self.probe_within(self.default_probe_timeout()).await - } - - /// Probe with an explicit deadline. A latency-sensitive caller — say a push - /// notification handler that must decide in a couple of seconds whether to - /// reuse the connection or re-open — passes a bound far below the default. - /// A tight bound trades occasional false `ProbeTimedOut`s for speed, which is - /// safe: re-opening replays from durable cursors and discards duplicates. - /// - /// The timeout covers the whole probe — submitting the ping (so a stalled - /// wire backpressuring the command queue can't park us) and awaiting the - /// pong. We deliberately don't fail-fast on a full command queue: a transient - /// burst of `mutate`s is "busy," not "dead." - pub async fn probe_within(&self, timeout: Duration) -> Result<(), BidiError> { - match tokio::time::timeout(timeout, self.probe_inner()).await { - Ok(result) => result, - Err(_elapsed) => Err(BidiError::ProbeTimedOut), - } - } - - async fn probe_inner(&self) -> Result<(), BidiError> { - if self.finished.load(Ordering::Acquire) { - return Err(BidiError::Closed); - } - let nonce = self - .probe_nonce - .fetch_add(1, Ordering::Relaxed) - .wrapping_add(1); - let (ack, ack_rx) = oneshot::channel(); - self.commands - .send(Command::Probe { nonce, ack }) - .await - .map_err(|_| BidiError::Closed)?; - ack_rx.await.map_err(|_| BidiError::Closed) - } - - fn default_probe_timeout(&self) -> Duration { - let keepalive = match self.keepalive_ms.load(Ordering::Relaxed) { - 0 => DEFAULT_KEEPALIVE_MS, - ms => ms, - }; - Duration::from_millis(u64::from(keepalive) * u64::from(PROBE_TIMEOUT_MULTIPLIER)) - } - - /// Next event, in wire order. `None` means the connection ended (server - /// close, network death, or reap) — resume from durable cursors on a fresh - /// connection. - /// - /// Only resolves on an event or end-of-stream: a server that goes quiet - /// without closing (a half-open link) leaves this pending. Detect that gap - /// out of band with [`Self::probe`] plus a timeout, never by waiting here. - pub async fn next(&mut self) -> Option { - self.events.recv().await - } -} - -impl Drop for BidiConnection { - fn drop(&mut self) { - // Abort the actor so it cannot keep auto-ponging — a zombie keepalive - // would hold the server-side subscription open forever. The abort drops - // the actor's wire-outbound and inbound, cancelling the underlying - // request and tearing the stream down in both directions. - self.actor.end(); - } -} +/// A v3 bidirectional subscription connection (XIP-83). See [`Connection`]. +pub type BidiConnection = Connection; +/// Events surfaced by a v3 [`BidiConnection`], in wire order. +pub type BidiEvent = Event; fn request_frame(request: subscribe_request::v1::Request) -> SubscribeRequest { SubscribeRequest { @@ -297,371 +29,84 @@ fn request_frame(request: subscribe_request::v1::Request) -> SubscribeRequest { } } -/// Hand `frame` to the wire if it has room right now, else queue it for the -/// reserve branch to flush. Returns `false` when the backlog has grown past -/// [`MAX_PENDING_FRAMES`] — the wire is wedged and the actor should give up -/// rather than buffer forever. -/// -/// Reaching the queue means the transport isn't draining the request half — -/// which on a healthy link should essentially never happen — so we warn the -/// first time we fall back to it (not on every frame: an existing backlog skips -/// straight to the queue without re-probing the wire). -#[must_use] -fn enqueue( - wire_out: &mpsc::Sender, - pending: &mut VecDeque, - frame: SubscribeRequest, -) -> bool { - if pending.is_empty() { - match wire_out.try_send(frame) { - Ok(()) => return true, - Err(mpsc::error::TrySendError::Full(frame)) => { - tracing::warn!( - "bidi wire is backpressured (transport not draining the request half); \ - queuing outbound frames" - ); - pending.push_back(frame); - } - // Transport gone: keep the frame queued so the reserve branch's `Err` - // arm tears the actor down on the next iteration. - Err(mpsc::error::TrySendError::Closed(frame)) => pending.push_back(frame), - } - } else { - pending.push_back(frame); - } - if pending.len() > MAX_PENDING_FRAMES { - tracing::warn!( - backlog = pending.len(), - "bidi wire wedged past the outbound backlog cap; giving up so the consumer re-opens" - ); - return false; - } - true -} +impl BidiBinding for V3Binding { + type Request = SubscribeRequest; + type Response = SubscribeResponse; + type Mutate = Mutate; + type GroupMessage = GroupMessage; + type WelcomeMessage = WelcomeMessage; -/// The single owner of the wire: sole reader of inbound, sole writer of -/// outbound, sole producer of events. Caller commands and server frames are -/// multiplexed onto one outbound FIFO via an internal `pending` queue, drained -/// to the wire by a `reserve()` branch — so no send ever blocks the loop and a -/// busy wire can't delay an auto-pong. (Event sends to the *consumer* do still -/// await: a hopelessly slow consumer backpressures here, the intended path to -/// reap-then-resume — distinct from outbound/wire pressure, which is what the -/// queue handles.) It ends when the wire ends/errors or the handle goes away; -/// ending drops `wire_out` (the request half closes, tearing the stream down), -/// the `commands` receiver (so `mutate`/`probe` see `Closed`), the `events` -/// sender (so `next` ends), and any outstanding probe acks (so pending `probe`s -/// see `Closed`). One place, every teardown. It also gives up the same way if -/// the outbound backlog blows past [`MAX_PENDING_FRAMES`] — a wedged wire that -/// will never drain, so we tear down and let the consumer re-open. -async fn run_actor( - mut inbound: S, - wire_out: mpsc::Sender, - mut commands: mpsc::Receiver, - events: mpsc::Sender, - keepalive_ms: Arc, -) where - S: futures::Stream> + Unpin, - E: std::fmt::Display, -{ - // Outstanding client probes awaiting their `Pong`, keyed by nonce. - let mut probes: HashMap> = HashMap::new(); - // Outbound frames awaiting wire capacity. We drain these through a `reserve()` - // branch below rather than `send().await` in a branch body — an `.await` in a - // `select!` arm blocks the whole loop, so a backed-up wire would stall inbound - // reads and delay auto-pongs (getting us reaped on an otherwise healthy link). - let mut pending: VecDeque = VecDeque::new(); - // Set by `Command::Finish`: leave the main loop and drain inbound to close, - // rather than tear everything down. Distinguishes a deliberate half-close - // from the wire/handle-gone breaks, which fall straight through to teardown. - let mut finished = false; - - loop { - tokio::select! { - // Hand one queued frame to the wire the moment it has capacity, - // concurrently with everything else. This is what keeps a busy wire - // from blocking the loop. Gated on a non-empty queue so we only - // contend for a permit when there's something to flush. - permit = wire_out.reserve(), if !pending.is_empty() => { - match permit { - Ok(permit) => { - permit.send(pending.pop_front().expect("queue is non-empty")); - // Drain as much of the backlog as the wire will take right - // now, so a cleared stall flushes promptly. - while !pending.is_empty() { - match wire_out.try_reserve() { - Ok(permit) => { - permit.send(pending.pop_front().expect("queue is non-empty")); - } - Err(_) => break, // wire full again, or closed - } - } - } - Err(_) => break, // transport gone - } - } - // A command from the handle: mutate, or a liveness-probe ping. Accept - // new commands only while the queue has room — that is the backpressure - // to `mutate`, mirroring the old bounded wire. Inbound is never gated - // this way, so auto-pong stays responsive under outbound pressure. - cmd = commands.recv(), if pending.len() < WIRE_BUFFER => { - let Some(cmd) = cmd else { - // Every handle dropped — nothing more will be sent. - break; - }; - let queued = match cmd { - Command::Mutate(mutate) => enqueue( - &wire_out, - &mut pending, - request_frame(subscribe_request::v1::Request::Mutate(mutate)), - ), - Command::Probe { nonce, ack } => { - probes.insert(nonce, ack); - enqueue( - &wire_out, - &mut pending, - request_frame(subscribe_request::v1::Request::Ping(Ping { nonce })), - ) - } - Command::Finish => { - // Half-close: stop accepting new commands, then flush the - // backlog and drain inbound to close (in `drain_after_finish`). - finished = true; - break; - } - }; - if !queued { - break; // wire wedged past the backlog cap — give up - } - } - // A frame from the server. Always read, so liveness never stalls behind - // outbound pressure. - frame = inbound.next() => { - let Some(frame) = frame else { - break; // inbound ended — the stream is closed - }; - let response = match frame { - Ok(r) => r, - Err(e) => { - tracing::warn!("bidi subscription stream errored: {e}"); - break; - } - }; - let Some(subscribe_response::Version::V1(v1)) = response.version else { - // A version we did not speak; XIP-83 pins responses to the - // request version, so this is a server bug — skip, don't die. - tracing::warn!("bidi subscription received unknown response version"); - continue; - }; - use subscribe_response::v1::Response; - match v1.response { - // Liveness is internal: auto-pong / probe-correlate here, never - // surface it to the consumer. - Some(Response::Ping(ping)) => { - // Auto-pong: hand it to the wire (or queue it FIFO behind a - // backlog). A busy wire delays but never drops the pong, and - // never blocks us from reading the next frame — but if the - // backlog has blown past the cap, the wire is wedged and we - // give up rather than keep piling pongs into a dead stream. - if !enqueue( - &wire_out, - &mut pending, - request_frame(subscribe_request::v1::Request::Pong(Pong { - nonce: ping.nonce, - })), - ) { - break; - } - } - Some(Response::Pong(pong)) => { - // Resolve the matching client probe. Unmatched pongs are - // ignored — never surfaced to the consumer. - if let Some(ack) = probes.remove(&pong.nonce) { - let _ = ack.send(()); - } - } - // Everything else is consumer-facing (or a skippable - // future-revision frame). The same emitter feeds the live loop - // and the post-`finish` drain, so their event semantics match. - other => { - if emit_response_events(&events, &keepalive_ms, other).await { - break; - } - } - } - } - } - } - // Resolve any in-flight probes now: dropping the probe-ack senders makes each - // waiting `probe()` see `Closed`. Must happen *before* `drain_after_finish`, - // which can block on `inbound` for a while on a half-open link — a probe must - // never hang on a connection that has already left the live loop. - drop(probes); - // A deliberate half-close drains inbound to close rather than tearing down; - // every other exit reason falls straight through to teardown. - if finished { - drain_after_finish(inbound, wire_out, pending, commands, &events, &keepalive_ms).await; + fn mutate_frame(mutate: Mutate) -> SubscribeRequest { + request_frame(subscribe_request::v1::Request::Mutate(mutate)) } - // One exit log for every reason (wire ended/errored, handle gone, or finished - // draining). The error a caller sees is just `Closed`; the "why" lives here. - tracing::debug!("bidi subscription actor stopped"); - // `wire_out`, `commands`, `events`, and the `pending` queue drop here — that - // *is* the teardown in both directions. -} -/// Translate a non-liveness server response into consumer events, returning true -/// if the consumer is gone (the actor should stop). `Ping`/`Pong` are handled by -/// the caller — they need the wire and the probe map; everything routed here is -/// informational or message data. Shared by the live loop and the post-`finish` -/// drain so both deliver identical events. -async fn emit_response_events( - events: &mpsc::Sender, - keepalive_ms: &Arc, - response: Option, -) -> bool { - use subscribe_response::v1::Response; - match response { - Some(Response::Started(started)) => { - // Record the server's cadence before surfacing `Started`, so a consumer - // that probes in reaction to it already gets the keepalive-derived - // default deadline. - keepalive_ms.store(started.keepalive_interval_ms, Ordering::Relaxed); - emit( - events, - BidiEvent::Started { - keepalive_interval_ms: started.keepalive_interval_ms, - capabilities: started.capabilities, - }, - ) - .await - } - Some(Response::CatchupComplete(complete)) => { - emit( - events, - BidiEvent::CatchUpComplete { - mutate_id: complete.mutate_id, - }, - ) - .await - } - Some(Response::TopicsLive(live)) => { - // Parse kind-prefixed wire bytes into typed `Topic`s, validating each - // kind byte. A malformed topic should never reach us — it means a - // server or wire-format bug — so log the offending bytes (a topic is - // at most 33 bytes: a 1-byte kind + a 32-byte id) and skip it rather - // than kill the connection. - let topics = live - .topics - .into_iter() - .filter_map(|bytes| { - let preview = hex::encode(&bytes[..bytes.len().min(33)]); - match Topic::try_from(bytes) { - Ok(topic) => Some(topic), - Err(e) => { - tracing::warn!( - topic = %preview, - "skipping malformed TopicsLive topic (server/wire-format bug): {e}" - ); - None - } - } - }) - .collect(); - emit(events, BidiEvent::TopicsLive { topics }).await - } - Some(Response::Messages(messages)) => { - if !messages.group_messages.is_empty() - && emit(events, BidiEvent::GroupMessages(messages.group_messages)).await - { - return true; - } - if !messages.welcome_messages.is_empty() - && emit( - events, - BidiEvent::WelcomeMessages(messages.welcome_messages), - ) - .await - { - return true; - } - false - } - // Ping/Pong are handled by the caller (and during the drain we cannot pong - // anyway — the wire is gone); `None` is a future-revision frame. All safe - // to skip: delivery correctness never depends on them. - Some(Response::Ping(_)) | Some(Response::Pong(_)) | None => false, + fn ping_frame(nonce: u64) -> SubscribeRequest { + request_frame(subscribe_request::v1::Request::Ping(Ping { nonce })) } -} -/// Drain inbound to completion after a half-close. First make a best-effort flush -/// of any frames the caller already had accepted (a `mutate` it got `Ok` for, an -/// auto-pong, a probe ping queued under backpressure) so half-close doesn't -/// silently discard them. Then dropping `wire_out` ends the outbound stream (the -/// server sees the half-close, finishes the wave, and closes its side); dropping -/// `commands` makes any late `mutate`/`probe` see `Closed`. We keep surfacing -/// events until the server closes inbound (`next` -> `None`) or the consumer goes -/// away. Outstanding probes are not answered post-finish — they drop with the -/// actor and resolve to `Closed`. -async fn drain_after_finish( - mut inbound: S, - wire_out: mpsc::Sender, - mut pending: VecDeque, - commands: mpsc::Receiver, - events: &mpsc::Sender, - keepalive_ms: &Arc, -) where - S: futures::Stream> + Unpin, - E: std::fmt::Display, -{ - // Best-effort flush of the accepted-but-unsent backlog before closing the - // request half. Non-blocking on purpose: `try_reserve` sends each frame only - // if the wire has room right now and stops otherwise. We must NOT `await` a - // reserve here — if the transport has stalled (wire full, server no longer - // reading the request stream) it would block forever, so `drop(wire_out)` - // would never run and the half-close would never reach the server. Frames we - // can't place are lost, but the consumer re-syncs from durable cursors on - // re-open, so a wedged actor is the strictly worse outcome. - while let Some(frame) = pending.pop_front() { - match wire_out.try_reserve() { - Ok(permit) => permit.send(frame), - Err(_) => break, // wire full or gone — close rather than block - } + fn pong_frame(nonce: u64) -> SubscribeRequest { + request_frame(subscribe_request::v1::Request::Pong(Pong { nonce })) } - drop(wire_out); - drop(commands); - while let Some(frame) = inbound.next().await { - let response = match frame { - Ok(r) => r, - Err(e) => { - tracing::warn!("bidi stream errored during finish drain: {e}"); - break; - } - }; + + fn handle(response: SubscribeResponse) -> Inbound { let Some(subscribe_response::Version::V1(v1)) = response.version else { - tracing::warn!("bidi received unknown response version during finish drain"); - continue; + // A version we did not speak; XIP-83 pins responses to the request + // version, so this is a server bug — skip, don't die. + tracing::warn!("bidi subscription received unknown response version"); + return Inbound::Skip; }; - if emit_response_events(events, keepalive_ms, v1.response).await { - break; // consumer gone + use subscribe_response::v1::Response; + match v1.response { + // Liveness is internal; the core auto-pongs and correlates probes. + Some(Response::Ping(ping)) => Inbound::Ping(ping.nonce), + Some(Response::Pong(pong)) => Inbound::Pong(pong.nonce), + Some(Response::Started(started)) => Inbound::Emit(Event::Started { + keepalive_interval_ms: started.keepalive_interval_ms, + capabilities: started.capabilities, + }), + Some(Response::CatchupComplete(complete)) => Inbound::Emit(Event::CatchUpComplete { + mutate_id: complete.mutate_id, + }), + Some(Response::TopicsLive(live)) => Inbound::Emit(Event::TopicsLive { + topics: parse_topics(live.topics), + }), + Some(Response::Messages(messages)) => Inbound::Messages { + group: messages.group_messages, + welcome: messages.welcome_messages, + }, + // A future-revision arm: informational frames are safe to skip. + None => Inbound::Skip, } } } -/// Send an event to the consumer; awaits a free slot (the backpressure that -/// stalls wire reads once [`EVENT_BUFFER`] fills) and returns true when the -/// consumer is gone and the actor should shut down. -async fn emit(events: &mpsc::Sender, event: BidiEvent) -> bool { - events.send(event).await.is_err() +impl BidiConnection { + /// Open the stream and send `initial` as the first Mutate (it names the + /// initial topic set with per-topic resume cursors; XIP-83 client req 3). + pub async fn open(api: &A, initial: Mutate) -> Result + where + A: XmtpMlsBidiStreams, + A::SubscribeStream: 'static, + { + Self::start(initial, |outbound| api.subscribe_bidi(outbound)).await + } } #[cfg(test)] mod tests { use super::*; + use crate::queries::bidi::{ + BidiError, DEFAULT_KEEPALIVE_MS, MAX_PENDING_FRAMES, PROBE_TIMEOUT_MULTIPLIER, WIRE_BUFFER, + }; + use futures::StreamExt; use futures::stream::BoxStream; use std::sync::Mutex; + use std::time::Duration; + use tokio::sync::mpsc; use xmtp_proto::api::ApiClientError; use xmtp_proto::mls_v1::subscribe_request::v1::mutate::Subscription; use xmtp_proto::mls_v1::{group_message, welcome_message}; - use xmtp_proto::types::TopicKind; + use xmtp_proto::types::{Topic, TopicKind}; /// A scripted peer: captures every frame the client sends and lets the test /// play server frames into the connection. @@ -763,9 +208,9 @@ mod tests { } fn wire_topic(kind: TopicKind, identifier: &[u8]) -> Vec { - let mut topic = vec![kind as u8]; - topic.extend_from_slice(identifier); - topic + // The production constructor, so these tests can't drift from the real + // kind-prefixed wire layout. + kind.create(identifier).into() } fn typed_topic(kind: TopicKind, identifier: &[u8]) -> Topic { @@ -1208,6 +653,50 @@ mod tests { )); } + /// A `Finish` queued behind a backlogged wire must still be processed — the + /// commands branch is deliberately ungated so a wedged wire can't starve it + /// (with a `pending`-room gate, the half-close would never reach the + /// transport and `next()` would hang forever). Regression test for that gate: + /// after `finish`, the flush budget expires, the un-flushable backlog is + /// dropped, and the request half closes — so draining the held outbound + /// stream yields exactly the frames the wire had already accepted, then ends. + #[xmtp_common::test(unwrap_try = true)] + async fn finish_is_processed_under_wire_backpressure() { + let (_to_client, inbound) = mpsc::unbounded_channel(); + let api = WedgedWireApi { + inbound: Mutex::new(Some(inbound)), + held: Mutex::new(None), + }; + let conn = BidiConnection::open(&api, Mutate::default()).await?; + + // Back the wire up well past its depth: the initial Mutate plus the + // first WIRE_BUFFER - 1 mutates fill the wire; the rest park in the + // actor's pending queue. + for _ in 0..(WIRE_BUFFER * 2) { + conn.mutate(Mutate::default()).await?; + } + + // The half-close must be accepted AND processed despite the backlog. + conn.finish().await?; + + // Give the drain's flush budget time to expire (the wire never drains, + // so only the budget can end the flush), then take the held outbound + // stream: it must yield the WIRE_BUFFER frames the wire had accepted and + // then END — proof the actor dropped the request half rather than + // parking forever with Finish stuck behind the backlog. + tokio::time::sleep(Duration::from_secs(2)).await; + let mut outbound = api.held.lock().unwrap().take().expect("wire was opened"); + let mut flushed = 0usize; + while let Some(_frame) = outbound.next().await { + flushed += 1; + } + assert_eq!( + flushed, WIRE_BUFFER, + "only the already-accepted wire frames flush; the backlog is dropped \ + and the request half closes" + ); + } + /// After `finish`, the request half is half-closed: `mutate`/`probe` must /// report `Closed` synchronously. Without the handle-side flag they would /// race the actor — `Command::Finish` is still in flight, the command receiver @@ -1260,51 +749,4 @@ mod tests { ); assert!(finish_res.is_ok()); } - - /// `finish` is a half-close, not an abort: frames the caller already had - /// accepted under backpressure (sitting in the actor's `pending` queue, not - /// yet on the wire) must be flushed before the request half closes. Drive - /// `drain_after_finish` directly with a pre-seeded `pending` and an - /// already-closed inbound, so the flush is the only behaviour under test. - #[xmtp_common::test(unwrap_try = true)] - async fn finish_flushes_pending_frames_before_closing() { - let (wire_out, mut wire_rx) = mpsc::channel::(WIRE_BUFFER); - let (_commands_tx, commands_rx) = mpsc::channel::(COMMAND_BUFFER); - let (events, _events_rx) = mpsc::channel::(EVENT_BUFFER); - let keepalive = Arc::new(AtomicU32::new(0)); - - // A distinctively-tagged frame queued as if wire backpressure had parked - // it before the caller half-closed. - let marker = wire_topic(TopicKind::GroupMessagesV1, b"flushed"); - let mut pending = VecDeque::new(); - pending.push_back(request_frame(subscribe_request::v1::Request::Mutate( - Mutate { - removes: vec![marker.clone()], - ..Default::default() - }, - ))); - - // Inbound is already closed, so the drain has nothing to read and returns - // as soon as the pending flush completes. - let inbound = futures::stream::empty::>(); - drain_after_finish(inbound, wire_out, pending, commands_rx, &events, &keepalive).await; - - // The queued frame reached the wire instead of being dropped on close. - let sent = wire_rx - .recv() - .await - .expect("the pending frame must be flushed to the wire"); - let Some(subscribe_request::Version::V1(v1)) = sent.version else { - panic!("flushed frame must be a v1 request"); - }; - let Some(subscribe_request::v1::Request::Mutate(m)) = v1.request else { - panic!("flushed frame must be the queued mutate"); - }; - assert_eq!(m.removes, vec![marker]); - // Nothing else, and the wire is closed (the actor dropped its sender). - assert!( - wire_rx.recv().await.is_none(), - "the wire closes once the flush is done" - ); - } } diff --git a/crates/xmtp_mls/src/subscriptions/d14n_bidi_tests.rs b/crates/xmtp_mls/src/subscriptions/d14n_bidi_tests.rs new file mode 100644 index 0000000000..4c476ffdb1 --- /dev/null +++ b/crates/xmtp_mls/src/subscriptions/d14n_bidi_tests.rs @@ -0,0 +1,402 @@ +//! Live integration tests for the XIP-83 bidirectional `Subscribe` connection on +//! the d14n (xmtpv4 `QueryApi`) backend, round-tripping to the local xmtpd. +//! +//! Counterpart to `bidi_tests.rs` (v3). These open a real [`D14nBidiConnection`] +//! against the feature-switched d14n test client and assert the same wire +//! contract — but exercise the d14n delivery path end to end: `OriginatorEnvelope` +//! batches decoded into unified `xmtp_proto::types` messages through the +//! per-envelope extractors (no transport-level reordering — see the binding's +//! module docs for why ordering belongs to the consumer). +//! +//! Assertions are derived from the stream: the unified `GroupMessage` carries an +//! openmls `ProtocolMessage`, so application messages are counted independently +//! of the MLS commits a membership change produces. d14n cursors are per-originator +//! vectors, so uniqueness is keyed on the full `(sequence_id, originator_id)` and +//! we assert phase + counts rather than a single global order. +//! +//! d14n-only and native-only. + +use crate::builder::ClientBuilder; +use crate::context::XmtpSharedContext; +use crate::groups::MlsGroup; +use crate::groups::send_message_opts::SendMessageOpts; +use crate::utils::test::{FullXmtpClient, TestXmtpMlsContext}; +use openmls::framing::ContentType; +use std::collections::BTreeSet; +use std::time::Duration; +use xmtp_api_d14n::{D14nBidiConnection, D14nBidiEvent}; +use xmtp_cryptography::utils::generate_local_wallet; +use xmtp_proto::types::{GroupMessage, Topic}; +use xmtp_proto::xmtp::xmtpv4::message_api::subscribe_request::v1::{Mutate, mutate::Subscription}; + +/// Stable key for a delivered group message: its full vector-cursor position. +fn cursor_key(m: &GroupMessage) -> (u64, u32) { + (m.cursor.sequence_id, m.cursor.originator_id) +} + +/// Whether a delivered group message is an application message (vs an MLS commit +/// from a membership change). +fn is_app(m: &GroupMessage) -> bool { + matches!(m.message.content_type(), ContentType::Application) +} + +/// Concise one-line summary of a frame, for clear panic messages. +fn summarize(ev: &D14nBidiEvent) -> String { + match ev { + D14nBidiEvent::Started { + keepalive_interval_ms, + capabilities, + } => format!("Started(keepalive={keepalive_interval_ms}ms, caps={capabilities:?})"), + D14nBidiEvent::CatchUpComplete { mutate_id } => { + format!("CatchUpComplete(mutate_id={mutate_id})") + } + D14nBidiEvent::TopicsLive { topics } => format!("TopicsLive(n={})", topics.len()), + D14nBidiEvent::GroupMessages(m) => format!( + "GroupMessages(cursors={:?})", + m.iter().map(cursor_key).collect::>() + ), + D14nBidiEvent::WelcomeMessages(w) => format!("WelcomeMessages(n={})", w.len()), + } +} + +/// A d14n group-subscription Mutate from cursor 0 (the beginning of the topic). +fn subscribe_group(topic: &Topic, history_only: bool, mutate_id: u64) -> Mutate { + Mutate { + adds: vec![Subscription { + topic: topic.cloned_vec(), + last_seen: None, + }], + removes: vec![], + history_only, + mutate_id, + } +} + +/// Next frame, failing fast (rather than hanging to the test timeout) if the +/// server goes quiet or the connection ends when a frame was expected. +async fn next_within(conn: &mut D14nBidiConnection, secs: u64) -> D14nBidiEvent { + tokio::time::timeout(Duration::from_secs(secs), conn.next()) + .await + .expect("timed out waiting for a d14n bidi frame") + .expect("d14n bidi connection closed unexpectedly") +} + +/// Prelude shared by the group-subscription tests: two clients, a group with +/// both members, `history` pre-subscription app messages, then a subscription +/// opened from cursor 0 with `Started` already consumed. The clients are +/// returned so the test keeps them (and their backing streams) alive. +async fn open_group_history_sub( + history: usize, + history_only: bool, + mutate_id: u64, +) -> ( + FullXmtpClient, + FullXmtpClient, + MlsGroup, + Topic, + D14nBidiConnection, +) { + let alix = ClientBuilder::new_test_client_vanilla(&generate_local_wallet()).await; + let bo = ClientBuilder::new_test_client_vanilla(&generate_local_wallet()).await; + + let group = alix.create_group(None, None).expect("create group"); + group + .add_members(&[bo.inbox_id()]) + .await + .expect("add member"); + + for i in 0..history { + group + .send_message( + format!("history-{i}").as_bytes(), + SendMessageOpts::default(), + ) + .await + .expect("send history message"); + } + + let topic = Topic::new_group_message(group.group_id); + let mut conn = D14nBidiConnection::open( + bo.context.api().api_client.inner(), + subscribe_group(&topic, history_only, mutate_id), + ) + .await + .expect("open d14n bidi connection"); + assert!( + matches!( + next_within(&mut conn, 10).await, + D14nBidiEvent::Started { .. } + ), + "first frame must be Started" + ); + (alix, bo, group, topic, conn) +} + +/// Handshake + live welcome delivery + probe — the minimal proof xmtpd speaks the +/// d14n dialect and the welcome extraction works. +#[xmtp_common::timeout(Duration::from_secs(20))] +#[xmtp_common::test(unwrap_try = true)] +async fn d14n_bidi_delivers_live_welcome_over_the_wire() { + let alix = ClientBuilder::new_test_client_vanilla(&generate_local_wallet()).await; + let caro = ClientBuilder::new_test_client_vanilla(&generate_local_wallet()).await; + + let welcome_topic = Topic::new_welcome_message(caro.installation_public_key()); + let initial = Mutate { + adds: vec![Subscription { + topic: welcome_topic.cloned_vec(), + last_seen: None, + }], + removes: vec![], + history_only: false, + mutate_id: 1, + }; + let mut conn = D14nBidiConnection::open(caro.context.api().api_client.inner(), initial).await?; + + let D14nBidiEvent::Started { + keepalive_interval_ms, + .. + } = conn.next().await.expect("connection closed before Started") + else { + panic!("first frame must be Started"); + }; + tracing::info!("d14n bidi started; server keepalive = {keepalive_interval_ms}ms"); + + let group = alix.create_group(None, None)?; + group.add_members(&[caro.inbox_id()]).await?; + + // The loop's break condition is the delivery proof: a non-empty + // WelcomeMessages frame arrived. + loop { + match conn.next().await { + Some(D14nBidiEvent::WelcomeMessages(w)) if !w.is_empty() => break, + Some(other) => tracing::info!("pre-welcome d14n event: {}", summarize(&other)), + None => panic!("connection closed before the welcome arrived"), + } + } + + conn.probe().await?; +} + +/// The happy path, in one test: everything published before the subscription is +/// caught up before the live marker; messages published while the stream starts +/// arrive exactly once; messages published after the marker stream live. Counts +/// application messages (excluding commits) and keys uniqueness on the full +/// vector cursor. +#[xmtp_common::timeout(Duration::from_secs(40))] +#[xmtp_common::test(unwrap_try = true)] +async fn d14n_bidi_catch_up_precedes_live_marker_then_streams_live() { + const HISTORY: usize = 5; + const CONCURRENT: usize = 3; + const LIVE: usize = 4; + const TOTAL_APP: usize = HISTORY + CONCURRENT + LIVE; + const MUTATE_ID: u64 = 77; + + let (_alix, _bo, group, topic, mut conn) = + open_group_history_sub(HISTORY, false, MUTATE_ID).await; + + // --- while the stream is starting, publish more (race the catch-up edge) --- + for i in 0..CONCURRENT { + group + .send_message( + format!("concurrent-{i}").as_bytes(), + SendMessageOpts::default(), + ) + .await?; + } + + let mut seen: BTreeSet<(u64, u32)> = BTreeSet::new(); + let mut app_count = 0usize; + let mut catchup_app = 0usize; + let mut catchup_complete: Option = None; + + // Phase 1: drain catch-up until the topic crosses to live. + loop { + match next_within(&mut conn, 10).await { + D14nBidiEvent::GroupMessages(m) => { + for g in &m { + assert!(seen.insert(cursor_key(g)), "duplicate cursor in catch-up"); + if is_app(g) { + app_count += 1; + catchup_app += 1; + } + } + } + D14nBidiEvent::CatchUpComplete { mutate_id } => catchup_complete = Some(mutate_id), + D14nBidiEvent::TopicsLive { topics } => { + assert!(topics.contains(&topic), "our topic must be in TopicsLive"); + break; + } + other => panic!("unexpected frame during catch-up: {}", summarize(&other)), + } + } + assert!( + catchup_app >= HISTORY, + "catch-up must contain at least the {HISTORY} pre-subscription messages, got {catchup_app}" + ); + + // --- live: published strictly after the marker; must stream live --- + for i in 0..LIVE { + group + .send_message(format!("live-{i}").as_bytes(), SendMessageOpts::default()) + .await?; + } + + // Phase 2: drain until every application message we sent has been delivered + // exactly once. + while app_count < TOTAL_APP { + match next_within(&mut conn, 10).await { + D14nBidiEvent::GroupMessages(m) => { + for g in &m { + assert!(seen.insert(cursor_key(g)), "cursor delivered twice"); + if is_app(g) { + app_count += 1; + } + } + } + D14nBidiEvent::CatchUpComplete { mutate_id } => catchup_complete = Some(mutate_id), + other => panic!("unexpected frame on live stream: {}", summarize(&other)), + } + } + + assert_eq!( + app_count, TOTAL_APP, + "exactly the application messages we sent" + ); + assert_eq!( + catchup_complete, + Some(MUTATE_ID), + "CatchUpComplete must echo our mutate_id" + ); + conn.probe().await?; +} + +/// `history_only` catches the subscription up to the live edge — history plus the +/// markers — but does NOT register the topic for live delivery: a later publish +/// must not stream. +#[xmtp_common::timeout(Duration::from_secs(40))] +#[xmtp_common::test(unwrap_try = true)] +async fn d14n_bidi_history_only_catches_up_then_delivers_nothing_live() { + const HISTORY: usize = 4; + const MUTATE_ID: u64 = 99; + + let (_alix, _bo, group, topic, mut conn) = + open_group_history_sub(HISTORY, true, MUTATE_ID).await; + + let mut catchup_app = 0usize; + let mut live_marker = false; + let mut catchup_complete: Option = None; + while !(live_marker && catchup_complete.is_some()) { + match next_within(&mut conn, 10).await { + D14nBidiEvent::GroupMessages(m) => { + for g in &m { + if is_app(g) { + catchup_app += 1; + } + } + } + D14nBidiEvent::TopicsLive { topics } => { + assert!(topics.contains(&topic), "our topic must be in TopicsLive"); + live_marker = true; + } + D14nBidiEvent::CatchUpComplete { mutate_id } => catchup_complete = Some(mutate_id), + other => panic!( + "unexpected frame during history-only catch-up: {}", + summarize(&other) + ), + } + } + assert!( + catchup_app >= HISTORY, + "history-only catch-up must contain the {HISTORY} pre-subscription messages, got {catchup_app}" + ); + assert_eq!( + catchup_complete, + Some(MUTATE_ID), + "CatchUpComplete must echo our mutate_id" + ); + + // history_only does not register for live delivery: a new publish must not + // arrive over this connection. + group + .send_message(b"should-not-stream", SendMessageOpts::default()) + .await?; + match tokio::time::timeout(Duration::from_secs(5), conn.next()).await { + Err(_) => {} // idle: correct — history only, nothing streams live + // Without a half-close the node MUST keep the stream open (XIP-83 server + // req 9 closes only after the *client* half-closes), so an ended stream + // is a teardown bug — and tolerating it would let a dead connection pass + // this negative check vacuously. + Ok(None) => panic!("stream ended without a half-close: history_only must stay open"), + Ok(Some(D14nBidiEvent::GroupMessages(m))) => panic!( + "history_only must not deliver live messages, got {:?}", + m.iter().map(cursor_key).collect::>() + ), + Ok(Some(other)) => { + panic!( + "history_only delivered an unexpected live frame: {}", + summarize(&other) + ) + } + } +} + +/// Bounded sync: `history_only` + half-close. The client catches up, then calls +/// `finish()` to half-close the request half; xmtpd finishes the wave and closes +/// the stream, so the consumer drains `next()` to `None`. (That `mutate` then +/// reports `Closed` is the handle-side latch, pinned by the unit test +/// `mutate_and_probe_report_closed_after_finish` — asserting it here would be +/// vacuously true once `finish()` succeeded.) +#[xmtp_common::timeout(Duration::from_secs(40))] +#[xmtp_common::test(unwrap_try = true)] +async fn d14n_bidi_history_only_half_close_drains_then_server_closes() { + const HISTORY: usize = 4; + const MUTATE_ID: u64 = 123; + + let (_alix, _bo, _group, topic, mut conn) = + open_group_history_sub(HISTORY, true, MUTATE_ID).await; + + conn.finish().await?; + + let mut catchup_app = 0usize; + let mut live_marker = false; + let mut catchup_complete: Option = None; + loop { + match tokio::time::timeout(Duration::from_secs(10), conn.next()).await { + Err(_) => { + panic!("bounded sync never closed: xmtpd kept the stream open after finish()") + } + Ok(None) => break, // server closed the bounded stream — the point of the test + Ok(Some(D14nBidiEvent::GroupMessages(m))) => { + for g in &m { + if is_app(g) { + catchup_app += 1; + } + } + } + Ok(Some(D14nBidiEvent::TopicsLive { topics })) => { + assert!(topics.contains(&topic), "our topic must be in TopicsLive"); + live_marker = true; + } + Ok(Some(D14nBidiEvent::CatchUpComplete { mutate_id })) => { + catchup_complete = Some(mutate_id) + } + Ok(Some(other)) => panic!( + "unexpected frame during bounded sync: {}", + summarize(&other) + ), + } + } + assert!( + catchup_app >= HISTORY, + "bounded sync must deliver the {HISTORY} pre-subscription messages, got {catchup_app}" + ); + assert!( + live_marker, + "bounded sync must emit the live marker before closing" + ); + assert_eq!( + catchup_complete, + Some(MUTATE_ID), + "CatchUpComplete must echo our mutate_id" + ); +} diff --git a/crates/xmtp_mls/src/subscriptions/mod.rs b/crates/xmtp_mls/src/subscriptions/mod.rs index c521984384..6435ae6d7f 100644 --- a/crates/xmtp_mls/src/subscriptions/mod.rs +++ b/crates/xmtp_mls/src/subscriptions/mod.rs @@ -15,11 +15,14 @@ use process_welcome::ProcessWelcomeResult; use stream_all::StreamAllMessages; use stream_conversations::{StreamConversations, WelcomeOrGroup}; -// Live integration test for the XIP-83 bidi connection. v3-only (the bidi -// transport is implemented for the v3 client) and native-only (full-duplex -// HTTP/2 is unavailable on the wasm gRPC-Web transport). +// Live integration tests for the XIP-83 bidi connection (native-only — +// full-duplex HTTP/2 is unavailable on the wasm gRPC-Web transport). The v3 and +// d14n backends have distinct wire types, so each has its own module gated on +// the backend feature switch. #[cfg(all(test, not(target_arch = "wasm32"), not(feature = "d14n")))] mod bidi_tests; +#[cfg(all(test, not(target_arch = "wasm32"), feature = "d14n"))] +mod d14n_bidi_tests; pub(crate) mod d14n_compat; pub mod process_message; pub mod process_welcome;