node: Infallible pdu encoding

node-data/src/encoding.rs

@@ -424,11 +424,9 @@ mod tests {
     /// Asserts if encoding/decoding of a serializable type runs properly.
     fn assert_serializable<S: Dummy<Faker> + Eq + Serializable>() {
         let obj: S = Faker.fake();
-        let mut buf = vec![];
-        obj.write(&mut buf).expect("should be writable");
+        let data = obj.write_to_vec();
 
-        assert!(obj
-            .eq(&S::read(&mut &buf.to_vec()[..]).expect("should be readable")));
+        assert!(obj.eq(&S::read(&mut &data[..]).expect("should be readable")));
     }
 
     #[test]

node-data/src/ledger/faults.rs

@@ -65,9 +65,8 @@ impl From<BlsSigError> for InvalidFault {
 impl Fault {
     /// Hash the serialized form
     pub fn hash(&self) -> [u8; 32] {
-        let mut b = vec![];
-        self.write(&mut b).expect("Write to a vec shall not fail");
-        BlsScalar::hash_to_scalar(&b[..]).to_bytes()
+        let data = self.write_to_vec();
+        BlsScalar::hash_to_scalar(&data).to_bytes()
     }
 
     pub fn same(&self, other: &Fault) -> bool {

node-data/src/lib.rs

@@ -32,6 +32,13 @@ pub trait Serializable {
     where
         Self: Sized;
 
+    fn write_to_vec(&self) -> Vec<u8> {
+        let mut buffer = vec![];
+        self.write(&mut buffer)
+            .expect("Writing to vec should succeed");
+        buffer
+    }
+
     fn read_bytes<R: Read, const N: usize>(r: &mut R) -> io::Result<[u8; N]> {
         let mut buffer = [0u8; N];
         r.read_exact(&mut buffer)?;

node-data/src/message.rs

@@ -332,9 +332,7 @@ impl ConsensusHeader {
     }
 
     pub fn signable(&self) -> Vec<u8> {
-        let mut buf = vec![];
-        self.write(&mut buf).expect("Writing to vec should succeed");
-        buf
+        self.write_to_vec()
     }
 }
 
@@ -1387,8 +1385,8 @@ mod tests {
     }
 
     fn assert_serialize<S: Serializable + PartialEq + core::fmt::Debug>(v: S) {
-        let mut buf = vec![];
-        assert!(v.write(&mut buf).is_ok());
+        let buf = v.write_to_vec();
+
         let dup = S::read(&mut &buf[..]).expect("deserialize is ok");
         assert_eq!(
             v,

node/src/chain.rs

@@ -210,9 +210,7 @@ impl<N: Network, DB: database::DB, VM: vm::VMExecution>
                         };
                     }
 
-                    if let Err(e) = network.read().await.broadcast(&msg).await {
-                        warn!("Unable to re-route message {e}");
-                    }
+                    network.read().await.broadcast(&msg).await;
                 },
                 // Handles accept_block_timeout event
                 _ = sleep_until(timeout) => {

node/src/chain/acceptor.rs

@@ -1009,9 +1009,7 @@ impl<DB: database::DB, VM: vm::VMExecution, N: Network> Acceptor<N, DB, VM> {
 }
 
 async fn broadcast<N: Network>(network: &Arc<RwLock<N>>, msg: &Message) {
-    let _ = network.read().await.broadcast(msg).await.map_err(|err| {
-        warn!("Unable to broadcast msg: {:?} {err} ", msg.topic())
-    });
+    network.read().await.broadcast(msg).await
 }
 
 /// Performs full verification of block header against prev_block header where

node/src/chain/consensus.rs

@@ -352,10 +352,7 @@ impl<DB: database::DB, VM: vm::VMExecution> Operations for Executor<DB, VM> {
                 metric.push_back(elapsed);
                 debug!(event = "avg_updated", ?step_name,  metric = ?metric);
 
-                let mut bytes = Vec::new();
-                metric.write(&mut bytes)?;
-
-                t.op_write(db_key, bytes)
+                t.op_write(db_key, metric.write_to_vec())
             })
             .map_err(Error::MetricsUpdate)?;
 

node/src/chain/fsm.rs

@@ -24,7 +24,7 @@ use std::time::Duration;
 use std::{sync::Arc, time::SystemTime};
 use tokio::sync::RwLock;
 use tokio::time::Instant;
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info};
 
 const MAX_BLOCKS_TO_REQUEST: i16 = 50;
 const EXPIRY_TIMEOUT_MILLIS: i16 = 5000;
@@ -122,15 +122,11 @@ impl<N: Network, DB: database::DB, VM: vm::VMExecution> SimpleFSM<N, DB, VM> {
             let get_blocks = Message::new_get_blocks(GetBlocks {
                 locator: last_finalized.header().hash,
             });
-            if let Err(e) = self
-                .network
+            self.network
                 .read()
                 .await
                 .send_to_alive_peers(&get_blocks, REDUNDANCY_PEER_FACTOR)
-                .await
-            {
-                warn!("Unable to request GetBlocks {e}");
-            }
+                .await;
         } else {
             error!("could not request blocks");
         }
@@ -674,14 +670,11 @@ impl<DB: database::DB, VM: vm::VMExecution, N: Network> InSyncImpl<DB, VM, N> {
         let req = GetResource::new(inv, this_peer, u64::MAX, 1);
         debug!(event = "request block by height", ?req, ?peer_addr);
 
-        if let Err(err) = network
+        network
             .read()
             .await
             .send_to_peer(&Message::new_get_resource(req), peer_addr)
             .await
-        {
-            warn!("could not request block {err}")
-        }
     }
 
     async fn on_heartbeat(&mut self) -> anyhow::Result<bool> {
@@ -747,15 +740,11 @@ impl<DB: database::DB, VM: vm::VMExecution, N: Network>
         // Request missing blocks from source peer
         let gb_msg = Message::new_get_blocks(GetBlocks { locator });
 
-        if let Err(e) = self
-            .network
+        self.network
             .read()
             .await
             .send_to_peer(&gb_msg, dest_addr)
-            .await
-        {
-            warn!("Unable to send GetBlocks: {e}")
-        };
+            .await;
 
         // add to the pool
         let key = blk.header().height;
@@ -873,12 +862,9 @@ impl<DB: database::DB, VM: vm::VMExecution, N: Network>
 async fn flood_request<N: Network>(network: &Arc<RwLock<N>>, inv: &Inv) {
     debug!(event = "flood_request", ?inv);
 
-    if let Err(err) = network
+    network
         .read()
         .await
         .flood_request(inv, None, DEFAULT_HOPS_LIMIT)
-        .await
-    {
-        warn!("could not request block {err}")
-    };
+        .await;
 }

node/src/chain/metrics.rs

@@ -82,8 +82,7 @@ mod tests {
         assert_eq!(metric.average().expect("positive number"), expected);
 
         // Marshal/Unmarshal
-        let mut buf = Vec::new();
-        metric.write(&mut buf).expect("all written");
+        let buf = metric.write_to_vec();
 
         assert_eq!(
             AverageElapsedTime::read(&mut &buf[..])

node/src/database/rocksdb.rs

@@ -290,19 +290,17 @@ impl<'db, DB: DBAccess> Ledger for DBTransaction<'db, DB> {
         {
             let cf = self.ledger_cf;
 
-            let mut buf = vec![];
-            LightBlock {
+            let block = LightBlock {
                 header: header.clone(),
                 transactions_ids: txs
                     .iter()
                     .map(|t| t.inner.id())
                     .collect::<Vec<[u8; 32]>>(),
 
-                faults_ids: faults.iter().map(|f| f.hash()).collect::<Vec<_>>(),
-            }
-            .write(&mut buf)?;
+                faults_ids: faults.iter().map(|f| f.hash()).collect(),
+            };
 
-            self.put_cf(cf, header.hash, buf)?;
+            self.put_cf(cf, header.hash, block.write_to_vec())?;
         }
 
         // Update metadata values
@@ -315,8 +313,7 @@ impl<'db, DB: DBAccess> Ledger for DBTransaction<'db, DB> {
 
             // store all block transactions
             for tx in txs {
-                let mut d = vec![];
-                tx.write(&mut d)?;
+                let d = tx.write_to_vec();
                 self.put_cf(cf, tx.inner.id(), d)?;
             }
         }
@@ -327,9 +324,9 @@ impl<'db, DB: DBAccess> Ledger for DBTransaction<'db, DB> {
 
             // store all block faults
             for f in faults {
-                let mut d = vec![];
-                f.write(&mut d)?;
-                self.put_cf(cf, f.hash(), d)?;
+                let key = f.hash();
+                let value = f.write_to_vec();
+                self.put_cf(cf, key, value)?;
             }
         }
         self.store_block_label(header.height, &header.hash, label)?;
@@ -570,8 +567,7 @@ impl<'db, DB: DBAccess> Candidate for DBTransaction<'db, DB> {
     /// Returns `Ok(())` if the block is successfully stored, or an error if the
     /// operation fails.
     fn store_candidate_block(&self, b: ledger::Block) -> Result<()> {
-        let mut serialized = vec![];
-        b.write(&mut serialized)?;
+        let serialized = b.write_to_vec();
 
         self.inner
             .put_cf(self.candidates_cf, b.header().hash, serialized)?;
@@ -681,10 +677,9 @@ impl<'db, DB: DBAccess> Persist for DBTransaction<'db, DB> {
 impl<'db, DB: DBAccess> Mempool for DBTransaction<'db, DB> {
     fn add_tx(&self, tx: &ledger::Transaction) -> Result<()> {
         // Map Hash to serialized transaction
-        let mut tx_data = vec![];
-        tx.write(&mut tx_data)?;
-
+        let tx_data = tx.write_to_vec();
         let hash = tx.id();
+
         self.put_cf(self.mempool_cf, hash, tx_data)?;
 
         // Add Secondary indexes //

node/src/databroker.rs

@@ -140,10 +140,7 @@ impl<N: Network, DB: database::DB, VM: vm::VMExecution>
                         // Send response
                         let net = network.read().await;
                         for msg in resp.msgs {
-                            let send = net.send_to_peer(&msg, resp.recv_peer);
-                            if let Err(e) = send.await {
-                                warn!("Unable to send_to_peer {e}")
-                            };
+                            net.send_to_peer(&msg, resp.recv_peer).await;
 
                             // Mitigate pressure on UDP buffers.
                             // Needed only in localnet.

node/src/lib.rs

@@ -40,29 +40,25 @@ pub type BoxedFilter = Box<dyn Filter + Sync + Send>;
 #[async_trait]
 pub trait Network: Send + Sync + 'static {
     /// Broadcasts a fire-and-forget message.
-    async fn broadcast(&self, msg: &Message) -> anyhow::Result<()>;
+    async fn broadcast(&self, msg: &Message);
 
     /// Broadcasts a request message
     async fn flood_request(
         &self,
         msg_inv: &Inv,
         ttl_as_sec: Option<u64>,
         hops_limit: u16,
-    ) -> anyhow::Result<()>;
+    );
 
     /// Sends a message to a specified peer.
     async fn send_to_peer(
         &self,
         msg: &Message,
         peer_addr: std::net::SocketAddr,
-    ) -> anyhow::Result<()>;
+    );
 
     /// Sends to random set of alive peers.
-    async fn send_to_alive_peers(
-        &self,
-        msg: &Message,
-        amount: usize,
-    ) -> anyhow::Result<()>;
+    async fn send_to_alive_peers(&self, msg: &Message, amount: usize);
 
     /// Routes any message of the specified type to this queue.
     async fn add_route(

node/src/mempool.rs

@@ -15,7 +15,7 @@ use node_data::message::{AsyncQueue, Payload, Topics};
 use std::sync::Arc;
 use thiserror::Error;
 use tokio::sync::RwLock;
-use tracing::{error, info, warn};
+use tracing::{error, info};
 
 const TOPICS: &[u8] = &[Topics::Tx as u8];
 
@@ -83,10 +83,7 @@ impl<N: Network, DB: database::DB, VM: vm::VMExecution>
                             continue;
                         }
 
-                        let network = network.read().await;
-                        if let Err(e) = network.broadcast(&msg).await {
-                            warn!("Unable to broadcast accepted tx: {e}")
-                        };
+                        network.read().await.broadcast(&msg).await;
                     }
                     _ => error!("invalid inbound message payload"),
                 }