feat(iroh-streamplace): peer list hydration, leaving message type

rust/iroh-streamplace/src/api.rs

···
 //! Protocol API
 
 use std::{
-    collections::{BTreeMap, BTreeSet},
+    collections::{BTreeMap, BTreeSet, HashMap},
     time::Duration,
 };
 
···
     data: Bytes,
 }
 
+/// Ask a remote peer to return a stream of it's known PeerInfos
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct RequestPeerInfos {
+    remote_id: NodeId,
+}
+
+/// Return a stream of peer infos
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct HandlePeerInfosRequest {}
+
+/// Return this peer's local state
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct MyPeerInfo {}
+
 /// List all peers, and the subscriptions that they're believed to have
 /// "believed", because subscription info can be out of date
 #[derive(Debug, Clone, Serialize, Deserialize)]
-struct Peers {}
+struct MyPeers {}
 
 /// Prune peers that haven't been seen since the given timestamp
 #[derive(Debug, Clone, Serialize, Deserialize)]
-struct PrunePeers {
+struct PruneMyPeers {
     cutoff_timestamp: u64,
 }
 
···
     info: PeerInfo,
 }
 
-/// list out our local subscriptions
-#[derive(Debug, Clone, Serialize, Deserialize)]
-struct MySubscriptions {}
-
 /// Request a node list out it's current subscriptions
 #[derive(Debug, Clone, Serialize, Deserialize)]
 struct GetSubscriptions {
···
     node_id: NodeId,
 }
 
