send mqtt disconnect on peer shutdown, derive keepalive from lease, add optional presence to mqp2p

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-<name>`) 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
 
 ```

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<String>,
     pub credentials: Option<(String, String)>,
     pub lease_secs: u64,
+    pub keepalive_secs: Option<u64>,
+    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);
+    }
+}

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<String, PresenceMark>;
+
+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<Mutex<PresenceTable>>,
 }
 
 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::<PresenceEvent>(&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<PeerInfo> = items
             .into_iter()
             .filter_map(|v| serde_json::from_value::<PeerInfo>(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));