api cleanup

clabby · clabby · commit 2edcf84c4fb6 · 2025-09-24T21:14:46.000-07:00
diff --git a/consensus/src/marshal/actor.rs b/consensus/src/marshal/actor.rs
@@ -15,7 +15,6 @@ use crate::{
     Block, Reporter,
 };
 use commonware_codec::{Decode, Encode};
-use commonware_coding::reed_solomon::Chunk;
 use commonware_cryptography::{
     bls12381::primitives::variant::Variant, Committable, Hasher, PublicKey,
 };
diff --git a/consensus/src/marshal/ingress/coding.rs b/consensus/src/marshal/ingress/coding.rs
@@ -39,11 +39,6 @@ struct BlockSubscription<B: Block> {
     subscribers: Vec<oneshot::Sender<B>>,
 }
 
-/// A subscription for a chunk by its commitment and index.
-struct ChunkSubscription<H: Hasher> {
-    subscribers: Vec<oneshot::Sender<Chunk<H>>>,
-}
-
 /// A layer that handles receiving erasure coded [Block]s from the [Actor](super::super::actor::Actor),
 /// broadcasting them to peers, and reassembling them from received [Shard]s.
 pub struct ShardLayer<P, B, H>
@@ -131,6 +126,9 @@ where
             .map(|c| c.chunk)
             .collect::<Vec<_>>();
 
