[p2p] Support Primary/Secondary Peers (#3499)

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

broadcast/fuzz/fuzz_targets/broadcast_engine_operations.rs

@@ -13,6 +13,7 @@ use commonware_cryptography::{
 };
 use commonware_p2p::{simulated::Network, Recipients};
 use commonware_runtime::{deterministic, Buf, BufMut, Clock, Metrics, Quota, Runner};
+use commonware_utils::NZUsize;
 use libfuzzer_sys::fuzz_target;
 use rand::{seq::SliceRandom, SeedableRng};
 use std::{collections::BTreeMap, num::NonZeroU32, time::Duration};
@@ -172,26 +173,30 @@ fn resolve_recipients(pattern: &RecipientPattern, peers: &[PublicKey]) -> Recipi
 fn fuzz(input: FuzzInput) {
     let executor = deterministic::Runner::default();
     executor.start(|context| async move {
+        // Generate peer identities before building the network so the initial
+        // peer set can be seeded through the constructor.
+        let peers = input
+            .peer_seeds
+            .iter()
+            .map(|&seed| PrivateKey::from_seed(seed).public_key())
+            .collect::<Vec<_>>();
+
         // Create network
-        let (network, oracle) = Network::<deterministic::Context, PublicKey>::new(
+        let (network, oracle) = Network::<deterministic::Context, PublicKey>::new_with_peers(
             context.with_label("network"),
             commonware_p2p::simulated::Config {
                 max_size: 1024 * 1024,
                 disconnect_on_block: false,
-                tracked_peer_sets: None,
+                tracked_peer_sets: NZUsize!(1),
             },
-        );
+            peers.clone(),
+        )
+        .await;
         network.start();
 
         // Create peers
-        let mut peers = Vec::new();
         let mut mailboxes: BTreeMap<PublicKey, Mailbox<PublicKey, FuzzMessage>> = BTreeMap::new();
-        for (i, &seed) in input.peer_seeds.iter().enumerate() {
-            // Create peer
-            let crypto = PrivateKey::from_seed(seed);
-            let public_key = crypto.public_key();
-            peers.push(public_key.clone());
-
+        for (i, public_key) in peers.iter().cloned().enumerate() {
             // Create channel
             let (sender, receiver) = oracle
                 .control(public_key.clone())

broadcast/src/buffered/config.rs

@@ -18,7 +18,6 @@ pub struct Config<P: PublicKey, MCfg, D: Provider<PublicKey = P>> {
     /// The configuration for the codec item.
     pub codec_config: MCfg,
 
-    /// Provider for peer set changes. Per-peer caches are freed when a
-    /// peer leaves all tracked peer sets.
+    /// Provider for peer set changes (eviction follows latest primary; see [`buffered`](super)).
     pub peer_provider: D,
 }

broadcast/src/buffered/engine.rs

@@ -25,6 +25,13 @@ struct Waiter<M> {
     responder: oneshot::Sender<M>,
 }
 
+/// Result of buffering an incoming or locally sent digest (inserted, duplicate, or ineligible).
+enum InsertMessageResult {
+    Inserted,
+    Duplicate,
+    Ineligible,
+}
+
 /// Instance of the main engine for the module.
 ///
 /// It is responsible for:
@@ -90,6 +97,9 @@ where
     /// the message once.
     counts: BTreeMap<M::Digest, usize>,
 
+    /// Latest primary peer set allowed to keep buffered messages resident.
+    latest_primary_peers: Set<P>,
+
     ////////////////////////////////////////
     // Metrics
     ////////////////////////////////////////
@@ -124,6 +134,7 @@ where
             deques: BTreeMap::new(),
             items: BTreeMap::new(),
             counts: BTreeMap::new(),
+            latest_primary_peers: Set::default(),
             peer_provider: cfg.peer_provider,
             metrics,
         };
@@ -159,6 +170,14 @@ where
             on_stopped => {
                 debug!("shutdown");
             },
+            // Handle peer set subscription messages
+            Some(update) = peer_set_subscription.recv() else {
+                debug!("peer set subscription closed");
+                break;
+            } => {
+                // Evict by latest primary only; see buffered module docs.
+                self.update_latest_primary_peers(update.latest.primary);
+            },
             // Handle mailbox messages
             Some(msg) = self.mailbox_receiver.recv() else {
                 error!("mailbox receiver failed");
@@ -210,12 +229,6 @@ where
                     .inc();
                 self.handle_network(peer, msg);
             },
-            Some((_, _, tracked_peers)) = peer_set_subscription.recv() else {
-                debug!("peer set subscription closed");
-                break;
-            } => {
-                self.evict_untracked_peers(&tracked_peers);
-            },
         }
     }
 
@@ -232,7 +245,8 @@ where
         responder: oneshot::Sender<Vec<P>>,
     ) {
         // Store the message, continue even if it was already stored
-        let _ = self.insert_message(self.public_key.clone(), msg.clone());
+        let digest = msg.digest();
+        let _ = self.insert_message(self.public_key.clone(), digest, msg.clone());
 
         // Broadcast the message to the network
         let sent_to = sender
@@ -271,13 +285,20 @@ where
 
     /// Handles a message that was received from a peer.
     fn handle_network(&mut self, peer: P, msg: M) {
-        if !self.insert_message(peer.clone(), msg) {
-            debug!(?peer, "message already stored");
-            self.metrics.receive.inc(Status::Dropped);
-            return;
+        let digest = msg.digest();
+        match self.insert_message(peer.clone(), digest, msg) {
+            InsertMessageResult::Inserted => {
+                self.metrics.receive.inc(Status::Success);
+            }
+            InsertMessageResult::Duplicate => {
+                debug!(?peer, "message already stored");
+                self.metrics.receive.inc(Status::Dropped);
+            }
+            InsertMessageResult::Ineligible => {
+                debug!(?peer, "message from peer outside latest.primary not cached");
+                self.metrics.receive.inc(Status::Dropped);
+            }
         }
-
-        self.metrics.receive.inc(Status::Success);
     }
 
     ////////////////////////////////////////
@@ -286,18 +307,21 @@ where
 
     /// Inserts a message into the cache.
     ///
-    /// Returns `true` if the message was inserted, `false` if it was already present.
-    /// Updates the deque, item count, and message cache, potentially evicting an old message.
-    fn insert_message(&mut self, peer: P, msg: M) -> bool {
-        let digest = msg.digest();
-
+    /// Waiters are notified even when a sender is not eligible to keep a
+    /// buffered cache entry resident.
+    fn insert_message(&mut self, peer: P, digest: M::Digest, msg: M) -> InsertMessageResult {
         // Send the message to the waiters, if any
         if let Some(waiters) = self.waiters.remove(&digest) {
             for waiter in waiters {
                 self.respond_subscribe(waiter.responder, msg.clone());
             }
         }
 
+        // Only peers listed in `latest.primary` may buffer
+        if self.latest_primary_peers.position(&peer).is_none() {
+            return InsertMessageResult::Ineligible;
+        }
+
         // Get the relevant deque for the peer
         let deque = self
             .deques
@@ -310,7 +334,7 @@ where
                 let v = deque.remove(i).unwrap(); // Must exist
                 deque.push_front(v);
             }
-            return false;
+            return InsertMessageResult::Duplicate;
         };
 
         // - Insert the digest into the peer cache
@@ -336,20 +360,20 @@ where
             decrement_digest_refcount(&mut self.counts, &mut self.items, &stale);
         }
 
-        true
+        InsertMessageResult::Inserted
     }
 
-    fn evict_untracked_peers(&mut self, tracked_peers: &Set<P>) {
-        let tracked = tracked_peers.as_ref();
+    fn update_latest_primary_peers(&mut self, peers: Set<P>) {
         for (peer, deque) in self
             .deques
-            .extract_if(.., |peer, _| !tracked.contains(peer))
+            .extract_if(.., |peer, _| peers.position(peer).is_none())
         {
             debug!(?peer, digests = deque.len(), "evicting disconnected peer");
             for digest in deque {
                 decrement_digest_refcount(&mut self.counts, &mut self.items, &digest);
             }
         }
+        self.latest_primary_peers = peers;
     }
 
     ////////////////////////////////////////