From 5d2c9c45d49a7c6c2c18b5edcab53ab61d578fa1 Mon Sep 17 00:00:00 2001 From: Tyler Hawkes Date: Fri, 26 Jun 2026 16:19:20 -0600 Subject: [PATCH] feat(xmtp_api_d14n): XIP-83 bidi half-close + live integration tests Builds on the merged BidiConnection actor (#3772): - TrackedStatsClient forwards XmtpMlsBidiStreams so the feature-switched test client can open a BidiConnection. - BidiConnection::finish() half-closes the request half for bounded-sync. - Live v3 integration tests: handshake+welcome, catch-up/marker/live ordering, history_only, and bounded-sync half-close. --- Cargo.lock | 4 +- crates/xmtp_api_d14n/src/queries/api_stats.rs | 25 + .../src/queries/v3/connection.rs | 386 ++++++++++++--- .../xmtp_mls/src/subscriptions/bidi_tests.rs | 447 ++++++++++++++++++ crates/xmtp_mls/src/subscriptions/mod.rs | 5 + deny.toml | 1 + 6 files changed, 806 insertions(+), 62 deletions(-) create mode 100644 crates/xmtp_mls/src/subscriptions/bidi_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 03b9794a11..173c9bac45 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1020,9 +1020,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.102" +version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "arbitrary" diff --git a/crates/xmtp_api_d14n/src/queries/api_stats.rs b/crates/xmtp_api_d14n/src/queries/api_stats.rs index b756fa6ecc..d4d2455402 100644 --- a/crates/xmtp_api_d14n/src/queries/api_stats.rs +++ b/crates/xmtp_api_d14n/src/queries/api_stats.rs @@ -200,6 +200,31 @@ where } } +xmtp_common::if_native! { + use xmtp_proto::api_client::XmtpMlsBidiStreams; + + // `XmtpMlsBidiStreams` is native-only, so this forward is gated like the + // trait. It carries no per-call stat (the bidi stream is opened once and + // mutated in place, not counted per RPC like the unary/stream calls above); + // it exists so the stats wrapper is bidi-capable, letting the standard + // feature-switched test client open a `BidiConnection`. + #[xmtp_common::async_trait] + impl XmtpMlsBidiStreams for TrackedStatsClient + where + C: XmtpMlsBidiStreams, + { + type SubscribeStream = ::SubscribeStream; + type Error = ::Error; + + async fn subscribe_bidi( + &self, + requests: futures::stream::BoxStream<'static, xmtp_proto::mls_v1::SubscribeRequest>, + ) -> Result { + self.inner.subscribe_bidi(requests).await + } + } +} + impl HasStats for TrackedStatsClient { fn aggregate_stats(&self) -> AggregateStats { AggregateStats { diff --git a/crates/xmtp_api_d14n/src/queries/v3/connection.rs b/crates/xmtp_api_d14n/src/queries/v3/connection.rs index 1c5e005422..387b2f781a 100644 --- a/crates/xmtp_api_d14n/src/queries/v3/connection.rs +++ b/crates/xmtp_api_d14n/src/queries/v3/connection.rs @@ -12,7 +12,7 @@ use std::collections::{HashMap, VecDeque}; use std::sync::Arc; -use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use std::time::Duration; use futures::StreamExt; @@ -90,6 +90,10 @@ enum Command { 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 @@ -98,6 +102,16 @@ 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. @@ -149,6 +163,7 @@ impl BidiConnection { commands: commands_tx, events, probe_nonce: AtomicU64::new(0), + finished: AtomicBool::new(false), keepalive_ms, actor: actor.abort_handle(), }) @@ -158,12 +173,50 @@ impl BidiConnection { /// (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`, @@ -191,6 +244,9 @@ impl BidiConnection { } 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) @@ -314,6 +370,10 @@ async fn run_actor( // `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! { @@ -362,6 +422,12 @@ async fn run_actor( 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 @@ -388,6 +454,8 @@ async fn run_actor( }; 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 @@ -411,73 +479,171 @@ async fn run_actor( let _ = ack.send(()); } } - 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); - if emit(&events, BidiEvent::Started { - keepalive_interval_ms: started.keepalive_interval_ms, - capabilities: started.capabilities, - }).await { - break; - } - } - Some(Response::CatchupComplete(complete)) => { - if emit(&events, BidiEvent::CatchUpComplete { - mutate_id: complete.mutate_id, - }).await { - break; - } - } - Some(Response::TopicsLive(live)) => { - // Parse kind-prefixed wire bytes into typed `Topic`s, - // validating each kind byte. A malformed topic is a - // server bug on an informational frame, so skip it (with - // a warn) rather than kill the connection. - let topics = live - .topics - .into_iter() - .filter_map(|bytes| match Topic::try_from(bytes) { - Ok(topic) => Some(topic), - Err(e) => { - tracing::warn!("skipping malformed TopicsLive topic: {e}"); - None - } - }) - .collect(); - if emit(&events, BidiEvent::TopicsLive { topics }).await { - break; - } - } - Some(Response::Messages(messages)) => { - if !messages.group_messages.is_empty() - && emit(&events, BidiEvent::GroupMessages(messages.group_messages)).await - { - break; - } - if !messages.welcome_messages.is_empty() - && emit( - &events, - BidiEvent::WelcomeMessages(messages.welcome_messages), - ) - .await - { + // 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; } } - // An arm a future server revision added: informational frames - // are safe to skip (delivery correctness never depends on them). - None => continue, } } } } - // One exit log for every reason (wire ended/errored, or handle gone). The - // error a caller sees is just `Closed`; the "why" lives here in the logs. + // 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; + } + // 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`, `probes`, and the `pending` queue drop - // here — that *is* the teardown in both directions. + // `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, + } +} + +/// 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 + } + } + 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; + } + }; + let Some(subscribe_response::Version::V1(v1)) = response.version else { + tracing::warn!("bidi received unknown response version during finish drain"); + continue; + }; + if emit_response_events(events, keepalive_ms, v1.response).await { + break; // consumer gone + } + } } /// Send an event to the consumer; awaits a free slot (the backpressure that @@ -1041,4 +1207,104 @@ mod tests { Err(BidiError::Closed) )); } + + /// 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 + /// is briefly alive, and a FIFO-following `mutate` would be accepted into the + /// buffer (returning `Ok`) only to be dropped unread when the actor drains. + #[xmtp_common::test(unwrap_try = true)] + async fn mutate_and_probe_report_closed_after_finish() { + let (api, mut server) = mock_pair(); + let conn = BidiConnection::open(&api, Mutate::default()).await?; + server.next_request().await; // initial mutate + + conn.finish().await?; + + assert!( + matches!(conn.mutate(Mutate::default()).await, Err(BidiError::Closed)), + "mutate after finish must report Closed, not land in the buffer" + ); + assert!( + matches!(conn.probe().await, Err(BidiError::Closed)), + "probe after finish must report Closed" + ); + } + + /// A `probe` in flight when `finish` half-closes must resolve to `Closed`, not + /// hang: the actor drops the probe-ack senders before the (possibly long) + /// inbound drain. Put the probe's ping on the wire, then half-close, and assert + /// the probe resolves `Closed` rather than waiting out its timeout. + #[xmtp_common::test(unwrap_try = true)] + async fn finish_resolves_in_flight_probe_to_closed() { + let (api, mut server) = mock_pair(); + let conn = BidiConnection::open(&api, Mutate::default()).await?; + server.next_request().await; // initial mutate + + // Once the probe's ping reaches the actor (registered, awaiting a pong that + // never comes), half-close. A generous bound keeps a regression — a probe + // that hangs through the drain — from waiting out the keepalive default. + let driver = async { + let req = server.next_request().await; + assert!( + matches!(req, subscribe_request::v1::Request::Ping(_)), + "probe must put a ping on the wire" + ); + conn.finish().await + }; + let (probe_res, finish_res) = + futures::join!(conn.probe_within(Duration::from_secs(5)), driver); + assert!( + matches!(probe_res, Err(BidiError::Closed)), + "an in-flight probe must resolve Closed when finish half-closes, got {probe_res:?}" + ); + 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/bidi_tests.rs b/crates/xmtp_mls/src/subscriptions/bidi_tests.rs new file mode 100644 index 0000000000..c43eb5db25 --- /dev/null +++ b/crates/xmtp_mls/src/subscriptions/bidi_tests.rs @@ -0,0 +1,447 @@ +//! Live integration tests for the XIP-83 bidirectional `Subscribe` connection. +//! +//! The actor's own unit tests (in `xmtp_api_d14n`) drive a mock wire and prove +//! its logic — auto-pong, probe correlation, teardown, backpressure. They prove +//! nothing about whether the node actually speaks the dialect. These open a real +//! [`BidiConnection`] against whichever backend the test feature switch selects +//! (local docker node by default; dev with `--features dev`) and assert the wire +//! contract end to end: the `Started` handshake, catch-up of pre-subscription +//! history strictly before the `TopicsLive` marker, live streaming after it, +//! `history_only` bounded catch-up with no live delivery, bounded-sync half-close +//! (the server closes the stream after the wave), and ping/pong probes. +//! +//! Assertions are derived entirely from the stream (not a side query): each +//! group-message frame carries an `is_commit` flag, so we count application +//! messages independently of the MLS commits the membership ops produce. +//! +//! v3-only and native-only: the bidi transport is implemented for the v3 client +//! and needs full-duplex HTTP/2 (the wasm gRPC-Web transport cannot carry it), +//! and `TestClient` is only bidi-capable under the v3 config. + +use crate::builder::ClientBuilder; +use crate::context::XmtpSharedContext; +use crate::groups::send_message_opts::SendMessageOpts; +use std::collections::BTreeSet; +use std::time::Duration; +use xmtp_api_d14n::{BidiConnection, BidiError, BidiEvent}; +use xmtp_cryptography::utils::generate_local_wallet; +use xmtp_proto::mls_v1::subscribe_request::v1::{Mutate, mutate::Subscription}; +use xmtp_proto::types::Topic; + +/// `(cursor, is_commit)` of a bidi group-message frame. The bidi event carries +/// the raw proto `GroupMessage`, whose `V1.id` is the topic cursor and whose +/// `is_commit` flag separates MLS commits from application messages. +fn gm(m: &xmtp_proto::mls_v1::GroupMessage) -> (u64, bool) { + use xmtp_proto::mls_v1::group_message::Version; + match &m.version { + Some(Version::V1(v1)) => (v1.id, v1.is_commit), + None => panic!("group message frame without a version"), + } +} + +/// Concise one-line summary of a frame, for clear panic messages. +fn summarize(ev: &BidiEvent) -> String { + match ev { + BidiEvent::Started { + keepalive_interval_ms, + capabilities, + } => format!("Started(keepalive={keepalive_interval_ms}ms, caps={capabilities:?})"), + BidiEvent::CatchUpComplete { mutate_id } => { + format!("CatchUpComplete(mutate_id={mutate_id})") + } + BidiEvent::TopicsLive { topics } => format!("TopicsLive(n={})", topics.len()), + BidiEvent::GroupMessages(m) => { + format!( + "GroupMessages(ids={:?})", + m.iter().map(|g| gm(g).0).collect::>() + ) + } + BidiEvent::WelcomeMessages(w) => format!("WelcomeMessages(n={})", w.len()), + } +} + +/// 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 BidiConnection, secs: u64) -> BidiEvent { + tokio::time::timeout(Duration::from_secs(secs), conn.next()) + .await + .expect("timed out waiting for a bidi frame") + .expect("bidi connection closed unexpectedly") +} + +/// Handshake + live welcome delivery + probe — the minimal proof the node speaks +/// the dialect at all. +#[xmtp_common::timeout(Duration::from_secs(20))] +#[xmtp_common::test(unwrap_try = true)] +async fn bidi_connection_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; + + // caro subscribes to its own welcome topic from the beginning of time. + let welcome_topic = Topic::new_welcome_message(caro.installation_public_key()); + let initial = Mutate { + adds: vec![Subscription { + topic: welcome_topic.cloned_vec(), + id_cursor: 0, + }], + removes: vec![], + history_only: false, + mutate_id: 1, + }; + + let mut conn = BidiConnection::open(&caro.context.api().api_client, initial).await?; + + let BidiEvent::Started { + keepalive_interval_ms, + .. + } = conn.next().await.expect("connection closed before Started") + else { + panic!("first frame must be Started"); + }; + tracing::info!("bidi started; server keepalive = {keepalive_interval_ms}ms"); + + let group = alix.create_group(None, None)?; + group.add_members(&[caro.inbox_id()]).await?; + + let welcomes = loop { + match conn.next().await { + Some(BidiEvent::WelcomeMessages(w)) if !w.is_empty() => break w, + Some(other) => tracing::info!("pre-welcome bidi event: {}", summarize(&other)), + None => panic!("connection closed before the welcome arrived"), + } + }; + assert!( + !welcomes.is_empty(), + "expected at least one welcome message" + ); + + conn.probe().await?; +} + +/// The happy path, in one test: everything published before the subscription is +/// caught up strictly before the live marker; messages published while the +/// stream starts arrive exactly once (either side of the marker); messages +/// published after the marker stream live, newer than every catch-up cursor. +#[xmtp_common::timeout(Duration::from_secs(40))] +#[xmtp_common::test(unwrap_try = true)] +async fn bidi_catch_up_precedes_live_marker_then_streams_live() { + 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)?; + group.add_members(&[bo.inbox_id()]).await?; + + const HISTORY: usize = 5; + const CONCURRENT: usize = 3; + const LIVE: usize = 4; + const TOTAL_APP: usize = HISTORY + CONCURRENT + LIVE; + + // --- history: published before the subscription exists --- + for i in 0..HISTORY { + group + .send_message( + format!("history-{i}").as_bytes(), + SendMessageOpts::default(), + ) + .await?; + } + + // --- open the subscription from the beginning of the topic --- + let topic = Topic::new_group_message(group.group_id); + const MUTATE_ID: u64 = 77; + let initial = Mutate { + adds: vec![Subscription { + topic: topic.cloned_vec(), + id_cursor: 0, + }], + removes: vec![], + history_only: false, + mutate_id: MUTATE_ID, + }; + let mut conn = BidiConnection::open(&bo.context.api().api_client, initial).await?; + assert!( + matches!(next_within(&mut conn, 10).await, BidiEvent::Started { .. }), + "first frame must be Started" + ); + + // --- 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 = BTreeSet::new(); + let mut app_count = 0usize; + let mut catchup_app = 0usize; + let mut catchup_max = 0u64; + let mut catchup_complete: Option = None; + + // Phase 1: drain catch-up until the topic crosses to live. `CatchUpComplete` + // straddles the marker (the node emits it just after `TopicsLive`), so track + // it on both sides rather than assuming an order. + loop { + match next_within(&mut conn, 10).await { + BidiEvent::GroupMessages(m) => { + for g in &m { + let (id, is_commit) = gm(g); + assert!(seen.insert(id), "duplicate cursor {id} in catch-up"); + catchup_max = catchup_max.max(id); + if !is_commit { + app_count += 1; + catchup_app += 1; + } + } + } + BidiEvent::CatchUpComplete { mutate_id } => catchup_complete = Some(mutate_id), + BidiEvent::TopicsLive { topics } => { + assert!(topics.contains(&topic), "our topic must be in TopicsLive"); + break; + } + other => panic!("unexpected frame during catch-up: {}", summarize(&other)), + } + } + // Every message published before the subscription is delivered in catch-up, + // before the live marker. (Late-concurrent sends may add more.) + 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. Everything here is newer than the whole catch-up wave (the + // live edge is monotonic). + while app_count < TOTAL_APP { + match next_within(&mut conn, 10).await { + BidiEvent::GroupMessages(m) => { + for g in &m { + let (id, is_commit) = gm(g); + assert!( + id > catchup_max, + "live cursor {id} must exceed catch-up max {catchup_max}" + ); + assert!( + seen.insert(id), + "cursor {id} delivered twice (catch-up/live overlap or dup)" + ); + if !is_commit { + app_count += 1; + } + } + } + BidiEvent::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, no more, no fewer" + ); + 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 +/// `TopicsLive` / `CatchUpComplete` markers ("you have everything as of now") — +/// 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 bidi_history_only_catches_up_then_delivers_nothing_live() { + 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)?; + group.add_members(&[bo.inbox_id()]).await?; + + const HISTORY: usize = 4; + for i in 0..HISTORY { + group + .send_message( + format!("history-{i}").as_bytes(), + SendMessageOpts::default(), + ) + .await?; + } + + let topic = Topic::new_group_message(group.group_id); + const MUTATE_ID: u64 = 99; + let initial = Mutate { + adds: vec![Subscription { + topic: topic.cloned_vec(), + id_cursor: 0, + }], + removes: vec![], + history_only: true, + mutate_id: MUTATE_ID, + }; + let mut conn = BidiConnection::open(&bo.context.api().api_client, initial).await?; + assert!( + matches!(next_within(&mut conn, 10).await, BidiEvent::Started { .. }), + "first frame must be Started" + ); + + // Catch-up still emits the markers; drain until both have arrived. + 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 { + BidiEvent::GroupMessages(m) => { + for g in &m { + if !gm(g).1 { + catchup_app += 1; + } + } + } + BidiEvent::TopicsLive { topics } => { + assert!(topics.contains(&topic), "our topic must be in TopicsLive"); + live_marker = true; + } + BidiEvent::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" + ); + + // The topic was NOT registered 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 does not stream live + Ok(None) => {} // server closed the bounded stream — also acceptable + Ok(Some(BidiEvent::GroupMessages(m))) => panic!( + "history_only must not deliver live messages, got ids {:?}", + m.iter().map(|g| gm(g).0).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 +/// [`BidiConnection::finish`] to half-close the request half; the server finishes +/// the wave and closes the stream itself, so the consumer drains `next()` to +/// `None`. After the close, `mutate` reports `Closed`. +#[xmtp_common::timeout(Duration::from_secs(40))] +#[xmtp_common::test(unwrap_try = true)] +async fn bidi_history_only_half_close_drains_then_server_closes() { + 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)?; + group.add_members(&[bo.inbox_id()]).await?; + + const HISTORY: usize = 4; + for i in 0..HISTORY { + group + .send_message( + format!("history-{i}").as_bytes(), + SendMessageOpts::default(), + ) + .await?; + } + + let topic = Topic::new_group_message(group.group_id); + const MUTATE_ID: u64 = 123; + let initial = Mutate { + adds: vec![Subscription { + topic: topic.cloned_vec(), + id_cursor: 0, + }], + removes: vec![], + history_only: true, + mutate_id: MUTATE_ID, + }; + let mut conn = BidiConnection::open(&bo.context.api().api_client, initial).await?; + assert!( + matches!(next_within(&mut conn, 10).await, BidiEvent::Started { .. }), + "first frame must be Started" + ); + + // Half-close the request half: we are done sending, so the server should + // finish the bounded wave and close the stream. + conn.finish().await?; + + // Drain the bounded wave, then the server must close (next() -> None). + 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: server kept the stream open after finish()") + } + Ok(None) => break, // server closed the bounded stream — the point of the test + Ok(Some(BidiEvent::GroupMessages(m))) => { + for g in &m { + if !gm(g).1 { + catchup_app += 1; + } + } + } + Ok(Some(BidiEvent::TopicsLive { topics })) => { + assert!(topics.contains(&topic), "our topic must be in TopicsLive"); + live_marker = true; + } + Ok(Some(BidiEvent::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" + ); + + // The connection is gone: further requests report Closed. + assert!( + matches!(conn.mutate(Mutate::default()).await, Err(BidiError::Closed)), + "mutate after the bounded-sync close must report Closed" + ); +} diff --git a/crates/xmtp_mls/src/subscriptions/mod.rs b/crates/xmtp_mls/src/subscriptions/mod.rs index 893f46d890..c521984384 100644 --- a/crates/xmtp_mls/src/subscriptions/mod.rs +++ b/crates/xmtp_mls/src/subscriptions/mod.rs @@ -15,6 +15,11 @@ 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). +#[cfg(all(test, not(target_arch = "wasm32"), not(feature = "d14n")))] +mod bidi_tests; pub(crate) mod d14n_compat; pub mod process_message; pub mod process_welcome; diff --git a/deny.toml b/deny.toml index f7d7c1a029..ef86cf6ddc 100644 --- a/deny.toml +++ b/deny.toml @@ -5,6 +5,7 @@ ignore = [ { id = "RUSTSEC-2021-0139", reason = "ansi_term is only used in tests" }, { id = "RUSTSEC-2025-0141", reason = "migrate from bincode" }, { id = "RUSTSEC-2026-0173", reason = "proc-macro-error2 unmaintained; transitive via alloy-sol-macro, no upstream fix yet" }, + { id = "RUSTSEC-2026-0192", reason = "ttf-parser unmaintained (transitive); no upstream fix, not worth failing builds over" }, ] # This rustsec can be added to ignore list if using mls `test_utils` for tests