+/// Tell all remote peers that we're leaving the network
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct BroadcastLeaving {}
+
+/// Tell a remote node that we're leaving the network
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct HandleLeaving {
+    /// the node id of the peer leaving
+    node_id: NodeId,
+}
+
 /// details about a peer in the network
 #[derive(Debug, Clone, Ord, PartialOrd, PartialEq, Eq, Serialize, Deserialize)]
 pub(crate) struct PeerInfo {
···
 enum Protocol {
     // swarm coordination
     #[rpc(tx=mpsc::Sender<PeerInfo>)]
-    Peers(Peers),
+    MyPeers(MyPeers),
     #[rpc(tx=oneshot::Sender<PeerInfo>)]
-    MyPeerInfo(MySubscriptions),
+    MyPeerInfo(MyPeerInfo),
+    #[rpc(tx=oneshot::Sender<()>)]
+    PruneMyPeers(PruneMyPeers),
+    #[rpc(tx=oneshot::Sender<()>)]
+    BroadcastLeaving(BroadcastLeaving),
+    #[rpc(tx=oneshot::Sender<()>)]
+    HandleLeaving(HandleLeaving),
+
     #[rpc(tx=oneshot::Sender<()>)]
-    PrunePeers(PrunePeers),
+    RequestPeerInfos(RequestPeerInfos),
+    #[rpc(tx=mpsc::Sender<PeerInfo>)]
+    HandlePeerInfosRequest(HandlePeerInfosRequest),
+
     #[rpc(tx=oneshot::Sender<()>)]
     SendPeerInfo(SendPeerInfo),
     #[rpc(tx=oneshot::Sender<()>)]
···
     /// peers we'll permanently broadcast to
     anchor_peers: Vec<NodeId>,
     /// set of all peers we believe to be life in the swarm
-    peers: BTreeSet<PeerInfo>,
+    peers: HashMap<NodeId, PeerInfo>,
     /// set of stream subscriptions we're receiving data for
     subscriptions: BTreeMap<String, BTreeSet<NodeId>>,
     /// pool of open RPC connections
···
             endpoint: endpoint.clone(),
             recv: rx,
             anchor_peers,
-            peers: BTreeSet::new(),
+            peers: HashMap::new(),
             subscriptions: BTreeMap::new(),
             connections: BTreeMap::new(),
             handler: Box::new(handler),
···
     async fn handle(&mut self, msg: Message) {
         match msg {
             // swarm coordination
-            Message::Peers(sub) => {
+            Message::MyPeers(sub) => {
                 debug!("peers {:?}", sub);
                 let WithChannels { tx, .. } = sub;
 
···
                 let mut sent = BTreeSet::new();
 
                 // stream over the list of peers we know about
-                for sub in &self.peers {
-                    sent.insert(&sub.node_id);
+                for (id, sub) in &self.peers {
+                    sent.insert(*id);
                     if tx.send(sub.clone()).await.is_err() {
                         break;
                     }
                 }
 
-                // send over any anchor peers we know about, but haven't already sent
-                // these go with empty subscription sets, which isn't great.
+                // send over any anchor peers we know about, but haven't already
+                // sent from our peers list. these go with empty subscription
+                // sets, which isn't great.
                 for anchor in &self.anchor_peers {
                     if sent.contains(anchor) {
                         continue;
···
                 let WithChannels { tx, .. } = sub;
                 tx.send(self.my_peer_info()).await.ok();
             }
+            Message::PruneMyPeers(sub) => {
+                let WithChannels { tx, inner, .. } = sub;
+                // prune peers that haven't been seen since the given timestamp
+                self.peers
+                    .retain(|_, peer| peer.timestamp >= inner.cutoff_timestamp);
+                tx.send(()).await.ok();
+            }
+
+            Message::RequestPeerInfos(list) => {
+                let WithChannels { inner, tx, .. } = list;
+                let conn = self.get_conn(&inner.remote_id).await;
+                let mut rx = conn
+                    .rpc
+                    .server_streaming(HandlePeerInfosRequest {}, 1000)
+                    .await
+                    .unwrap();
+                while let Some(mut peer_info) = rx.recv().await.unwrap() {
+                    // update our tracked state about this peer, using timestamps
+                    // to avoid confusion from external sources
+                    peer_info.timestamp = timestamp();
+                    self.peers.insert(peer_info.node_id, peer_info);
+                }
+
+                tx.send(()).await.ok();
+            }
+            Message::HandlePeerInfosRequest(list) => {
+                let WithChannels { tx, .. } = list;
+                for (_, peer) in self.peers.clone() {
+                    if let Err(e) = tx.send(peer).await {
+                        tracing::error!("send peer error: {:?}", e);
+                    }
+                }
+            }
+
             Message::SendPeerInfo(info) => {
                 let WithChannels { inner, tx, .. } = info;
                 debug!(
···
                 // update our tracked state about this peer, using timestamps
                 // to avoid confusion from external sources
                 inner.timestamp = timestamp();
-                self.peers.insert(inner);
+                self.peers.insert(inner.node_id, inner);
                 tx.send(()).await.ok();
             }
-            Message::PrunePeers(sub) => {
-                let WithChannels { tx, inner, .. } = sub;
-                // prune peers that haven't been seen since the given timestamp
-                self.peers
-                    .retain(|peer| peer.timestamp >= inner.cutoff_timestamp);
+            Message::BroadcastLeaving(leaving) => {
+                let WithChannels { tx, .. } = leaving;
+                let node_id = self.endpoint.node_id();
+                let remotes = self
+                    .peers
+                    .values()
+                    .map(|peer| peer.node_id)
+                    .collect::<Vec<_>>();
+                for remote_node_id in remotes {
+                    // ensure connection
+                    let conn = self.get_conn(&remote_node_id).await;
+                    if let Err(err) = conn.rpc.rpc(HandleLeaving { node_id }).await {
+                        tracing::error!("failed to handle leaving: {}", err);
+                    }
+                }
+                tx.send(()).await.ok();
+            }
+            Message::HandleLeaving(leaving) => {
+                let WithChannels { tx, inner, .. } = leaving;
+                self.peers.remove(&inner.node_id);
                 tx.send(()).await.ok();
             }
 
···
         peer_info_broadcast_interval: Duration,
         peer_prune_interval: Duration,
     ) -> Self {
+        let anchors = anchor_peers.clone();
         let api = Actor::spawn(endpoint, anchor_peers, handler);
 
+        // hydrate our peers list from anchor nodes
+        let api2 = api.clone();
+        n0_future::task::spawn(async move {
+            for anchor in anchors {
+                if let Err(e) = api2.inner.rpc(RequestPeerInfos { remote_id: anchor }).await {
+                    tracing::error!("requesting peer infos: {:?}", e);
+                }
+            }
+        });
+
         // re-broadcast our subscriptions every interval
         if peer_info_broadcast_interval > Duration::from_millis(0) {
             let api2 = api.clone();
···
                 loop {
                     tokio::time::sleep(peer_prune_interval).await;
                     let cutoff_timestamp = timestamp() - peer_prune_interval.as_secs();
-                    if let Err(e) = api2.inner.rpc(PrunePeers { cutoff_timestamp }).await {
+                    if let Err(e) = api2.inner.rpc(PruneMyPeers { cutoff_timestamp }).await {
                         tracing::error!("pruning stale subscriptions: {:?}", e);
                     }
                 }
···
 
     /// List all peers we know about, and the subscriptions they have
     pub(crate) async fn peers(&self) -> irpc::Result<Vec<PeerInfo>> {
-        let mut rx = self.inner.server_streaming(Peers {}, 1000).await?;
+        let mut rx = self.inner.server_streaming(MyPeers {}, 1000).await?;
         let mut peers = Vec::new();
         while let Some(peer) = rx.recv().await? {
             peers.push(peer);
···
         Ok(peers)
     }
 
-    pub(crate) async fn my_subscriptions(&self) -> irpc::Result<PeerInfo> {
-        self.inner.rpc(MySubscriptions {}).await
+    pub(crate) async fn my_peer_info(&self) -> irpc::Result<PeerInfo> {
+        self.inner.rpc(MyPeerInfo {}).await
+    }
+
+    async fn broadcast_peer_info(&self) -> irpc::Result<JoinHandle<()>> {
+        let peers = self.peers().await?;
+        let subs = self.my_peer_info().await?;
+        let client = self.inner.clone();
+        let handle = n0_future::task::spawn(async move {
+            if let Err(e) = broadcast_peer_info_inner(client, peers, subs).await {
+                tracing::error!("Peer announcement task failed: {:?}", e);
+            }
+        });
+        Ok(handle)
+    }
+
+    pub(crate) async fn leaving(&self) -> irpc::Result<()> {
+        self.inner.rpc(BroadcastLeaving {}).await
     }
 
     pub(crate) async fn subscribe(&self, key: String, self_id: NodeId) -> irpc::Result<()> {
···
         self.broadcast_peer_info().await?;
 
         Ok(())
-    }
-
-    async fn broadcast_peer_info(&self) -> irpc::Result<JoinHandle<()>> {
-        let peers = self.peers().await?;
-        let subs = self.my_subscriptions().await?;
-        let client = self.inner.clone();
-        let handle = n0_future::task::spawn(async move {
-            if let Err(e) = broadcast_peer_info_inner(client, peers, subs).await {
-                tracing::error!("Peer announcement task failed: {:?}", e);
-            }
-        });
-        Ok(handle)
     }
 
     pub(crate) async fn unsubscribe(&self, key: String, self_id: NodeId) -> irpc::Result<()> {

rust/iroh-streamplace/src/receiver.rs

···
         Ok(())
     }
 
+    /// Get our node address
     #[uniffi::method(async_runtime = "tokio")]
     pub async fn node_addr(&self) -> NodeAddr {
         self.endpoint.node_addr().await
+    }
+
+    /// tell the network that we're leaving. This should only be called just before disconnecting.
+    #[uniffi::method(async_runtime = "tokio")]
+    pub async fn leaving(&self) -> Result<(), Error> {
+        self.api.leaving().await?;
+        Ok(())
     }
 }
 

rust/iroh-streamplace/swarm.md

···
 * when a node unsubscribes from a feed, it broadcasts it's updated `PeerInfo` to all known peers
 * every `DEFAULT_PEER_INFO_REPUBLISH_INTERVAL`, a node broadcasts it's current `PeerInfo` to all known peers
 
-every `DEFAULT_PEER_PRUNE_INTERVAL`, nodes will examine their local list of peers, and prune any who's latest timestamp is older than the current time, minus the prune interval, this is to purge peers that die off without notice
+every `DEFAULT_PEER_PRUNE_INTERVAL`, nodes will examine their local list of peers, and prune any who's latest timestamp is older than the current time, minus the prune interval, this is to purge peers that die off without notice.
 
 ### Anchor Peers
-Anchor peers are _always_ transmitted to. They're expcted to be high-availability nodes. Any broadcast message will always try to
+Anchor peers are _always_ transmitted to. They're expected to be high-availability nodes. Any broadcast message will always broadcast to anchor peers, regardless of whether they are online at the time, or not.
+
+### Peer Listing Messages
+At startup, the new nodes will send a `RequestPeerInfos` request to all anchor nodes. Each anchor node will respond with their list of `PeerInfo`s to inform new nodes of their current view of the swarm. There's room to grow on maintaining swarm health, but this message type is a good primitive as a start.
 
 ### FFI API
-The FFI API to goland is a single method on the `Receiver`: `peers`. It returns an array of `PeerInfo`, representing the nodes current view of the swarm.
+The FFI API to goland is 2 methods on the `Receiver`: `peers`, and `leaving`. It returns an array of `PeerInfo`, representing the nodes current view of the swarm. `leaving` should be called just before a node shuts down to notify the network that the node is going away. It's not a critical that `leaving` is called, but will cut down on stale data living in the network.