+        // TODO: Make sure min is all valid chunks
+        // TODO: If we do encounter a block that's invalid, block the peer.
+
         if coded_chunks.len() < min as usize {
             // Not enough chunks to recover the block yet.
             debug!(
@@ -149,9 +147,9 @@ where
         // Attempt to decode the block from the recovered data.
         let block = B::decode_cfg(&mut recovered.as_slice(), &self.block_codec_cfg)?;
 
-        // Attempt to resolve any open subscriptions for this block.
-        if let Some(mut subs) = self.block_subscriptions.remove(&commitment) {
-            for sub in subs.subscribers.drain(..) {
+        // Notify any subscribers that have been waiting for this block.
+        if let Some(mut sub) = self.block_subscriptions.remove(&commitment) {
+            for sub in sub.subscribers.drain(..) {
                 let _ = sub.send(block.clone());
             }
         }
@@ -198,10 +196,7 @@ where
         commitment: B::Commitment,
         index: u16,
     ) -> Option<Shard<B, H>> {
-        let mut buf = vec![0u8; <H::Digest as FixedSize>::SIZE + 2];
-        buf[..<H::Digest as FixedSize>::SIZE].copy_from_slice(commitment.as_ref());
-        buf[<H::Digest as FixedSize>::SIZE..].copy_from_slice(index.to_le_bytes().as_ref());
-        let index_hash = H::hash(buf.as_ref());
+        let index_hash = shard_uuid::<B, H>(commitment, index);
         self.mailbox
             .get(None, commitment, Some(index_hash))
             .await
@@ -221,10 +216,7 @@ where
         index: u16,
         responder: oneshot::Sender<Shard<B, H>>,
     ) {
-        let mut buf = vec![0u8; <H::Digest as FixedSize>::SIZE + 2];
-        buf[..<H::Digest as FixedSize>::SIZE].copy_from_slice(commitment.as_ref());
-        buf[<H::Digest as FixedSize>::SIZE..].copy_from_slice(index.to_le_bytes().as_ref());
-        let index_hash = H::hash(buf.as_ref());
+        let index_hash = shard_uuid::<B, H>(commitment, index);
         self.mailbox
             .subscribe_prepared(None, commitment, Some(index_hash), responder)
             .await;
@@ -320,11 +312,7 @@ where
     type Digest = H::Digest;
 
     fn digest(&self) -> Self::Digest {
-        let mut buf = vec![0u8; <H::Digest as FixedSize>::SIZE + 2];
-        buf[..<H::Digest as FixedSize>::SIZE].copy_from_slice(self.commitment.as_ref());
-        buf[<H::Digest as FixedSize>::SIZE..]
-            .copy_from_slice(self.chunk.index.to_le_bytes().as_ref());
-        H::hash(buf.as_ref())
+        shard_uuid::<B, H>(self.commitment, self.chunk.index)
     }
 }
 
@@ -407,31 +395,32 @@ where
     config: (u16, u16),
     /// The erasure coding commitment.
     commitment: H::Digest,
+    /// The coded chunks.
+    chunks: Vec<Chunk<H>>,
 }
 
 impl<B, H> CodedBlock<B, H>
 where
     B: Block<Digest = H::Digest, Commitment = H::Digest>,
     H: Hasher,
 {
-    /// Erasure codes the block to create the commitment.
-    fn commit(inner: &B, config: (u16, u16)) -> H::Digest {
+    /// Erasure codes the block.
+    fn encode(inner: &B, config: (u16, u16)) -> (H::Digest, Vec<Chunk<H>>) {
         let mut buf = Vec::with_capacity(config.encode_size() + inner.encode_size());
         inner.write(&mut buf);
         config.write(&mut buf);
 
-        let (commitment, _) =
-            reed_solomon::encode::<H>(config.0, config.1, buf).expect("failed to commit to block");
-        commitment
+        reed_solomon::encode::<H>(config.0, config.1, buf).expect("failed to commit to block")
     }
 
     /// Create a new [CodedBlock] from a [Block] and a configuration.
     pub fn new(inner: B, config: (u16, u16)) -> Self {
-        let commitment = Self::commit(&inner, config);
+        let (commitment, chunks) = Self::encode(&inner, config);
         Self {
             inner,
             config,
             commitment,
+            chunks,
         }
     }
 
@@ -444,6 +433,16 @@ where
     pub fn take_inner(self) -> B {
         self.inner
     }
+
+    /// Returns the erasure coding configuration.
+    pub fn config(&self) -> (u16, u16) {
+        self.config
+    }
+
+    /// Returns a reference to the coded chunks.
+    pub fn chunks(&self) -> &[Chunk<H>] {
+        self.chunks.as_slice()
+    }
 }
 
 impl<B, H> Write for CodedBlock<B, H>
@@ -470,12 +469,13 @@ where
     ) -> Result<Self, commonware_codec::Error> {
         let inner = B::read_cfg(buf, cfg)?;
         let config = <(u16, u16)>::read_cfg(buf, &((), ()))?;
-        let commitment = Self::commit(&inner, config);
+        let (commitment, chunks) = Self::encode(&inner, config);
 
         Ok(Self {
             inner,
             config,
             commitment,
+            chunks,
         })
     }
 }
@@ -547,6 +547,18 @@ where
 {
 }
 
+/// Creates a unique identifier for a shard based on the block commitment and shard index.
+fn shard_uuid<B, H>(commitment: B::Commitment, index: u16) -> H::Digest
+where
+    B: Block<Digest = H::Digest, Commitment = H::Digest>,
+    H: Hasher,
+{
+    let mut buf = vec![0u8; H::Digest::SIZE + u16::SIZE];
+    buf[..H::Digest::SIZE].copy_from_slice(commitment.as_ref());
+    buf[H::Digest::SIZE..].copy_from_slice(index.to_le_bytes().as_ref());
+    H::hash(buf.as_ref())
+}
+
 #[cfg(test)]
 mod test {
     use super::*;
diff --git a/consensus/src/marshal/mod.rs b/consensus/src/marshal/mod.rs
@@ -334,19 +334,6 @@ mod tests {
         }
     }
 
-    #[allow(clippy::type_complexity)]
-    fn shard(block: &B, peers: &[P]) -> (D, (u16, u16), Vec<(P, Chunk<H>)>) {
-        let (total, min) = (peers.len() as u16, peers.len() as u16 / 2);
-        let block = CodedBlock::<B, H>::new(block.clone(), (total, min));
-        let (commitment, chunks) =
-            reed_solomon::encode::<H>(total, min, block.encode().into()).unwrap();
-        (
-            commitment,
-            (total, min),
-            peers.iter().cloned().zip(chunks).collect(),
-        )
-    }
-
     #[test_traced("DEBUG")]
     fn test_finalize_good_links() {
         for seed in 0..5 {
@@ -401,14 +388,14 @@ mod tests {
             setup_network_links(&mut oracle, &peers, link.clone()).await;
 
             // Generate blocks, skipping the genesis block.
-            let mut blocks = Vec::<B>::new();
+            let mut blocks = Vec::<CodedBlock<B, H>>::new();
             let mut parent = Sha256::hash(b"");
             let (total, min) = (peers.len() as u16, (peers.len() / 2) as u16);
             for i in 1..=NUM_BLOCKS {
-                let block = Block::new::<Sha256>(parent, i, i);
-                let block = CodedBlock::<B, H>::new(block, (total, min));
+                let inner = Block::new::<Sha256>(parent, i, i);
+                let block = CodedBlock::<B, H>::new(inner, (total, min));
                 parent = block.commitment();
-                blocks.push(block.take_inner());
+                blocks.push(block);
             }
 
             // Broadcast and finalize blocks in random order
@@ -426,8 +413,10 @@ mod tests {
                 let actor_index: usize = (height % (NUM_VALIDATORS as u64)) as usize;
                 let mut actor = actors[actor_index].clone();
 
-                let (commitment, config, chunks) = shard(block, &peers);
-                actor.broadcast(commitment, config, chunks).await;
+                let peers_and_chunks = peers.iter().cloned().zip(block.chunks().to_vec()).collect();
+                actor
+                    .broadcast(block.commitment(), block.config(), peers_and_chunks)
+                    .await;
 
                 // Wait for the block chunks to be delivered; Before making notarization votes,
                 // the chunks must be present.
@@ -436,7 +425,7 @@ mod tests {
                 let proposal = Proposal {
                     round,
                     parent: height.checked_sub(1).unwrap(),
-                    payload: commitment,
+                    payload: block.commitment(),
                 };
 
                 // All validators send a notarization for the block.
