diff --git a/crates/mqp2p/README.md b/crates/mqp2p/README.md index 4d0566d..ec82862 100644 --- a/crates/mqp2p/README.md +++ b/crates/mqp2p/README.md @@ -44,6 +44,13 @@ This RTT-synchronized approach (similar to [libp2p DCUtR](https://github.com/lib - QUIC mutual TLS verifies both sides match the expected fingerprint (TOFU model) - File integrity is verified end-to-end with SHA-256 +### Liveness + +- **Lease:** a registered peer's record carries `_expires_at`, set `lease_secs` ahead (default 15 s, `PeerConfig::with_lease_secs`), and a heartbeat renews it every third of the lease. Discovery skips records whose lease has passed, so a peer that vanishes without deregistering disappears within one lease. +- **MQTT keepalive:** the signaling client's keepalive defaults to a third of the lease (5 s for the default lease), The mqtt5 broker drops a client after 1.5 keepalives of silence and checks once per keepalive, so it notices a dead link 1.5 to 2.5 keepalives after it goes quiet: 7.5 to 12.5 s with the default lease, where mqtt5's own 60 s default took about 115 s for a partitioned peer in the c2-fusion harness. Raise it with `PeerConfig::with_keepalive_secs` on links where pings are expensive. +- **Clean shutdown:** `Peer::shutdown` deregisters the peer and sends MQTT `DISCONNECT`; dropping a `Peer` inside a Tokio runtime also sends `DISCONNECT`, so a stopped peer never leaves its client connected and reconnecting under its client id. +- **Presence (optional):** with `PeerConfig::with_presence` and a broker started with `--presence`, the peer follows MQDB's presence feed (`$DB/_presence/#`) and discovery also skips a peer whose MQTT client (`mqp2p-`) has disconnected, until it connects again. This removes a stopped or partitioned peer as soon as the broker sees its connection end; the lease stays as the backstop when the broker runs without `--presence`. + ## Usage ### As a library @@ -95,10 +102,12 @@ mqp2p uses [MQDB](https://github.com/LabOverWire/MQDB) as both the MQTT broker a # Generate password file mqdb passwd -b mypassword -n myuser >> passwd.txt -# Start broker with authentication +# Start broker with authentication (add --presence for PeerConfig::with_presence) mqdb agent start --bind 0.0.0.0:1883 --passwd passwd.txt ``` +With an ACL file, grant each peer user readwrite on `$DB/peers/#`, `mqp2p/responses/#` and `p2p/#`, and read on `$DB/_presence/#` when presence is used. MQDB 0.8.33 and later sends a response only to a topic the requester may publish to, so read-only access to `mqp2p/responses/#` makes every registry request time out. The presence feed shows every client id and user name on the broker, so grant it only to users that may see them. + ## Architecture ``` diff --git a/crates/mqp2p/src/peer.rs b/crates/mqp2p/src/peer.rs index 591872e..612c6ce 100644 --- a/crates/mqp2p/src/peer.rs +++ b/crates/mqp2p/src/peer.rs @@ -2,7 +2,7 @@ use crate::error::{Error, Result}; use crate::quic::{self, CertIdentity, QuicEndpoint, SegmentationOffload}; use crate::signaling::{ Candidate, CandidateKind, ConnectionOffer, ConnectionSync, MIN_LEASE_SECS, PEER_LEASE_SECS, - PeerInfo, SignalingClient, heartbeat_interval_secs, + PeerInfo, SignalingClient, client_id_for, heartbeat_interval_secs, }; use crate::stun; use crate::transfer::{self, FileOffer, TransferProgress, TransferResult}; @@ -11,6 +11,7 @@ use mqtt5::client::MqttClient; use quinn::Connection; use std::net::SocketAddr; use std::path::Path; +use std::time::Duration; use tracing::{debug, info, warn}; pub struct PeerConfig { @@ -20,6 +21,8 @@ pub struct PeerConfig { pub stun_server: Option, pub credentials: Option<(String, String)>, pub lease_secs: u64, + pub keepalive_secs: Option, + pub presence: bool, pub segmentation_offload: SegmentationOffload, } @@ -32,6 +35,8 @@ impl PeerConfig { stun_server: Some(stun::DEFAULT_STUN_SERVER.into()), credentials: None, lease_secs: PEER_LEASE_SECS, + keepalive_secs: None, + presence: false, segmentation_offload: SegmentationOffload::Enabled, } } @@ -41,6 +46,25 @@ impl PeerConfig { self } + pub fn with_keepalive_secs(mut self, keepalive_secs: u64) -> Self { + self.keepalive_secs = Some(keepalive_secs); + self + } + + pub fn with_presence(mut self) -> Self { + self.presence = true; + self + } + + fn effective_lease_secs(&self) -> u64 { + self.lease_secs.max(MIN_LEASE_SECS) + } + + fn effective_keepalive_secs(&self) -> u64 { + self.keepalive_secs + .unwrap_or_else(|| heartbeat_interval_secs(self.effective_lease_secs())) + } + pub fn with_bind_addr(mut self, addr: SocketAddr) -> Self { self.bind_addr = addr; self @@ -135,15 +159,24 @@ impl Peer { config.segmentation_offload, )?; - let client_id = format!("mqp2p-{}", config.name); - let mqtt = if let Some((ref user, ref pass)) = config.credentials { - let opts = ConnectOptions::new(client_id).with_credentials(user, pass.as_bytes()); - MqttClient::with_options(opts) - } else { - MqttClient::new(client_id) + let lease_secs = config.effective_lease_secs(); + let keepalive_secs = config.effective_keepalive_secs(); + let options = ConnectOptions::new(client_id_for(&config.name)) + .with_keep_alive(Duration::from_secs(keepalive_secs)); + let options = match config.credentials { + Some((ref user, ref pass)) => options.with_credentials(user, pass.as_bytes()), + None => options, }; - let signaling = SignalingClient::new(mqtt); + let signaling = SignalingClient::new(MqttClient::with_options(options)); signaling.connect(&config.broker_addr).await?; + if config.presence { + signaling.track_presence().await?; + } + debug!( + keepalive_secs, + presence = config.presence, + "signaling configured" + ); Ok(Self { name: config.name, @@ -155,7 +188,7 @@ impl Peer { std_socket, host_addr, srflx_addr, - lease_secs: config.lease_secs.max(MIN_LEASE_SECS), + lease_secs, heartbeat: None, }) } @@ -450,8 +483,13 @@ impl Peer { if let Some(handle) = self.heartbeat.take() { handle.abort(); } - if let Some(peer_id) = &self.peer_id { - let _ = self.signaling.deregister_peer(peer_id).await; + if let Some(peer_id) = &self.peer_id + && let Err(e) = self.signaling.deregister_peer(peer_id).await + { + warn!(peer_id, error = %e, "peer deregistration failed at shutdown"); + } + if let Err(e) = self.signaling.disconnect().await { + warn!(error = %e, "signaling disconnect failed at shutdown"); } self.quic.close(); info!("peer shut down"); @@ -464,6 +502,16 @@ impl Drop for Peer { if let Some(handle) = self.heartbeat.take() { handle.abort(); } + if let Ok(runtime) = tokio::runtime::Handle::try_current() { + let signaling = self.signaling.clone(); + runtime.spawn(async move { + if signaling.mqtt_client().is_connected().await + && let Err(e) = signaling.disconnect().await + { + debug!(error = %e, "signaling disconnect failed on drop"); + } + }); + } } } @@ -528,3 +576,31 @@ impl P2pConnection { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn keepalive_defaults_to_a_third_of_the_lease() { + let config = PeerConfig::new("edge", "127.0.0.1:1883"); + assert_eq!(config.effective_keepalive_secs(), PEER_LEASE_SECS / 3); + let config = config.with_lease_secs(30); + assert_eq!(config.effective_keepalive_secs(), 10); + } + + #[test] + fn keepalive_follows_the_lease_floor() { + let config = PeerConfig::new("edge", "127.0.0.1:1883").with_lease_secs(0); + assert_eq!(config.effective_lease_secs(), MIN_LEASE_SECS); + assert_eq!(config.effective_keepalive_secs(), 1); + } + + #[test] + fn explicit_keepalive_overrides_the_default() { + let config = PeerConfig::new("edge", "127.0.0.1:1883") + .with_lease_secs(15) + .with_keepalive_secs(60); + assert_eq!(config.effective_keepalive_secs(), 60); + } +} diff --git a/crates/mqp2p/src/signaling.rs b/crates/mqp2p/src/signaling.rs index ff061a8..2ffdbd3 100644 --- a/crates/mqp2p/src/signaling.rs +++ b/crates/mqp2p/src/signaling.rs @@ -3,7 +3,9 @@ use mqtt5::client::MqttClient; use mqtt5::types::{PublishOptions, PublishProperties}; use serde::{Deserialize, Serialize}; use serde_json::Value; +use std::collections::HashMap; use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tracing::{debug, error, trace}; @@ -13,10 +15,65 @@ pub const PEER_LEASE_SECS: u64 = 15; pub const MIN_LEASE_SECS: u64 = 2; +pub const CLIENT_ID_PREFIX: &str = "mqp2p-"; + +const PRESENCE_TOPIC: &str = "$DB/_presence/#"; + +const MAX_TRACKED_CLIENTS: usize = 4096; + pub fn heartbeat_interval_secs(lease_secs: u64) -> u64 { (lease_secs / 3).max(1) } +pub fn client_id_for(name: &str) -> String { + format!("{CLIENT_ID_PREFIX}{name}") +} + +#[derive(Debug, Deserialize)] +struct PresenceEvent { + client_id: String, + event: String, + ts: u64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct PresenceMark { + ts: u64, + online: bool, +} + +type PresenceTable = HashMap; + +fn apply_presence(table: &mut PresenceTable, event: PresenceEvent) { + if !event.client_id.starts_with(CLIENT_ID_PREFIX) { + return; + } + let online = match event.event.as_str() { + "connect" => true, + "disconnect" => false, + _ => return, + }; + match table.get(&event.client_id) { + Some(mark) if mark.ts > event.ts => {} + None if table.len() >= MAX_TRACKED_CLIENTS => {} + _ => { + table.insert( + event.client_id, + PresenceMark { + ts: event.ts, + online, + }, + ); + } + } +} + +fn departed(table: &PresenceTable, name: &str) -> bool { + table + .get(&client_id_for(name)) + .is_some_and(|mark| !mark.online) +} + fn unix_now_secs() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -73,11 +130,15 @@ pub enum CandidateKind { #[derive(Clone)] pub struct SignalingClient { client: MqttClient, + presence: Arc>, } impl SignalingClient { pub fn new(client: MqttClient) -> Self { - Self { client } + Self { + client, + presence: Arc::new(Mutex::new(PresenceTable::new())), + } } pub fn mqtt_client(&self) -> &MqttClient { @@ -93,6 +154,31 @@ impl SignalingClient { Ok(()) } + pub async fn disconnect(&self) -> Result<()> { + self.client + .disconnect() + .await + .map_err(|e| Error::Signaling(format!("MQTT disconnect failed: {e}")))?; + debug!("signaling client disconnected"); + Ok(()) + } + + pub async fn track_presence(&self) -> Result<()> { + let presence = Arc::clone(&self.presence); + self.client + .subscribe(PRESENCE_TOPIC, move |msg| { + if let Ok(event) = serde_json::from_slice::(&msg.payload) + && let Ok(mut table) = presence.lock() + { + apply_presence(&mut table, event); + } + }) + .await + .map_err(|e| Error::Signaling(format!("subscribe presence failed: {e}")))?; + debug!("tracking broker presence"); + Ok(()) + } + pub async fn register_peer( &self, id: &str, @@ -196,10 +282,14 @@ impl SignalingClient { }; let now = unix_now_secs(); + let presence = self + .presence + .lock() + .map_err(|_| Error::Signaling("presence table poisoned".into()))?; let peers: Vec = items .into_iter() .filter_map(|v| serde_json::from_value::(v).ok()) - .filter(|p| peer_is_live(p.expires_at, now)) + .filter(|p| peer_is_live(p.expires_at, now) && !departed(&presence, &p.name)) .collect(); debug!(count = peers.len(), "peers discovered"); @@ -377,6 +467,55 @@ fn check_response(response: &Value) -> Result<()> { mod tests { use super::*; + fn event(client_id: &str, event: &str, ts: u64) -> PresenceEvent { + PresenceEvent { + client_id: client_id.into(), + event: event.into(), + ts, + } + } + + #[test] + fn disconnect_marks_a_peer_departed_until_it_reconnects() { + let mut table = PresenceTable::new(); + apply_presence(&mut table, event("mqp2p-edge", "connect", 1)); + assert!(!departed(&table, "edge")); + apply_presence(&mut table, event("mqp2p-edge", "disconnect", 2)); + assert!(departed(&table, "edge")); + apply_presence(&mut table, event("mqp2p-edge", "connect", 3)); + assert!(!departed(&table, "edge")); + } + + #[test] + fn an_older_presence_event_does_not_override_a_newer_one() { + let mut table = PresenceTable::new(); + apply_presence(&mut table, event("mqp2p-edge", "connect", 5)); + apply_presence(&mut table, event("mqp2p-edge", "disconnect", 4)); + assert!(!departed(&table, "edge")); + } + + #[test] + fn unknown_peers_and_foreign_clients_are_not_departed() { + let mut table = PresenceTable::new(); + apply_presence(&mut table, event("stitch-app", "disconnect", 1)); + apply_presence(&mut table, event("mqp2p-edge", "expired", 1)); + assert!(table.is_empty()); + assert!(!departed(&table, "edge")); + } + + #[test] + fn presence_table_stops_growing_at_its_limit() { + let mut table = PresenceTable::new(); + for i in 0..MAX_TRACKED_CLIENTS { + apply_presence(&mut table, event(&format!("mqp2p-p{i}"), "connect", 1)); + } + apply_presence(&mut table, event("mqp2p-late", "disconnect", 2)); + assert_eq!(table.len(), MAX_TRACKED_CLIENTS); + assert!(!departed(&table, "late")); + apply_presence(&mut table, event("mqp2p-p0", "disconnect", 2)); + assert!(departed(&table, "p0")); + } + #[test] fn leaseless_peer_is_live() { assert!(peer_is_live(None, 1000)); diff --git a/crates/mqp2p/tests/integration.rs b/crates/mqp2p/tests/integration.rs index 716d4d1..7f338f8 100644 --- a/crates/mqp2p/tests/integration.rs +++ b/crates/mqp2p/tests/integration.rs @@ -1,4 +1,6 @@ use mqp2p::{Peer, PeerConfig, TransferProgress}; +use mqtt5::ConnectOptions; +use mqtt5::client::MqttClient; use rand::RngExt; use ring::digest; use std::io::Write; @@ -23,7 +25,12 @@ fn pick_free_port() -> u16 { .port() } -fn start_broker(port: u16, db_dir: &std::path::Path, passwd_file: &std::path::Path) -> BrokerGuard { +fn start_broker( + port: u16, + db_dir: &std::path::Path, + passwd_file: &std::path::Path, + extra_args: &[&str], +) -> BrokerGuard { let child = Command::new("mqdb") .args([ "agent", @@ -35,6 +42,7 @@ fn start_broker(port: u16, db_dir: &std::path::Path, passwd_file: &std::path::Pa "--passwd", &passwd_file.to_string_lossy(), ]) + .args(extra_args) .stdout(std::process::Stdio::null()) .stderr(std::process::Stdio::null()) .spawn() @@ -138,7 +146,7 @@ async fn test_full_file_transfer() { let passwd_file = create_passwd_file(tmpdir.path()); let broker_port = pick_free_port(); - let _broker = start_broker(broker_port, &db_dir, &passwd_file); + let _broker = start_broker(broker_port, &db_dir, &passwd_file, &[]); wait_for_port(broker_port); wait_for_broker_ready(broker_port); @@ -249,7 +257,7 @@ async fn presence_lease_expires_without_heartbeat() { let passwd_file = create_passwd_file(tmpdir.path()); let broker_port = pick_free_port(); - let _broker = start_broker(broker_port, &db_dir, &passwd_file); + let _broker = start_broker(broker_port, &db_dir, &passwd_file, &[]); wait_for_port(broker_port); wait_for_broker_ready(broker_port); @@ -339,7 +347,7 @@ async fn presence_recreates_after_record_reaped() { let passwd_file = create_passwd_file(tmpdir.path()); let broker_port = pick_free_port(); - let _broker = start_broker(broker_port, &db_dir, &passwd_file); + let _broker = start_broker(broker_port, &db_dir, &passwd_file, &[]); wait_for_port(broker_port); wait_for_broker_ready(broker_port); @@ -393,7 +401,7 @@ async fn reregister_preserves_public_addr() { let passwd_file = create_passwd_file(tmpdir.path()); let broker_port = pick_free_port(); - let _broker = start_broker(broker_port, &db_dir, &passwd_file); + let _broker = start_broker(broker_port, &db_dir, &passwd_file, &[]); wait_for_port(broker_port); wait_for_broker_ready(broker_port); @@ -440,3 +448,184 @@ async fn reregister_preserves_public_addr() { "public_addr must survive re-registration" ); } + +struct PresenceBroker { + _broker: BrokerGuard, + _tmpdir: tempfile::TempDir, + addr: String, +} + +async fn start_presence_broker() -> PresenceBroker { + let tmpdir = tempfile::tempdir().expect("failed to create tmpdir"); + let db_dir = tmpdir.path().join("db"); + std::fs::create_dir_all(&db_dir).expect("failed to create db dir"); + let passwd_file = create_passwd_file(tmpdir.path()); + let port = pick_free_port(); + let broker = start_broker(port, &db_dir, &passwd_file, &["--presence"]); + wait_for_port(port); + wait_for_broker_ready(port); + PresenceBroker { + _broker: broker, + _tmpdir: tmpdir, + addr: format!("127.0.0.1:{port}"), + } +} + +fn peer_config(name: &str, broker_addr: &str) -> PeerConfig { + PeerConfig::new(name, broker_addr) + .with_bind_addr("127.0.0.1:0".parse().unwrap()) + .without_stun() + .with_credentials("testuser", "testpass") +} + +async fn watch_presence( + broker_addr: &str, + client_id: &str, +) -> (MqttClient, flume::Receiver) { + let watcher_id = format!("presence-watcher-{}", uuid::Uuid::new_v4()); + let watcher = MqttClient::with_options( + ConnectOptions::new(watcher_id).with_credentials("testuser", "testpass".as_bytes()), + ); + watcher + .connect(broker_addr) + .await + .expect("watcher connect failed"); + let (tx, rx) = flume::unbounded(); + watcher + .subscribe(&format!("$DB/_presence/{client_id}"), move |msg| { + if let Ok(event) = serde_json::from_slice::(&msg.payload) { + let _ = tx.try_send(event); + } + }) + .await + .expect("watcher subscribe failed"); + (watcher, rx) +} + +async fn next_disconnect(events: &flume::Receiver) -> Option { + tokio::time::timeout(Duration::from_secs(3), async { + loop { + let event = events.recv_async().await.ok()?; + if event["event"] == "disconnect" { + return Some(event); + } + } + }) + .await + .ok() + .flatten() +} + +#[tokio::test] +async fn shutdown_sends_a_clean_mqtt_disconnect() { + let broker = start_presence_broker().await; + let (watcher, events) = watch_presence(&broker.addr, "mqp2p-leaver").await; + + let mut leaver = Peer::new(peer_config("leaver", &broker.addr)) + .await + .expect("peer creation failed"); + leaver.register().await.expect("registration failed"); + leaver.shutdown().await.expect("shutdown failed"); + + let event = next_disconnect(&events) + .await + .expect("the broker must see the peer's MQTT session end at shutdown"); + assert_eq!( + event["unexpected"], false, + "shutdown must send DISCONNECT, not drop the socket: {event}" + ); + watcher + .disconnect() + .await + .expect("watcher disconnect failed"); +} + +#[tokio::test] +async fn dropping_a_peer_sends_a_clean_mqtt_disconnect() { + let broker = start_presence_broker().await; + let (watcher, events) = watch_presence(&broker.addr, "mqp2p-dropped").await; + + let peer = Peer::new(peer_config("dropped", &broker.addr)) + .await + .expect("peer creation failed"); + drop(peer); + + let event = next_disconnect(&events) + .await + .expect("a dropped peer must not leave its MQTT client connected"); + assert_eq!(event["unexpected"], false, "{event}"); + watcher + .disconnect() + .await + .expect("watcher disconnect failed"); +} + +#[tokio::test] +async fn presence_hides_a_departed_peer_before_its_lease_expires() { + let broker = start_presence_broker().await; + + let mut observer = Peer::new( + peer_config("watcher", &broker.addr) + .with_lease_secs(60) + .with_presence(), + ) + .await + .expect("observer creation failed"); + observer + .register() + .await + .expect("observer registration failed"); + + let mut departing = Peer::new(peer_config("departing", &broker.addr).with_lease_secs(60)) + .await + .expect("departing peer creation failed"); + let departing_id = departing.register().await.expect("registration failed"); + assert!( + observer + .discover_peers() + .await + .expect("discovery failed") + .iter() + .any(|p| p.id == departing_id), + "a connected peer must be discoverable" + ); + + drop(departing); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while observer + .discover_peers() + .await + .expect("discovery failed") + .iter() + .any(|p| p.id == departing_id) + { + assert!( + tokio::time::Instant::now() < deadline, + "a peer whose MQTT client disconnected must leave discovery long before its 60 s lease" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + } + + let mut returning = Peer::new(peer_config("departing", &broker.addr).with_lease_secs(60)) + .await + .expect("returning peer creation failed"); + let returning_id = returning.register().await.expect("re-registration failed"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while !observer + .discover_peers() + .await + .expect("discovery failed") + .iter() + .any(|p| p.id == returning_id) + { + assert!( + tokio::time::Instant::now() < deadline, + "a peer that reconnects under the same name must be discoverable again" + ); + tokio::time::sleep(Duration::from_millis(200)).await; + } + + returning.shutdown().await.expect("shutdown failed"); + observer.shutdown().await.expect("observer shutdown failed"); +}