From c14831474260847c680b4fc5c1c5846aef111cd7 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Tue, 14 Jul 2026 00:14:53 +0200 Subject: [PATCH] [WIP] Pings, Pongs & Streams --- Cargo.lock | 1 + Cargo.toml | 18 +++- client/Cargo.toml | 4 +- client/src/lib.rs | 194 ++++++++++++++++++++++++++++++++---- codec/src/registry.rs | 38 ++++++- docs/NATIVE-CLIENT.md | 49 ++++++++- docs/NATIVE-HOST.md | 30 ++++++ docs/WASM-CLIENT.md | 33 ++++++ host/Cargo.toml | 3 + host/src/lib.rs | 74 +++++++++----- transport/Cargo.toml | 6 ++ transport/src/client.rs | 3 +- transport/src/connection.rs | 126 +++++++++++++++++++++-- wasm/src/client.rs | 13 +-- wasm/src/config.rs | 7 ++ wasm/src/lib.rs | 3 + wasm/src/transport.rs | 9 +- 17 files changed, 541 insertions(+), 70 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5eb8ab0..a507421 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -985,6 +985,7 @@ dependencies = [ "mtp-crypto", "mtp-files", "mtp-host", + "mtp-transport", "mtp-type-map", "rand 0.8.6", "rcgen", diff --git a/Cargo.toml b/Cargo.toml index daac4c7..8c0dbdf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -51,6 +51,7 @@ edition = "2024" mtp-common = { version = "0.1.0", path = "common" } mtp-type-map = { version = "0.1.0", path = "type-map" } mtp-codec = { version = "0.1.0", path = "codec" } +mtp-transport = { version = "0.1.0", path = "transport", optional = true } # --- optional, behind features --- mtp-crypto = { version = "0.1.0", path = "crypto", optional = true, features = [ "serde", @@ -76,10 +77,23 @@ crypto = [ ] # MTP server host - version negotiation, Registry, incoming QUIC connections. -host = ["dep:mtp-host", "mtp-codec/registry"] +host = ["dep:mtp-host", "mtp-codec/registry", "transport"] # MTP client - outgoing QUIC connections to a host. -client = ["dep:mtp-client"] +client = ["dep:mtp-client", "transport"] + +# Opt into stream-specific host/client facade APIs. The transport itself is +# always framed over QUIC/WebTransport streams for compatibility. +streaming = [ + "transport", + "mtp-transport/streaming", + "mtp-host?/streaming", + "mtp-client?/streaming", +] + +# Direct access to the framed QUIC transport. Host/client features enable it +# automatically; this feature is useful for low-level integrations. +transport = ["dep:mtp-transport"] # On-disk storage for keyrings (`.mk`) and public key bundles (`.mpkb`). # Pulls in `crypto` so the `Keyring` / `PublicKeyBundle` types are in scope. diff --git a/client/Cargo.toml b/client/Cargo.toml index 94f8fa2..b913851 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -9,7 +9,9 @@ mtp-codec = { version = "0.1.0", path = "../codec" } mtp-transport = { version = "0.1.0", path = "../transport" } mtp-crypto = { version = "0.1.0", path = "../crypto", optional = true } rand = "0.8" -tokio = { version = "1", features = ["time"] } +tokio = { version = "1", features = ["rt", "sync", "time"] } [features] crypto = ["dep:mtp-crypto", "mtp-codec/crypto"] +# Enables stream-specific convenience exports and configuration. +streaming = ["mtp-transport/streaming"] diff --git a/client/src/lib.rs b/client/src/lib.rs index e764f55..9c8b49b 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -1,12 +1,15 @@ use mtp_codec::{CommunicationValue, DataType, DataValue, PROTOCOL_VERSION, Version}; use mtp_common::CommunicationError; -#[cfg(feature = "crypto")] -use tokio::time::Duration; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::{mpsc, Mutex}; +use tokio::time::{Duration, Instant}; pub use MTPClient as Client; pub use MTPConnection as Connection; pub use mtp_transport::Policy; pub use mtp_transport::Receiver; +#[cfg(feature = "streaming")] pub use mtp_transport::SendMode; pub use mtp_transport::Sender; @@ -30,6 +33,9 @@ pub struct ClientConfig { pub client_id: u64, pub description: Option, pub policy: Policy, + pub ping_interval: Duration, + pub max_missed_pings: usize, + pub ping_timestamp: bool, #[cfg(feature = "crypto")] pub auth_timeout: Duration, } @@ -48,6 +54,9 @@ impl ClientConfig { client_id: 0, description: None, policy: Policy::default(), + ping_interval: Duration::ZERO, + max_missed_pings: 3, + ping_timestamp: true, #[cfg(feature = "crypto")] auth_timeout: Duration::from_secs(30), } @@ -77,6 +86,21 @@ impl ClientConfig { self } + pub fn with_ping_interval(mut self, interval: Duration) -> Self { + self.ping_interval = interval; + self + } + + pub fn with_max_missed_pings(mut self, max_missed_pings: usize) -> Self { + self.max_missed_pings = max_missed_pings; + self + } + + pub fn with_ping_timestamp(mut self, ping_timestamp: bool) -> Self { + self.ping_timestamp = ping_timestamp; + self + } + #[cfg(feature = "crypto")] pub fn with_auth_timeout(mut self, timeout: Duration) -> Self { self.auth_timeout = timeout; @@ -97,13 +121,36 @@ pub struct MTPConnection { pub sender: Sender, pub receiver: Receiver, pub description: Option, + ping: Option, #[cfg(feature = "crypto")] pub auth_state: AuthState, #[cfg(feature = "crypto")] pub client_id: u64, } +struct PingSession { + last_ping: Arc>>, + task: tokio::task::JoinHandle<()>, +} + +impl PingSession { + fn get_ping(&self) -> Option { + self.last_ping.try_lock().ok().and_then(|ping| *ping) + } +} + +impl Drop for PingSession { + fn drop(&mut self) { + self.task.abort(); + } +} + impl MTPConnection { + /* Returns the round-trip time for the latest Ping/Pong exchange. */ + pub fn get_ping(&self) -> Option { + self.ping.as_ref().and_then(PingSession::get_ping) + } + /* * Send a request frame and wait for the response with the same frame id. * Any expected response type is validated after the id match. Frames with @@ -148,6 +195,97 @@ impl MTPConnection { } } +fn start_ping_session( + config: &ClientConfig, + sender: Sender, + receiver: &Receiver, +) -> Option { + if config.ping_interval.is_zero() { + return None; + } + + let (pong_tx, mut pong_rx) = mpsc::unbounded_channel(); + receiver.observe_pongs(pong_tx); + let last_ping = Arc::new(Mutex::new(None)); + let ping_state = last_ping.clone(); + let interval = config.ping_interval; + let max_missed_pings = config.max_missed_pings; + let ping_timestamp = config.ping_timestamp; + let mut close_rx = receiver.handle().subscribe_close(); + + let task = tokio::spawn(async move { + let mut ticker = tokio::time::interval(interval); + ticker.tick().await; + let mut pending = HashMap::new(); + + loop { + tokio::select! { + _ = close_rx.changed() => { + if close_rx.borrow().is_some() { + break; + } + } + _ = ticker.tick() => { + if max_missed_pings > 0 && !pending.is_empty() && pending.len() >= max_missed_pings { + sender.close(); + break; + } + + let mut ping = CommunicationValue::new(mtp_codec::CommunicationType::Ping); + if ping_timestamp { + let sent_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis(); + ping = ping.add_typed_default( + DataType::Timestamp, + DataValue::UnsignedNumber(sent_at), + ); + } + let id = ping.get_id(); + if sender.send(&ping).await.is_err() { + sender.close(); + break; + } + pending.insert(id, Instant::now()); + } + pong = pong_rx.recv() => match pong { + Some(pong) => { + if let Some(sent_at) = pending.remove(&pong.get_id()) { + let mut last_ping = ping_state.lock().await; + *last_ping = Some(sent_at.elapsed()); + } + } + None => break, + }, + } + } + }); + + Some(PingSession { last_ping, task }) +} + +fn connection_from_parts( + config: ClientConfig, + sender: Sender, + receiver: Receiver, + #[cfg(feature = "crypto")] auth_state: AuthState, + #[cfg(feature = "crypto")] client_id: u64, +) -> MTPConnection { + let ping = start_ping_session(&config, sender.clone(), &receiver); + MTPConnection { + version: PROTOCOL_VERSION, + sender, + receiver, + description: config.description, + ping, + #[cfg(feature = "crypto")] + auth_state, + #[cfg(feature = "crypto")] + client_id, + } +} + #[cfg(feature = "crypto")] #[derive(Debug, Clone, PartialEq, Eq)] pub enum AuthState { @@ -183,16 +321,18 @@ impl MTPClient { sender.send(&ident).await?; - Ok(MTPConnection { - version: PROTOCOL_VERSION, + #[cfg(feature = "crypto")] + let client_id = config.client_id; + #[cfg(feature = "crypto")] + return Ok(connection_from_parts( + config, sender, receiver, - description: config.description, - #[cfg(feature = "crypto")] - auth_state: AuthState::Unauthenticated, - #[cfg(feature = "crypto")] - client_id: config.client_id, - }) + AuthState::Unauthenticated, + client_id, + )); + #[cfg(not(feature = "crypto"))] + Ok(connection_from_parts(config, sender, receiver)) } } @@ -488,14 +628,14 @@ impl MTPClient { return Err(e); } - Ok(MTPConnection { - version: PROTOCOL_VERSION, + let client_id = config.client_id; + Ok(connection_from_parts( + config, sender, receiver, - description: config.description, - auth_state: AuthState::Authenticated, - client_id: config.client_id, - }) + AuthState::Authenticated, + client_id, + )) } pub async fn auth_register( @@ -636,14 +776,13 @@ impl MTPClient { return Err(e); } - Ok(MTPConnection { - version: PROTOCOL_VERSION, + Ok(connection_from_parts( + config, sender, receiver, - description: config.description, - auth_state: AuthState::Authenticated, - client_id: assigned_id, - }) + AuthState::Authenticated, + assigned_id, + )) } } @@ -671,6 +810,17 @@ mod tests { assert_eq!(config.client_id, 42); } + #[test] + fn test_ping_config() { + let config = ClientConfig::new("https://localhost:4433") + .with_ping_interval(Duration::from_secs(5)) + .with_max_missed_pings(2) + .with_ping_timestamp(false); + assert_eq!(config.ping_interval, Duration::from_secs(5)); + assert_eq!(config.max_missed_pings, 2); + assert!(!config.ping_timestamp); + } + #[cfg(feature = "crypto")] #[test] fn test_auth_state_unauthenticated_is_not_authenticated() { diff --git a/codec/src/registry.rs b/codec/src/registry.rs index 78757bf..1d06860 100644 --- a/codec/src/registry.rs +++ b/codec/src/registry.rs @@ -1,4 +1,7 @@ -use mtp_type_map::Version; +use mtp_common::CodecError; +use mtp_type_map::{PROTOCOL_VERSION, TypeMap, Version}; + +use crate::CommunicationValue; pub use mtp_type_map::Registry; @@ -9,11 +12,42 @@ pub use mtp_type_map::Registry; #[derive(Clone, Debug)] pub struct VersionedCodec { registry: Registry, + type_map: TypeMap, } impl VersionedCodec { pub fn new(registry: Registry) -> Self { - Self { registry } + let type_map = registry + .latest() + .cloned() + .unwrap_or_else(|| TypeMap::new(PROTOCOL_VERSION)); + Self { registry, type_map } + } + + /// Create a codec bound to a negotiated protocol version. + pub fn for_version(registry: Registry, version: Version) -> Option { + let type_map = registry.get(&version)?.clone(); + Some(Self { registry, type_map }) + } + + /// Return the type map used by this codec. + pub fn type_map(&self) -> &TypeMap { + &self.type_map + } + + /// Return the protocol version used by this codec. + pub fn version(&self) -> &Version { + &self.type_map.version + } + + /// Encode a value using the codec's negotiated framing rules. + pub fn encode(&self, value: &CommunicationValue) -> Result, CodecError> { + value.to_bytes() + } + + /// Decode a frame and retain the negotiated type map for typed access. + pub fn decode(&self, bytes: &[u8]) -> Result { + CommunicationValue::from_bytes_with(bytes, &self.type_map) } pub fn negotiate(&self, client_versions: &[Version]) -> Option { diff --git a/docs/NATIVE-CLIENT.md b/docs/NATIVE-CLIENT.md index 027aa29..421828e 100644 --- a/docs/NATIVE-CLIENT.md +++ b/docs/NATIVE-CLIENT.md @@ -18,10 +18,14 @@ mtp = { path = "/path/to/mtp", features = ["client", "crypto"] } ```rust use mtp::client::{ClientConfig, ClientTlsConfig}; +use std::time::Duration; let config = ClientConfig::new("https://host.example.com:4433") .with_tls(ClientTlsConfig::SystemRoots) - .with_client_id(0); + .with_client_id(0) + .with_ping_interval(Duration::from_secs(5)) + .with_max_missed_pings(3) + .with_ping_timestamp(true); ``` | Field | Type | Description | @@ -30,6 +34,9 @@ let config = ClientConfig::new("https://host.example.com:4433") | `tls` | `ClientTlsConfig` | `SystemRoots` or `PinnedPem(pem_bytes)` | | `client_id` | `u64` | Client identifier (ignored during `auth_register`) | | `description` | `Option` | Optional label sent during handshake (e.g. `"phone"`) | +| `ping_interval` | `Duration` | Interval between Ping frames; zero disables pings | +| `max_missed_pings` | `usize` | Unanswered Ping frames allowed before the connection closes | +| `ping_timestamp` | `bool` | Adds a `Timestamp` entry to each Ping frame | | `auth_timeout` | `Duration` (crypto) | Authentication handshake timeout (default 30s) | ### TLS Certificate Handling @@ -74,6 +81,46 @@ pub struct MTPConnection { - `description` -- the label sent during handshake (set via `ClientConfig::with_description`) - `client_id` -- the confirmed/assigned client identifier (crypto only) +When `ping_interval` is non-zero, MTP sends Ping frames in the background and +consumes their Pong responses before application message handling. `get_ping()` +returns the round-trip duration of the latest matched Pong, or `None` until a +Pong arrives. A connection closes when the configured unanswered Ping limit is +reached. + +### Ping-Pong + +Ping/Pong is part of the protocol, not just a transport keepalive. Each Ping +frame is matched against a Pong with the same frame id, and the client uses the +response to update `get_ping()`. If the host does not answer within the +configured limit, the connection closes. + +Enable it in `ClientConfig`, then inspect the latest round-trip time on the +connection. Pings start after the connection has been established; `None` is +normal until the first matching Pong arrives. + +```rust +use mtp::client::{ClientConfig, MTPClient}; +use std::time::Duration; + +let config = ClientConfig::new("https://host.example.com:4433") + .with_client_id(42) + .with_ping_interval(Duration::from_secs(5)) + .with_max_missed_pings(3) + .with_ping_timestamp(true); + +let conn = MTPClient::connect(config).await?; + +if let Some(round_trip) = conn.get_ping() { + println!("latest MTP round trip: {round_trip:?}"); +} +``` + +The client consumes the Pong frames used by this loop, so they are not returned +by `conn.receiver.receive()`. Set `ping_interval` to `Duration::ZERO` (the +default) to disable protocol pings. `max_missed_pings` is the number of +outstanding Ping frames allowed before the client closes the connection; use a +host with automatic Pong responses, or provide an equivalent responder. + ### Unauthenticated Connect ```rust diff --git a/docs/NATIVE-HOST.md b/docs/NATIVE-HOST.md index 631c1ea..324cd66 100644 --- a/docs/NATIVE-HOST.md +++ b/docs/NATIVE-HOST.md @@ -48,6 +48,7 @@ let config = HostConfig::new( | `port` | `u16` | Listen port | | `tls_fullchain` | `Vec` | PEM-encoded TLS certificate chain | | `tls_key` | `Vec` | PEM-encoded TLS private key | +| `send_pongs` | `bool` | Sends a Pong for each received Ping (default `true`) | | `authentication_policy` | `AuthenticationPolicy` (crypto) | `ForceAuthentication`, `AllowAuthentication`, or `Unauthenticated` | | `host_keyring` | `Keyring` (crypto) | Host's signing and KEM keys | | `get_existing_user` | `Fn(u64) -> Pin> + Send>> + Send + Sync` (crypto) | Async lookup callback for login | @@ -77,6 +78,35 @@ let config = HostConfig::new(ip, port, cert, key); The host requires a TLS certificate. For development, generate a self-signed certificate using `rcgen`. For production, use a CA-signed certificate. +### Ping-Pong + +The host handles protocol Ping/Pong automatically unless you disable it with +`with_pongs(false)`. Enable the default responder explicitly when constructing +the host if you want to make the choice visible in application configuration: + +```rust +let config = HostConfig::new(ip, port, cert, key) + .with_pongs(true); +``` + +For every received Ping, the responder sends a Pong with the same frame id and +copies the optional `Timestamp` data entry. Ping and Pong frames handled this +way are not delivered by `conn.receiver.receive()`. This lets native clients +use `ClientConfig::with_ping_interval` and `MTPConnection::get_ping()` without +adding application-level handlers. + +Disable it only when the application needs to handle Ping frames itself: + +```rust +let config = HostConfig::new(ip, port, cert, key) + .with_pongs(false); +``` + +With automatic responses disabled, Ping frames are delivered through the normal +receiver and the application is responsible for sending a compatible Pong (the +same frame id, and normally the Ping's `Timestamp`) if it wants clients to +continue their protocol ping loop. + ## Accepting Connections ```rust diff --git a/docs/WASM-CLIENT.md b/docs/WASM-CLIENT.md index 3cd4699..2802036 100644 --- a/docs/WASM-CLIENT.md +++ b/docs/WASM-CLIENT.md @@ -149,6 +149,39 @@ If hashes are omitted, the browser uses its normal TLS root store. `maxMessageSize` caps inbound and outbound MTP frames before buffering/sending. `authTimeoutMs` bounds connect/login/register promises at the SDK layer. +## Streams + +The browser client uses one WebTransport session per `MTPClient` instance. +`send()`, `request()`, and `subscribe()` all operate over that session; the SDK +does not expose browser stream objects directly. + +Use the normal message APIs to send and receive over that session: + +```typescript +const client = await MTPClient.create({ url, hostPublicKey }); +await client.connect(); + +const unsubscribe = client.subscribe("SomeType", (message) => { + console.log(message.data); +}); + +await client.send("SomeType", { value: "hello" }); +unsubscribe(); +``` + +Internally, each outbound MTP frame is written to a new WebTransport +unidirectional stream as a four-byte big-endian length followed by the frame, +then that stream is closed. Incoming frames are read from the session's +incoming unidirectional streams. The reader accepts both one-frame streams and +native peers that place several frames on a persistent stream, so browser and +native clients interoperate without stream configuration. + +The SDK deliberately owns stream lifetime and framing. Do not create browser +streams for MTP frames yourself through the SDK. For direct generated bindings, +use `client.raw.client` or import `WasmClient` from `mtp/raw`; a `WasmClient` +still owns one active WebTransport session, so create another instance for an +independent connection. + ## Sending, Requests, Subscriptions, And Pings `send` accepts either a typed message or a prebuilt raw frame: diff --git a/host/Cargo.toml b/host/Cargo.toml index ea4e661..71cb38c 100644 --- a/host/Cargo.toml +++ b/host/Cargo.toml @@ -13,3 +13,6 @@ tokio = { version = "1", features = ["time"] } [features] crypto = ["dep:mtp-crypto", "mtp-codec/crypto"] +# Stream mode is optional at the facade boundary. The underlying transport +# remains framed/stream based so the default API stays backwards compatible. +streaming = ["mtp-transport/streaming"] diff --git a/host/src/lib.rs b/host/src/lib.rs index 3650385..44d73a9 100644 --- a/host/src/lib.rs +++ b/host/src/lib.rs @@ -14,6 +14,7 @@ pub use MTPConnection as Connection; pub use MTPHost as Host; pub use mtp_transport::Policy; pub use mtp_transport::Receiver; +#[cfg(feature = "streaming")] pub use mtp_transport::SendMode; pub use mtp_transport::Sender; @@ -55,6 +56,7 @@ pub struct HostConfig { pub tls_key: Vec, pub policy: Policy, + pub send_pongs: bool, #[cfg(feature = "crypto")] pub authentication_policy: AuthenticationPolicy, @@ -76,6 +78,7 @@ impl HostConfig { tls_fullchain, tls_key, policy: Policy::default(), + send_pongs: true, #[cfg(feature = "crypto")] authentication_policy: AuthenticationPolicy::Unauthenticated, #[cfg(feature = "crypto")] @@ -101,6 +104,11 @@ impl HostConfig { self } + pub fn with_pongs(mut self, send_pongs: bool) -> Self { + self.send_pongs = send_pongs; + self + } + #[cfg(feature = "crypto")] pub fn with_authentication( mut self, @@ -186,7 +194,6 @@ pub struct MTPConnection { pub struct MTPHost { transport: mtp_transport::Host, registry: Registry, - #[cfg(feature = "crypto")] config: HostConfig, } @@ -206,7 +213,6 @@ impl MTPHost { Ok(Self { transport, registry, - #[cfg(feature = "crypto")] config, }) } @@ -228,7 +234,7 @@ impl MTPHost { match self.config.authentication_policy { AuthenticationPolicy::ForceAuthentication => { let timeout = self.config.auth_timeout; - return match tokio::time::timeout( + let connection = match tokio::time::timeout( timeout, self.accept_authenticated(sender, receiver), ) @@ -237,9 +243,11 @@ impl MTPHost { Ok(result) => result, Err(_) => Err(AcceptError::AuthenticationTimedOut), }; + return Ok(self.configure_pongs(connection?)); } AuthenticationPolicy::AllowAuthentication => { - return self.accept_allow_auth(sender, receiver).await; + let connection = self.accept_allow_auth(sender, receiver).await?; + return Ok(self.configure_pongs(connection)); } AuthenticationPolicy::Unauthenticated => { let first_msg = match receiver.receive().await { @@ -265,12 +273,13 @@ impl MTPHost { Some(v) => v, None => return Err(AcceptError::UnsupportedVersion(client_version)), }; - let codec = VersionedCodec::new(self.registry.clone()); + let codec = VersionedCodec::for_version(self.registry.clone(), negotiated.clone()) + .expect("negotiated version must be registered"); let description = match first_msg.get_data(DataType::Description) { DataValue::Str(s) => Some(s.clone()), _ => None, }; - Ok(Some(MTPConnection { + Ok(self.configure_pongs(Some(MTPConnection { version: negotiated, codec, sender, @@ -282,7 +291,7 @@ impl MTPHost { client_id: rand::random(), #[cfg(feature = "crypto")] client_public_key: None, - })) + }))) } } @@ -304,18 +313,19 @@ impl MTPHost { Some(v) => v, None => return Err(AcceptError::UnsupportedVersion(client_version)), }; - let codec = VersionedCodec::new(self.registry.clone()); + let codec = VersionedCodec::for_version(self.registry.clone(), negotiated.clone()) + .expect("negotiated version must be registered"); let description = match first_msg.get_data(DataType::Description) { DataValue::Str(s) => Some(s.clone()), _ => None, }; - return Ok(Some(MTPConnection { + return Ok(self.configure_pongs(Some(MTPConnection { version: negotiated, codec, sender, receiver, description, - })); + }))); } } @@ -326,6 +336,19 @@ impl MTPHost { pub fn registry(&self) -> &Registry { &self.registry } + + fn configure_pongs(&self, connection: Option) -> Option { + if let Some(connection) = connection { + if self.config.send_pongs { + connection + .receiver + .respond_to_pings(connection.sender.clone()); + } + Some(connection) + } else { + None + } + } } #[cfg(feature = "crypto")] @@ -480,6 +503,13 @@ impl MTPHost { }; let tm = mtp_codec::TypeMap::latest(); + let negotiated = self + .registry + .negotiate(std::slice::from_ref(&client_version)) + .ok_or_else(|| { + sender.close(); + AcceptError::UnsupportedVersion(client_version.clone()) + })?; let pq_enabled = !self .config .host_keyring @@ -636,18 +666,8 @@ impl MTPHost { return Err(AcceptError::Send(e)); } - // ===== Version negotiation ===== - let negotiated = match self - .registry - .negotiate(std::slice::from_ref(&client_version)) - { - Some(v) => v, - None => { - sender.close(); - return Err(AcceptError::UnsupportedVersion(client_version)); - } - }; - let codec = VersionedCodec::new(self.registry.clone()); + let codec = VersionedCodec::for_version(self.registry.clone(), negotiated.clone()) + .expect("negotiated version must be registered"); Ok(Some(MTPConnection { version: negotiated, @@ -761,7 +781,8 @@ impl MTPHost { Some(v) => v, None => return Err(AcceptError::UnsupportedVersion(client_version)), }; - let codec = VersionedCodec::new(self.registry.clone()); + let codec = VersionedCodec::for_version(self.registry.clone(), negotiated.clone()) + .expect("negotiated version must be registered"); return Ok(Some(MTPConnection { version: negotiated, codec, @@ -842,4 +863,11 @@ mod tests { assert_ne!(AuthState::Unauthenticated, AuthState::Authenticated); assert_ne!(AuthState::Pending, AuthState::Authenticated); } + + #[test] + fn host_config_pongs_default_to_enabled() { + let config = HostConfig::new("127.0.0.1".parse().unwrap(), 4433, Vec::new(), Vec::new()); + assert!(config.send_pongs); + assert!(!config.with_pongs(false).send_pongs); + } } diff --git a/transport/Cargo.toml b/transport/Cargo.toml index 5ff48ac..69c85b4 100644 --- a/transport/Cargo.toml +++ b/transport/Cargo.toml @@ -27,3 +27,9 @@ required-features = ["host"] default = [] # Enables hosting a MTP server host = [] +# Documents and exposes the framed uni-directional stream transport used by +# the host and client facades. The transport is stream based by design, so +# keeping this as a marker feature lets facade crates opt into their +# stream-specific convenience exports without making the core transport +# unusable for existing consumers. +streaming = [] diff --git a/transport/src/client.rs b/transport/src/client.rs index f356ea5..e0a7b68 100644 --- a/transport/src/client.rs +++ b/transport/src/client.rs @@ -2,10 +2,9 @@ use std::sync::Arc; use mtp_common::CommunicationError; use rustls::{ - ClientConfig as RustlsClientConfig, RootCertStore, + ClientConfig as RustlsClientConfig, DigitallySignedStruct, RootCertStore, SignatureScheme, client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier}, pki_types::{ServerName, UnixTime, pem::PemObject}, - DigitallySignedStruct, SignatureScheme, }; use wtransport::{ClientConfig, Endpoint}; diff --git a/transport/src/connection.rs b/transport/src/connection.rs index 8a88fc3..b4d6e8a 100644 --- a/transport/src/connection.rs +++ b/transport/src/connection.rs @@ -2,7 +2,7 @@ use crate::ConnectionHandle; use mtp_codec::CommunicationValue; use mtp_common::CommunicationError; use std::sync::Arc; -use tokio::sync::{Mutex, mpsc}; +use tokio::sync::{mpsc, Mutex, RwLock, Semaphore}; use tokio::time::{Duration, sleep, timeout}; use wtransport::Connection; @@ -29,7 +29,10 @@ pub struct Policy { pub force_close_delay: Duration, pub max_transient_recv_errors: usize, pub transient_recv_backoff: Duration, + pub persistent_stream_max_retries: usize, + pub persistent_stream_retry_backoff: Duration, pub receiver_queue_capacity: usize, + pub max_concurrent_stream_tasks: usize, } impl Default for Policy { @@ -48,7 +51,10 @@ impl Default for Policy { force_close_delay: Duration::from_millis(300), max_transient_recv_errors: 20, transient_recv_backoff: Duration::from_millis(100), + persistent_stream_max_retries: 4, + persistent_stream_retry_backoff: Duration::from_millis(20), receiver_queue_capacity: 1000, + max_concurrent_stream_tasks: 128, } } } @@ -90,6 +96,24 @@ impl Policy { self.receiver_queue_capacity = receiver_queue_capacity; self } + + pub fn with_persistent_stream_retries( + mut self, + persistent_stream_max_retries: usize, + persistent_stream_retry_backoff: Duration, + ) -> Self { + self.persistent_stream_max_retries = persistent_stream_max_retries; + self.persistent_stream_retry_backoff = persistent_stream_retry_backoff; + self + } + + pub fn with_max_concurrent_stream_tasks( + mut self, + max_concurrent_stream_tasks: usize, + ) -> Self { + self.max_concurrent_stream_tasks = max_concurrent_stream_tasks; + self + } } enum ReceivedFrame { @@ -98,8 +122,9 @@ enum ReceivedFrame { Idle, } +#[derive(Clone)] pub struct Sender { - send_guard: Mutex<()>, + send_guard: Arc>, stream_guard: Arc>>, handle: Arc, connection: Connection, @@ -109,7 +134,7 @@ pub struct Sender { impl Sender { pub fn new(connection: Connection, handle: Arc, policy: Arc) -> Self { Self { - send_guard: Mutex::new(()), + send_guard: Arc::new(Mutex::new(())), stream_guard: Arc::new(Mutex::new(None)), handle, connection, @@ -208,23 +233,24 @@ impl Sender { return Err(CommunicationError::StreamClosed); } - let res = { - let stream = Self::ensure_stream(conn, stream_opt, policy).await?; - Self::write_frame(stream, data, policy).await + let res = match Self::ensure_stream(conn, stream_opt, policy).await { + Ok(stream) => Self::write_frame(stream, data, policy).await, + Err(e) => Err(e), }; if res.is_ok() { return Ok(()); } + let err = res.err().unwrap_or(CommunicationError::StreamError); *stream_opt = None; tries += 1; - if tries >= 4 { - let stream = Self::ensure_stream(conn, stream_opt, policy).await?; - return Self::write_frame(stream, data, policy).await; + if tries > policy.persistent_stream_max_retries { + return Err(err); } - tokio::time::sleep(std::time::Duration::from_millis(20 * tries as u64)).await; + let backoff = policy.persistent_stream_retry_backoff * tries as u32; + tokio::time::sleep(backoff).await; } } @@ -433,6 +459,13 @@ pub struct Receiver { rx: Mutex>>, _accept_task: tokio::task::JoinHandle<()>, handle: Arc, + ping_control: Arc>, +} + +#[derive(Clone, Default)] +struct PingControl { + pong_sender: Option, + pong_observer: Option>, } impl Drop for Receiver { @@ -456,6 +489,10 @@ impl Receiver { let conn_handle = handle.clone(); let accept_connection = connection.clone(); let accept_policy = policy.clone(); + let ping_control = Arc::new(RwLock::new(PingControl::default())); + let accept_ping_control = ping_control.clone(); + let stream_limit = Arc::new(Semaphore::new(policy.max_concurrent_stream_tasks.max(1))); + let accept_stream_limit = stream_limit.clone(); let accept_task = tokio::spawn(async move { let mut close_rx = conn_handle.subscribe_close(); @@ -474,15 +511,62 @@ impl Receiver { ) => { match accepted { Ok(Ok(stream)) => { + let permit = match accept_stream_limit.clone().acquire_owned().await { + Ok(permit) => permit, + Err(_) => break, + }; let tx_stream = tx.clone(); let stream_handle = conn_handle.clone(); let stream_policy = accept_policy.clone(); + let stream_ping_control = accept_ping_control.clone(); tokio::spawn(async move { + let _permit = permit; let mut s = stream; loop { match Self::read_one_frame(&mut s, &stream_policy).await { Ok(ReceivedFrame::Message(msg)) => { + let ping_type = mtp_codec::CommunicationType::Ping + .to_id(&mtp_codec::TypeMap::latest()); + let pong_type = mtp_codec::CommunicationType::Pong + .to_id(&mtp_codec::TypeMap::latest()); + let control = { + let control = stream_ping_control.read().await; + if msg.get_type() == ping_type { + control + .pong_sender + .clone() + .map(|sender| (Some(sender), None)) + } else if msg.get_type() == pong_type { + control + .pong_observer + .clone() + .map(|observer| (None, Some(observer))) + } else { + None + } + }; + + if let Some((Some(sender), _)) = control { + let mut pong = CommunicationValue::new(mtp_codec::CommunicationType::Pong) + .with_id(msg.get_id()); + if let Some(timestamp) = msg.get_data_opt(mtp_codec::DataType::Timestamp) { + pong = pong.add_typed_default( + mtp_codec::DataType::Timestamp, + timestamp.clone(), + ); + } + if let Err(e) = sender.send(&pong).await { + log::warn!("[Receiver] failed to send Pong: {e}"); + } + continue; + } + + if let Some((_, Some(observer))) = control { + let _ = observer.send(msg); + continue; + } + if tx_stream.send(Ok(msg)).await.is_err() { break; } @@ -545,6 +629,25 @@ impl Receiver { rx: Mutex::new(rx), _accept_task: accept_task, handle, + ping_control, + } + } + + /* Respond to reserved Ping frames without exposing them to application I/O. */ + pub fn respond_to_pings(&self, sender: Sender) { + if let Ok(mut control) = self.ping_control.try_write() { + control.pong_sender = Some(sender); + } else { + log::warn!("[Receiver] could not register Ping responder: control lock busy"); + } + } + + /* Route reserved Pong frames to a connection-level observer. */ + pub fn observe_pongs(&self, observer: mpsc::UnboundedSender) { + if let Ok(mut control) = self.ping_control.try_write() { + control.pong_observer = Some(observer); + } else { + log::warn!("[Receiver] could not register Pong observer: control lock busy"); } } @@ -689,7 +792,10 @@ mod tests { assert_eq!(p.force_close_delay, Duration::from_millis(300)); assert_eq!(p.max_transient_recv_errors, 20); assert_eq!(p.transient_recv_backoff, Duration::from_millis(100)); + assert_eq!(p.persistent_stream_max_retries, 4); + assert_eq!(p.persistent_stream_retry_backoff, Duration::from_millis(20)); assert_eq!(p.receiver_queue_capacity, 1000); + assert_eq!(p.max_concurrent_stream_tasks, 128); } #[test] diff --git a/wasm/src/client.rs b/wasm/src/client.rs index 280bbf1..503d1e9 100644 --- a/wasm/src/client.rs +++ b/wasm/src/client.rs @@ -267,16 +267,17 @@ pub struct WasmClient { impl WasmClient { #[wasm_bindgen(constructor)] pub fn new( - on_state_change: &js_sys::Function, - on_message: &js_sys::Function, - on_error: &js_sys::Function, + on_state_change: Option, + on_message: Option, + on_error: Option, ) -> Self { + let noop = || js_sys::Function::new_no_args(""); Self { transport: Rc::new(RefCell::new(None)), state: Rc::new(Cell::new(ConnectionState::Disconnected)), - on_state_change: on_state_change.clone(), - on_message: on_message.clone(), - on_error: on_error.clone(), + on_state_change: on_state_change.unwrap_or_else(noop), + on_message: on_message.unwrap_or_else(noop), + on_error: on_error.unwrap_or_else(noop), subscriptions: Rc::new(RefCell::new(HashMap::new())), next_subscription_id: Rc::new(Cell::new(1)), pending_requests: Rc::new(RefCell::new(HashMap::new())), diff --git a/wasm/src/config.rs b/wasm/src/config.rs index 8ac13b5..46cdf5e 100644 --- a/wasm/src/config.rs +++ b/wasm/src/config.rs @@ -9,6 +9,13 @@ pub struct ConnectionConfig { pub(crate) description: Option, } +/// Newer API name for the browser connection configuration. +/// +/// `ConnectionConfig` remains the concrete wasm-bindgen class for backwards +/// compatibility with the existing raw JavaScript bindings. The alias keeps +/// Rust consumers aligned with the native/WASM naming used by the public API. +pub type WasmClientConfig = ConnectionConfig; + #[wasm_bindgen] impl ConnectionConfig { #[wasm_bindgen(constructor)] diff --git a/wasm/src/lib.rs b/wasm/src/lib.rs index 5b49dc6..4c9f783 100644 --- a/wasm/src/lib.rs +++ b/wasm/src/lib.rs @@ -7,6 +7,9 @@ pub mod logging; pub mod subscription; pub mod transport; +pub use client::WasmClient; +pub use config::{ConnectionConfig, WasmClientConfig}; + #[cfg(not(test))] use wasm_bindgen::prelude::*; diff --git a/wasm/src/transport.rs b/wasm/src/transport.rs index 6860552..4ccb14c 100644 --- a/wasm/src/transport.rs +++ b/wasm/src/transport.rs @@ -402,7 +402,14 @@ impl WasmTransport { if let Some(reader) = self.stream_reader.borrow_mut().take() { release_reader_lock(&reader); } - self.buffer.borrow_mut().clear(); + // A frame is never allowed to span stream boundaries. The + // native persistent-stream sender packs frames on one + // stream, while the WASM sender uses one stream per frame; + // either mode must reject a truncated frame instead of + // silently dropping its prefix. + if !self.buffer.borrow().is_empty() { + return Err(js_error("stream ended in the middle of a frame")); + } } } }