[runtime/p2p] Use Buf / BufMut in networking APIs

codec/src/codec.rs

@@ -1,7 +1,7 @@
 //! Core traits for encoding and decoding.
 
 use crate::error::Error;
-use bytes::{Buf, BufMut, BytesMut};
+use bytes::{Buf, BufMut, Bytes, BytesMut};
 
 /// Trait for types with a known, fixed encoded size.
 ///
@@ -63,6 +63,20 @@ pub trait Read: Sized {
 /// This trait provides the convenience [Encode::encode] method which handles
 /// buffer allocation, writing, and size assertion in one go.
 pub trait Encode: Write + EncodeSize {
+    /// Encodes `self` into a new [Bytes] buffer.
+    ///
+    /// This method calculates the required size using [EncodeSize::encode_size], allocates a
+    /// buffer of that exact capacity, writes the value using [Write::write], and performs a
+    /// sanity check assertion.
+    ///
+    /// # Panics
+    ///
+    /// Panics if `encode_size()` does not return the same number of bytes actually written by
+    /// `write()`
+    fn encode(&self) -> Bytes {
+        self.encode_mut().freeze()
+    }
+
     /// Encodes `self` into a new [BytesMut] buffer.
     ///
     /// This method calculates the required size using [EncodeSize::encode_size], allocates a
@@ -73,7 +87,7 @@ pub trait Encode: Write + EncodeSize {
     ///
     /// Panics if `encode_size()` does not return the same number of bytes actually written by
     /// `write()`
-    fn encode(&self) -> BytesMut {
+    fn encode_mut(&self) -> BytesMut {
         let len = self.encode_size();
         let mut buffer = BytesMut::with_capacity(len);
         self.write(&mut buffer);

codec/src/types/hash_map.rs

@@ -302,7 +302,7 @@ mod tests {
         let mut map = HashMap::new();
         map.insert(1u32, 100u64);
 
-        let mut encoded = map.encode();
+        let mut encoded = map.encode_mut();
         encoded.put_u8(0xFF); // Add extra byte
 
         // Use decode_cfg which enforces buffer is fully consumed

codec/src/types/hash_set.rs

@@ -253,7 +253,7 @@ mod tests {
         let mut set = HashSet::new();
         set.insert(1u32);
 
-        let mut encoded = set.encode();
+        let mut encoded = set.encode_mut();
         encoded.put_u8(0xFF); // Add extra byte
 
         // Use decode_cfg which enforces buffer is fully consumed

collector/fuzz/fuzz_targets/collector.rs

@@ -231,7 +231,7 @@ impl CheckedSender for MockCheckedSender {
 
     async fn send(
         self,
-        _message: bytes::Bytes,
+        _message: impl bytes::Buf + Send,
         _priority: bool,
     ) -> Result<Vec<Self::PublicKey>, Self::Error> {
         Ok(vec![])

collector/src/p2p/mocks/sender.rs

@@ -1,6 +1,6 @@
 //! Mock sender implementations for testing.
 
-use bytes::Bytes;
+use bytes::Buf;
 use commonware_cryptography::PublicKey;
 use commonware_p2p::{CheckedSender, LimitedSender, Recipients};
 use std::time::SystemTime;
@@ -47,7 +47,7 @@ impl<P: PublicKey> CheckedSender for CheckedFailing<P> {
     type PublicKey = P;
     type Error = Error;
 
-    async fn send(self, _message: Bytes, _priority: bool) -> Result<Vec<P>, Self::Error> {
+    async fn send(self, _message: impl Buf + Send, _priority: bool) -> Result<Vec<P>, Self::Error> {
         Err(Error::Failed)
     }
 }

collector/src/p2p/mod.rs

@@ -828,7 +828,7 @@ mod tests {
                 .0
                 .send(
                     Recipients::One(peers[0].clone()),
-                    response_to_peer1.encode().into(),
+                    response_to_peer1.encode(),
                     true,
                 )
                 .await

consensus/fuzz/src/disrupter.rs

@@ -216,36 +216,36 @@ where
             Vote::Notarize(notarize) => {
                 if self.fuzz_input.random_bool() {
                     let mutated = self.mutate_bytes(&msg);
-                    let _ = sender.send(Recipients::All, mutated.into(), true).await;
+                    let _ = sender.send(Recipients::All, &mutated[..], true).await;
                 } else {
                     let proposal = self.mutate_proposal(&notarize.proposal);
                     if let Some(v) = Notarize::sign(&self.scheme, proposal) {
-                        let msg = Vote::<S, Sha256Digest>::Notarize(v).encode().into();
+                        let msg = Vote::<S, Sha256Digest>::Notarize(v).encode();
                         let _ = sender.send(Recipients::All, msg, true).await;
                     }
                 }
             }
             Vote::Finalize(finalize) => {
                 if self.fuzz_input.random_bool() {
                     let mutated = self.mutate_bytes(&msg);
-                    let _ = sender.send(Recipients::All, mutated.into(), true).await;
+                    let _ = sender.send(Recipients::All, &mutated[..], true).await;
                 } else {
                     let proposal = self.mutate_proposal(&finalize.proposal);
                     if let Some(v) = Finalize::sign(&self.scheme, proposal) {
-                        let msg = Vote::<S, Sha256Digest>::Finalize(v).encode().into();
+                        let msg = Vote::<S, Sha256Digest>::Finalize(v).encode();
                         let _ = sender.send(Recipients::All, msg, true).await;
                     }
                 }
             }
             Vote::Nullify(_) => {
                 if self.fuzz_input.random_bool() {
                     let mutated = self.mutate_bytes(&msg);
-                    let _ = sender.send(Recipients::All, mutated.into(), true).await;
+                    let _ = sender.send(Recipients::All, &mutated[..], true).await;
                 } else {
                     let v = self.random_view(self.last_vote);
                     let round = Round::new(Epoch::new(EPOCH), View::new(v));
                     if let Some(v) = Nullify::<S>::sign::<Sha256Digest>(&self.scheme, round) {
-                        let msg = Vote::<S, Sha256Digest>::Nullify(v).encode().into();
+                        let msg = Vote::<S, Sha256Digest>::Nullify(v).encode();
                         let _ = sender.send(Recipients::All, msg, true).await;
                     }
                 }
@@ -280,7 +280,7 @@ where
         // Optionally send mutated certificate
         if self.fuzz_input.random_bool() {
             let mutated = self.mutate_bytes(&msg);
-            let _ = sender.send(Recipients::All, mutated.into(), true).await;
+            let _ = sender.send(Recipients::All, &mutated[..], true).await;
         }
     }
 
@@ -327,20 +327,20 @@ where
 
         if self.participants.index(&self.validator).is_none() {
             let bytes = self.bytes();
-            let _ = sender.send(Recipients::All, bytes.into(), true).await;
+            let _ = sender.send(Recipients::All, &bytes[..], true).await;
             return;
         }
 
         match self.message() {
             Message::Notarize => {
                 if let Some(vote) = Notarize::sign(&self.scheme, proposal) {
-                    let msg = Vote::<S, Sha256Digest>::Notarize(vote).encode().into();
+                    let msg = Vote::<S, Sha256Digest>::Notarize(vote).encode();
                     let _ = sender.send(Recipients::All, msg, true).await;
                 }
             }
             Message::Finalize => {
                 if let Some(vote) = Finalize::sign(&self.scheme, proposal) {
-                    let msg = Vote::<S, Sha256Digest>::Finalize(vote).encode().into();
+                    let msg = Vote::<S, Sha256Digest>::Finalize(vote).encode();
                     let _ = sender.send(Recipients::All, msg, true).await;
                 }
             }
@@ -350,13 +350,13 @@ where
                     View::new(self.random_view(self.last_vote)),
                 );
                 if let Some(vote) = Nullify::<S>::sign::<Sha256Digest>(&self.scheme, round) {
-                    let msg = Vote::<S, Sha256Digest>::Nullify(vote).encode().into();
+                    let msg = Vote::<S, Sha256Digest>::Nullify(vote).encode();
                     let _ = sender.send(Recipients::All, msg, true).await;
                 }
             }
             Message::Random => {
                 let bytes = self.bytes();
-                let _ = sender.send(Recipients::All, bytes.into(), true).await;
+                let _ = sender.send(Recipients::All, &bytes[..], true).await;
             }
         }
     }

consensus/src/aggregation/types.rs

@@ -148,7 +148,7 @@ impl<D: Digest> Subject for &Item<D> {
     }
 
     fn message(&self) -> Bytes {
-        self.encode().freeze()
+        self.encode()
     }
 }
 

consensus/src/marshal/actor.rs

@@ -553,7 +553,7 @@ where
                                         debug!(?commitment, "block missing on request");
                                         continue;
                                     };
-                                    let _ = response.send(block.encode().into());
+                                    let _ = response.send(block.encode());
                                 }
                                 Request::Finalized { height } => {
                                     // Get finalization
@@ -569,7 +569,7 @@ where
                                     };
 
                                     // Send finalization
-                                    let _ = response.send((finalization, block).encode().into());
+                                    let _ = response.send((finalization, block).encode());
                                 }
                                 Request::Notarized { round } => {
                                     // Get notarization
@@ -584,7 +584,7 @@ where
                                         debug!(?commitment, "block missing on request");
                                         continue;
                                     };
-                                    let _ = response.send((notarization, block).encode().into());
+                                    let _ = response.send((notarization, block).encode());
                                 }
                             }
                         },

consensus/src/ordered_broadcast/engine.rs

@@ -838,7 +838,7 @@ impl<
         node_sender
             .send(
                 Recipients::Some(validators.iter().cloned().collect()),
-                node.encode().into(),
+                node.encode(),
                 self.priority_proposals,
             )
             .await

consensus/src/simplex/actors/batcher/mod.rs

@@ -206,7 +206,7 @@ mod tests {
             injector_sender
                 .send(
                     Recipients::One(me.clone()),
-                    Certificate::Notarization(notarization.clone()).encode().into(),
+                    Certificate::Notarization(notarization.clone()).encode(),
                     true,
                 )
                 .await
@@ -224,8 +224,7 @@ mod tests {
                 .send(
                     Recipients::One(me.clone()),
                     Certificate::<S, Sha256Digest>::Nullification(nullification.clone())
-                        .encode()
-                        .into(),
+                        .encode(),
                     true,
                 )
                 .await
@@ -242,7 +241,7 @@ mod tests {
             injector_sender
                 .send(
                     Recipients::One(me.clone()),
-                    Certificate::Finalization(finalization.clone()).encode().into(),
+                    Certificate::Finalization(finalization.clone()).encode(),
                     true,
                 )
                 .await
@@ -372,7 +371,7 @@ mod tests {
                     sender
                         .send(
                             Recipients::One(me.clone()),
-                            Vote::Notarize(vote).encode().into(),
+                            Vote::Notarize(vote).encode(),
                             true,
                         )
                         .await
@@ -527,7 +526,7 @@ mod tests {
                     sender
                         .send(
                             Recipients::One(me.clone()),
-                            Vote::Notarize(vote).encode().into(),
+                            Vote::Notarize(vote).encode(),
                             true,
                         )
                         .await
@@ -550,7 +549,7 @@ mod tests {
             injector_sender
                 .send(
                     Recipients::One(me.clone()),
-                    Certificate::Notarization(notarization.clone()).encode().into(),
+                    Certificate::Notarization(notarization.clone()).encode(),
                     true,
                 )
                 .await
@@ -572,7 +571,7 @@ mod tests {
                 sender
                     .send(
                         Recipients::One(me.clone()),
-                        Vote::Notarize(last_vote).encode().into(),
+                        Vote::Notarize(last_vote).encode(),
                         true,
                     )
                     .await
@@ -706,7 +705,7 @@ mod tests {
                 sender
                     .send(
                         Recipients::One(me.clone()),
-                        Vote::Notarize(leader_vote).encode().into(),
+                        Vote::Notarize(leader_vote).encode(),
                         true,
                     )
                     .await
@@ -731,7 +730,7 @@ mod tests {
                     sender
                         .send(
                             Recipients::One(me.clone()),
-                            Vote::Notarize(vote).encode().into(),
+                            Vote::Notarize(vote).encode(),
                             true,
                         )
                         .await
@@ -770,7 +769,7 @@ mod tests {
                 sender
                     .send(
                         Recipients::One(me.clone()),
-                        Vote::Notarize(vote6).encode().into(),
+                        Vote::Notarize(vote6).encode(),
                         true,
                     )
                     .await
@@ -902,7 +901,7 @@ mod tests {
             leader_sender
                 .send(
                     Recipients::One(me.clone()),
-                    Vote::Notarize(leader_vote).encode().into(),
+                    Vote::Notarize(leader_vote).encode(),
                     true,
                 )
                 .await
@@ -1019,7 +1018,7 @@ mod tests {
             leader_sender
                 .send(
                     Recipients::One(me.clone()),
-                    Vote::Notarize(leader_vote).encode().into(),
+                    Vote::Notarize(leader_vote).encode(),
                     true,
                 )
                 .await
@@ -1162,7 +1161,7 @@ mod tests {
             leader_sender
                 .send(
                     Recipients::One(me.clone()),
-                    Vote::Notarize(leader_vote).encode().into(),
+                    Vote::Notarize(leader_vote).encode(),
                     true,
                 )
                 .await
@@ -1318,7 +1317,7 @@ mod tests {
                     sender
                         .send(
                             Recipients::One(me.clone()),
-                            Vote::Notarize(vote).encode().into(),
+                            Vote::Notarize(vote).encode(),
                             true,
                         )
                         .await
@@ -1364,7 +1363,7 @@ mod tests {
                     sender
                         .send(
                             Recipients::One(me.clone()),
-                            Vote::Notarize(vote).encode().into(),
+                            Vote::Notarize(vote).encode(),
                             true,
                         )
                         .await

consensus/src/simplex/actors/resolver/actor.rs

@@ -260,7 +260,7 @@ impl<
                     // the full timeout)
                     return;
                 };
-                let _ = response.send(certificate.encode().into());
+                let _ = response.send(certificate.encode());
             }
         }
     }