diff --git a/crates/xmtp_api_d14n/src/middleware/auth.rs b/crates/xmtp_api_d14n/src/middleware/auth.rs index 671416f20a..9d6cc9bafc 100644 --- a/crates/xmtp_api_d14n/src/middleware/auth.rs +++ b/crates/xmtp_api_d14n/src/middleware/auth.rs @@ -181,6 +181,20 @@ impl Client for AuthMiddleware { let request = self.modify_request(request).await?; self.inner.stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: http::request::Builder, + path: http::uri::PathAndQuery, + body: xmtp_common::BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + // Inject the auth header exactly as `stream`/`request` do — otherwise an + // authenticated client's bidi subscription would either skip auth or, if + // we forgot to override at all, fall through to the trait's default + // "unsupported" error. + let request = self.modify_request(request).await?; + self.inner.bidi_stream(request, path, body).await + } } #[xmtp_common::async_trait] @@ -273,6 +287,29 @@ mod tests { futures::stream::once(Box::pin(async move { Ok(body) })), ))) } + + async fn bidi_stream( + &self, + request: http::request::Builder, + _path: http::uri::PathAndQuery, + _body: xmtp_common::BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + // Same header assertion as `stream`: this only passes if the + // middleware actually forwarded `bidi_stream` (rather than inheriting + // the trait's "unsupported" default) AND injected the auth header. + let headers = request.headers_ref().unwrap(); + if let Some(expected_credential) = &self.expected_credential { + assert_eq!( + headers.get(&expected_credential.name).unwrap(), + &expected_credential.value + ); + } else { + assert!(headers.is_empty()); + } + Ok(http::Response::new(BytesStream::new( + futures::stream::empty(), + ))) + } } impl AuthMiddleware { @@ -297,7 +334,28 @@ mod tests { } let request = http::request::Builder::new(); - let result = self.stream(request, path, body).await; + let result = self.stream(request, path.clone(), body).await; + match (&expected, result) { + (Ok(()), Ok(response)) => { + assert_eq!(response.status(), http::StatusCode::OK); + } + (Err(e), Ok(_)) => { + panic!("Expected error: {e}, got Ok"); + } + (Ok(()), Err(e)) => { + panic!("Expected Ok, got error: {e}"); + } + (Err(e), Err(res)) => { + assert_eq!(e, &res.to_string()); + } + } + + let request = http::request::Builder::new(); + // `Box::pin` (not `.boxed()`) so this coerces to `BoxDynStream` on + // both native (`BoxStream`) and wasm (`LocalBoxStream`) targets. + let bidi_body: xmtp_common::BoxDynStream<'static, Bytes> = + Box::pin(futures::stream::empty()); + let result = self.bidi_stream(request, path, bidi_body).await; match (&expected, result) { (Ok(()), Ok(response)) => { assert_eq!(response.status(), http::StatusCode::OK); diff --git a/crates/xmtp_api_d14n/src/middleware/multi_node_client/client.rs b/crates/xmtp_api_d14n/src/middleware/multi_node_client/client.rs index 9b9a019516..c088395b5a 100644 --- a/crates/xmtp_api_d14n/src/middleware/multi_node_client/client.rs +++ b/crates/xmtp_api_d14n/src/middleware/multi_node_client/client.rs @@ -78,6 +78,17 @@ impl Client for MultiNodeClient { inner.stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: http::request::Builder, + path: http::uri::PathAndQuery, + body: xmtp_common::BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + let inner = self.init_inner().await?; + + inner.bidi_stream(request, path, body).await + } } #[xmtp_common::async_trait] diff --git a/crates/xmtp_api_d14n/src/middleware/read_write_client/client.rs b/crates/xmtp_api_d14n/src/middleware/read_write_client/client.rs index 3248f90fe1..e21002d212 100644 --- a/crates/xmtp_api_d14n/src/middleware/read_write_client/client.rs +++ b/crates/xmtp_api_d14n/src/middleware/read_write_client/client.rs @@ -62,6 +62,19 @@ where self.read.stream(request, path, body).await } } + + async fn bidi_stream( + &self, + request: http::request::Builder, + path: http::uri::PathAndQuery, + body: xmtp_common::BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + if path.path().contains(&self.filter) { + self.write.bidi_stream(request, path, body).await + } else { + self.read.bidi_stream(request, path, body).await + } + } } #[xmtp_common::async_trait] diff --git a/crates/xmtp_api_d14n/src/middleware/readonly_client.rs b/crates/xmtp_api_d14n/src/middleware/readonly_client.rs index 1c454ed868..17649e5a64 100644 --- a/crates/xmtp_api_d14n/src/middleware/readonly_client.rs +++ b/crates/xmtp_api_d14n/src/middleware/readonly_client.rs @@ -66,6 +66,20 @@ where self.inner.stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: http::request::Builder, + path: http::uri::PathAndQuery, + body: xmtp_common::BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + let p = path.path(); + if DENY.iter().any(|d| p.contains(d)) { + return Err(ApiClientError::WritesDisabled); + } + + self.inner.bidi_stream(request, path, body).await + } } #[xmtp_common::async_trait] diff --git a/crates/xmtp_api_d14n/src/queries/v3.rs b/crates/xmtp_api_d14n/src/queries/v3.rs index 6627c4bb18..ad93b520f4 100644 --- a/crates/xmtp_api_d14n/src/queries/v3.rs +++ b/crates/xmtp_api_d14n/src/queries/v3.rs @@ -1,3 +1,5 @@ +#[cfg(not(target_arch = "wasm32"))] +mod bidi; mod client; mod identity; mod mls; diff --git a/crates/xmtp_api_d14n/src/queries/v3/bidi.rs b/crates/xmtp_api_d14n/src/queries/v3/bidi.rs new file mode 100644 index 0000000000..1ebe37abff --- /dev/null +++ b/crates/xmtp_api_d14n/src/queries/v3/bidi.rs @@ -0,0 +1,175 @@ +//! XIP-83 bidirectional Subscribe transport for the v3 backend (native-only). + +use crate::V3Client; +use crate::protocol::CursorStore; +use futures::StreamExt; +use prost::Message; +use prost::bytes::Bytes; +use xmtp_proto::ApiEndpoint; +use xmtp_proto::api::{ApiClientError, Client, XmtpStream}; +use xmtp_proto::api_client::XmtpMlsBidiStreams; +use xmtp_proto::mls_v1::{SubscribeRequest, SubscribeResponse}; + +const SUBSCRIBE_PATH: &str = "/xmtp.mls.api.v1.MlsApi/Subscribe"; + +#[xmtp_common::async_trait] +impl XmtpMlsBidiStreams for V3Client +where + C: Client, + Store: CursorStore, +{ + type SubscribeStream = XmtpStream; + + type Error = ApiClientError; + + // Spans the open handshake (not the stream's lifetime) as `rpc.subscribe_bidi`. + // Bidi is consumed directly by `xmtp_mls` with no `xmtp_api` wrapper in front, + // so this transport impl is the RPC boundary — the same layer the other + // `rpc_span`s live at for unary calls. Move it up if a wrapper is ever added. + #[xmtp_common::rpc_span] + async fn subscribe_bidi( + &self, + requests: futures::stream::BoxStream<'static, SubscribeRequest>, + ) -> Result { + tracing::debug!("opening bidirectional subscription"); + let outbound = requests.map(|frame| Bytes::from(frame.encode_to_vec())); + let response = self + .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()), + )) + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used)] + use super::*; + use crate::protocol::NoCursorStore; + use futures::stream; + use xmtp_common::BoxDynStream; + use xmtp_proto::api::BytesStream; + use xmtp_proto::api::mock::MockNetworkClient; + use xmtp_proto::mls_v1::subscribe_request::v1::Mutate; + use xmtp_proto::mls_v1::{Ping, Pong, subscribe_request, subscribe_response}; + + fn req(request: subscribe_request::v1::Request) -> SubscribeRequest { + SubscribeRequest { + version: Some(subscribe_request::Version::V1(subscribe_request::V1 { + request: Some(request), + })), + } + } + + fn resp(response: subscribe_response::v1::Response) -> SubscribeResponse { + SubscribeResponse { + version: Some(subscribe_response::Version::V1(subscribe_response::V1 { + response: Some(response), + })), + } + } + + fn ping_req(nonce: u64) -> SubscribeRequest { + req(subscribe_request::v1::Request::Ping(Ping { nonce })) + } + + /// `subscribe_bidi` must prost-encode each outbound `SubscribeRequest` in + /// order, dial the `Subscribe` path, and decode the inbound byte frames back + /// into `SubscribeResponse`s through `XmtpStream`. + #[xmtp_common::test(unwrap_try = true)] + async fn encodes_outbound_and_decodes_inbound() { + let captured: std::sync::Arc>>> = + Default::default(); + let sink = captured.clone(); + + let mut mock = MockNetworkClient::new(); + mock.expect_bidi_stream() + .return_once(move |_req, path, body| { + assert_eq!(path.path(), "/xmtp.mls.api.v1.MlsApi/Subscribe"); + *sink.lock().unwrap() = Some(body); + let frames: Vec> = vec![ + Ok(Bytes::from( + resp(subscribe_response::v1::Response::Ping(Ping { nonce: 7 })) + .encode_to_vec(), + )), + Ok(Bytes::from( + resp(subscribe_response::v1::Response::Pong(Pong { nonce: 9 })) + .encode_to_vec(), + )), + ]; + Ok(http::Response::new(BytesStream::new(stream::iter(frames)))) + }); + + let client = V3Client::new(mock, NoCursorStore); + let outbound = stream::iter(vec![ + req(subscribe_request::v1::Request::Mutate(Mutate::default())), + ping_req(3), + ]) + .boxed(); + + let inbound = client.subscribe_bidi(outbound).await?; + let decoded: Vec = inbound.map(|r| r.unwrap()).collect().await; + assert_eq!( + decoded, + vec![ + resp(subscribe_response::v1::Response::Ping(Ping { nonce: 7 })), + resp(subscribe_response::v1::Response::Pong(Pong { nonce: 9 })), + ], + ); + + // The outbound stream handed to the transport carries the same requests, + // prost-encoded, in order. Take it out of the mutex before awaiting so no + // lock is held across the `.await`. + let captured_outbound = captured.lock().unwrap().take().unwrap(); + let sent_bytes: Vec = captured_outbound.collect().await; + let sent: Vec = sent_bytes + .iter() + .map(|b| SubscribeRequest::decode(b.clone()).unwrap()) + .collect(); + assert_eq!( + sent, + vec![ + req(subscribe_request::v1::Request::Mutate(Mutate::default())), + ping_req(3), + ], + ); + } + + /// A transport error opening the stream is surfaced as a `ClientWithEndpoint` + /// tagged with the `Subscribe` path, not a bare client error. + #[xmtp_common::test(unwrap_try = true)] + async fn tags_open_error_with_subscribe_endpoint() { + #[derive(Debug, thiserror::Error)] + #[error("boom")] + struct Boom; + impl xmtp_common::RetryableError for Boom { + fn is_retryable(&self) -> bool { + false + } + } + + let mut mock = MockNetworkClient::new(); + mock.expect_bidi_stream() + .return_once(|_req, _path, _body| Err(ApiClientError::client(Boom))); + + let client = V3Client::new(mock, NoCursorStore); + let outbound = stream::iter(vec![ping_req(1)]).boxed(); + + // `XmtpStream` isn't `Debug`, so match instead of `unwrap_err`. + match client.subscribe_bidi(outbound).await { + Err(ApiClientError::ClientWithEndpoint { endpoint, .. }) => { + assert_eq!(endpoint, "/xmtp.mls.api.v1.MlsApi/Subscribe"); + } + Err(other) => panic!("expected ClientWithEndpoint, got {other:?}"), + Ok(_) => panic!("subscribe_bidi should error when the transport fails to open"), + } + } +} diff --git a/crates/xmtp_api_grpc/src/grpc_client/client.rs b/crates/xmtp_api_grpc/src/grpc_client/client.rs index 5b65cd90a7..1c4c4b2141 100644 --- a/crates/xmtp_api_grpc/src/grpc_client/client.rs +++ b/crates/xmtp_api_grpc/src/grpc_client/client.rs @@ -80,12 +80,17 @@ impl GrpcClient { } } - /// Builds a tonic request from a body and a generic HTTP Request - fn build_tonic_request( + /// Builds a tonic request from a body and a generic HTTP Request. + /// + /// Generic over the body type `B` so the same path serves both the + /// unary / server-streaming transports (where `B` is `Bytes`) and the + /// XIP-83 bidirectional transport (where `B` is a `BoxDynStream` of + /// outbound frames). `tonic::Request` is happy with either. + fn build_tonic_request( &self, request: http::request::Builder, - body: Bytes, - ) -> Result, Status> { + body: B, + ) -> Result, Status> { let request = request .body(body) .map_err(|e| tonic::Status::from_error(Box::new(e)))?; @@ -187,6 +192,36 @@ impl Client for GrpcClient { }); Ok(response.to_http().map(Into::into)) } + + // Full-duplex needs a real HTTP/2 transport; the gRPC-Web service used on + // wasm cannot carry it, so the browser keeps the trait's default error. + #[cfg(not(target_arch = "wasm32"))] + async fn bidi_stream( + &self, + request: request::Builder, + path: PathAndQuery, + body: xmtp_common::BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + let this = self.clone(); + // client requires to be moved so it lives long enough for streaming response future. + let response = async move { + let mut client = this.inner.clone(); + this.wait_for_ready(&mut client).await?; + let request = this.build_tonic_request(request, body)?; + let codec = TransparentCodec::default(); + client.streaming(request, path, codec).await + }; + let req = crate::streams::NonBlockingStreamRequest::new( + Box::pin(response) as crate::streams::ResponseFuture + ); + let response = crate::streams::send(req).await.map_err(GrpcError::from)?; + let response = response.map(|body| { + BytesStream::new(GrpcStream { + inner: EscapableTonicStream::new(body), + }) + }); + Ok(response.to_http().map(Into::into)) + } } #[xmtp_common::async_trait] @@ -311,7 +346,7 @@ pub mod tests { let request = client .build_tonic_request( Default::default(), - PublishRequest { envelopes: vec![] }.encode_to_vec().into(), + prost::bytes::Bytes::from(PublishRequest { envelopes: vec![] }.encode_to_vec()), ) .unwrap(); diff --git a/crates/xmtp_proto/src/api_client.rs b/crates/xmtp_proto/src/api_client.rs index a0a0df5cdf..4e6374f10b 100644 --- a/crates/xmtp_proto/src/api_client.rs +++ b/crates/xmtp_proto/src/api_client.rs @@ -167,6 +167,30 @@ pub trait XmtpMlsStreams: MaybeSend + MaybeSync { ) -> Result; } +xmtp_common::if_native! { + /// The XIP-83 bidirectional subscription: one long-lived stream carrying + /// group and welcome messages, mutated in place (no reconnect on membership + /// change) and kept alive with WebSocket-style ping/pong. Native-only — + /// gRPC-Web transports cannot speak full-duplex, so browsers stay on + /// [`XmtpMlsStreams`] with a client-side watchdog. + #[xmtp_common::async_trait] + pub trait XmtpMlsBidiStreams: MaybeSend + MaybeSync { + type SubscribeStream: Stream> + + MaybeSend; + + type Error: RetryableError + 'static; + + /// Open the bidirectional stream. `requests` is the outbound + /// client→server frame stream (typically fed by a channel; the first + /// frame is usually a `Mutate` naming the initial topic set); the + /// returned stream yields the server→client frames. + async fn subscribe_bidi( + &self, + requests: futures::stream::BoxStream<'static, crate::mls_v1::SubscribeRequest>, + ) -> Result; + } +} + /// Represents the backend API required for the XMTP /// Identity Service described by [XIP-46 Multi-Wallet Identity](https://github.com/xmtp/XIPs/blob/main/XIPs/xip-46-multi-wallet-identity.md) #[xmtp_common::async_trait] diff --git a/crates/xmtp_proto/src/api_client/impls.rs b/crates/xmtp_proto/src/api_client/impls.rs index 2a7968d6dd..ff352b8dec 100644 --- a/crates/xmtp_proto/src/api_client/impls.rs +++ b/crates/xmtp_proto/src/api_client/impls.rs @@ -403,3 +403,41 @@ where .await } } + +xmtp_common::if_native! { + // `XmtpMlsBidiStreams` is native-only (see `api_client.rs`), so its boxed / + // arced forwarders are gated the same way. Without these, erasing a client + // into `Box` / `Arc` would + // fail to compile on `subscribe_bidi`, unlike every other client trait here. + #[xmtp_common::async_trait] + impl XmtpMlsBidiStreams for Box + where + T: XmtpMlsBidiStreams + Sync + ?Sized, + { + type Error = ::Error; + type SubscribeStream = ::SubscribeStream; + + async fn subscribe_bidi( + &self, + requests: futures::stream::BoxStream<'static, crate::mls_v1::SubscribeRequest>, + ) -> Result { + (**self).subscribe_bidi(requests).await + } + } + + #[xmtp_common::async_trait] + impl XmtpMlsBidiStreams for Arc + where + T: XmtpMlsBidiStreams + ?Sized, + { + type Error = ::Error; + type SubscribeStream = ::SubscribeStream; + + async fn subscribe_bidi( + &self, + requests: futures::stream::BoxStream<'static, crate::mls_v1::SubscribeRequest>, + ) -> Result { + (**self).subscribe_bidi(requests).await + } + } +} diff --git a/crates/xmtp_proto/src/traits.rs b/crates/xmtp_proto/src/traits.rs index a46bba0eb5..470464c6d6 100644 --- a/crates/xmtp_proto/src/traits.rs +++ b/crates/xmtp_proto/src/traits.rs @@ -143,6 +143,22 @@ pub trait Client: MaybeSend + MaybeSync { body: Bytes, ) -> Result, ApiClientError>; + /// Open a bidirectional stream (XIP-83). `body` is the outbound stream of + /// encoded protobuf messages (one `Bytes` item per message); the response + /// carries the inbound message stream. Transports without full-duplex + /// support (e.g. gRPC-Web in the browser) keep this default and error. + async fn bidi_stream( + &self, + request: request::Builder, + path: http::uri::PathAndQuery, + body: BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + let _ = (request, path, body); + Err(ApiClientError::OtherUnretryable( + "bidirectional streaming is not supported by this transport".into(), + )) + } + /// start a "fake" stream that does not create a TCP connection and will always be pending fn fake_stream(&self) -> http::Response { let fake = FakeEmptyStream::new(); @@ -179,6 +195,15 @@ where ) -> Result, ApiClientError> { (**self).stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: request::Builder, + path: http::uri::PathAndQuery, + body: BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + (**self).bidi_stream(request, path, body).await + } } #[xmtp_common::async_trait] @@ -203,6 +228,15 @@ where ) -> Result, ApiClientError> { (**self).stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: request::Builder, + path: http::uri::PathAndQuery, + body: BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + (**self).bidi_stream(request, path, body).await + } } #[xmtp_common::async_trait] @@ -227,6 +261,15 @@ where ) -> Result, ApiClientError> { (**self).stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: request::Builder, + path: PathAndQuery, + body: BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + (**self).bidi_stream(request, path, body).await + } } #[xmtp_common::async_trait] diff --git a/crates/xmtp_proto/src/traits/boxed_client.rs b/crates/xmtp_proto/src/traits/boxed_client.rs index 9507b0de39..9fe9bf1bd2 100644 --- a/crates/xmtp_proto/src/traits/boxed_client.rs +++ b/crates/xmtp_proto/src/traits/boxed_client.rs @@ -48,6 +48,15 @@ where ) -> Result, ApiClientError> { self.inner.stream(request, path, body).await } + + async fn bidi_stream( + &self, + request: request::Builder, + path: http::uri::PathAndQuery, + body: xmtp_common::BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError> { + self.inner.bidi_stream(request, path, body).await + } } pub trait ToBoxedClient { diff --git a/crates/xmtp_proto/src/traits/mock.rs b/crates/xmtp_proto/src/traits/mock.rs index c6adc54263..4300d14338 100644 --- a/crates/xmtp_proto/src/traits/mock.rs +++ b/crates/xmtp_proto/src/traits/mock.rs @@ -82,6 +82,13 @@ mockall::mock! { path: http::uri::PathAndQuery, body: Bytes, ) -> Result, ApiClientError>; + + async fn bidi_stream( + &self, + request: http::request::Builder, + path: http::uri::PathAndQuery, + body: xmtp_common::BoxDynStream<'static, Bytes>, + ) -> Result, ApiClientError>; } }