group id enum

Author: michael-yxchen

monad-raptorcast/benches/encode_bench.rs

@@ -22,11 +22,12 @@ use monad_crypto::certificate_signature::{CertificateSignature, CertificateSigna
 use monad_dataplane::udp::DEFAULT_SEGMENT_SIZE;
 use monad_raptorcast::{
     packet,
+    udp::GroupId,
     util::{BuildTarget, EpochValidators, Redundancy},
 };
 use monad_secp::SecpSignature;
 use monad_testutil::signing::get_key;
-use monad_types::{NodeId, Stake};
+use monad_types::{Epoch, NodeId, Stake};
 
 const NUM_NODES: usize = 100;
 
@@ -63,8 +64,8 @@ pub fn bench_build_messages(c: &mut Criterion, name: &str, message_size: usize,
                 DEFAULT_SEGMENT_SIZE, // segment_size
                 message.clone(),
                 Redundancy::from_u8(2),
-                0, // epoch_no
-                0, // unix_ts_ms
+                GroupId::Primary(Epoch(0)), // epoch_no
+                0,                          // unix_ts_ms
                 build_target.clone(),
                 &known_addrs,
             );

monad-raptorcast/benches/raptor_bench.rs

@@ -24,11 +24,11 @@ use monad_dataplane::udp::DEFAULT_SEGMENT_SIZE;
 use monad_raptor::ManagedDecoder;
 use monad_raptorcast::{
     packet::build_messages,
-    udp::{parse_message, MAX_REDUNDANCY, SIGNATURE_CACHE_SIZE},
+    udp::{parse_message, GroupId, MAX_REDUNDANCY, SIGNATURE_CACHE_SIZE},
     util::{BuildTarget, EpochValidators, Redundancy},
 };
 use monad_secp::{KeyPair, SecpSignature};
-use monad_types::{NodeId, Stake};
+use monad_types::{Epoch, NodeId, Stake};
 
 #[allow(clippy::useless_vec)]
 pub fn criterion_benchmark(c: &mut Criterion) {
@@ -71,8 +71,8 @@ pub fn criterion_benchmark(c: &mut Criterion) {
                 DEFAULT_SEGMENT_SIZE, // segment_size
                 message.clone(),
                 Redundancy::from_u8(2),
-                0, // epoch_no
-                0, // unix_ts_ms
+                GroupId::Primary(Epoch(0)), // epoch_no
+                0,                          // unix_ts_ms
                 BuildTarget::Raptorcast(epoch_validators),
                 &known_addresses,
             );
@@ -112,8 +112,8 @@ pub fn criterion_benchmark(c: &mut Criterion) {
             DEFAULT_SEGMENT_SIZE, // segment_size
             message.clone(),
             Redundancy::from_u8(2),
-            0, // epoch_no
-            0, // unix_ts_ms
+            GroupId::Primary(Epoch(0)), // epoch_no
+            0,                          // unix_ts_ms
             BuildTarget::Raptorcast(epoch_validators),
             &known_addresses,
         )

monad-raptorcast/src/decoding.rs

@@ -436,19 +436,13 @@ pub struct PruneConfig {
 pub struct DecodingContext<'a, PT: PubKey> {
     validator_set: Option<&'a ValidatorSet<PT>>,
     unix_ts_now: UnixTimestamp,
-    current_epoch: Epoch,
 }
 
 impl<'a, PT: PubKey> DecodingContext<'a, PT> {
-    pub fn new(
-        validator_set: Option<&'a ValidatorSet<PT>>,
-        unix_ts_now: UnixTimestamp,
-        current_epoch: Epoch,
-    ) -> Self {
+    pub fn new(validator_set: Option<&'a ValidatorSet<PT>>, unix_ts_now: UnixTimestamp) -> Self {
         Self {
             validator_set,
             unix_ts_now,
-            current_epoch,
         }
     }
 }
@@ -806,15 +800,11 @@ impl<PT: PubKey> AuthorIndex<PT> {
             .prune_config
             .max_unix_ts_ms_delta
             .and_then(|delta| context.unix_ts_now.checked_sub(delta));
-        let epoch_threshold: Option<Epoch> = self
-            .prune_config
-            .max_epoch_delta
-            .and_then(|delta| context.current_epoch.checked_sub(delta));
 
         let mut evicted_keys = PrunedKeys::empty();
 
         // we first try only pruning expired keys
-        let expired_keys = author_index.prune_expired(unix_ts_threshold, epoch_threshold);
+        let expired_keys = author_index.prune_expired(unix_ts_threshold);
         evicted_keys.extend(expired_keys);
 
         // if still over quota, compact the cache to fit under quota
@@ -847,18 +837,14 @@ impl<PT: PubKey> AuthorIndex<PT> {
             .prune_config
             .max_unix_ts_ms_delta
             .and_then(|delta| context.unix_ts_now.checked_sub(delta));
-        let epoch_threshold: Option<Epoch> = self
-            .prune_config
-            .max_epoch_delta
-            .and_then(|delta| context.current_epoch.checked_sub(delta));
 
         let mut authors_to_drop = vec![];
         let mut total_slots = 0;
 
         let mut pruned_keys = PrunedKeys::empty();
         for (author, author_index) in &mut self.per_author_index {
             total_slots += author_index.len();
-            pruned_keys.extend(author_index.prune_expired(unix_ts_threshold, epoch_threshold));
+            pruned_keys.extend(author_index.prune_expired(unix_ts_threshold));
 
             if author_index.is_empty() {
                 authors_to_drop.push(*author);
@@ -1053,11 +1039,7 @@ impl PerAuthorIndex {
     }
 
     // Remove expired entries.
-    pub fn prune_expired(
-        &mut self,
-        unix_ts_threshold: Option<UnixTimestamp>,
-        epoch_threshold: Option<Epoch>,
-    ) -> PrunedKeys {
+    pub fn prune_expired(&mut self, unix_ts_threshold: Option<UnixTimestamp>) -> PrunedKeys {
         let mut evicted_keys = PrunedKeys::empty();
         // first, we prune all expired keys
         if let Some(threshold) = unix_ts_threshold {
@@ -1551,7 +1533,7 @@ mod test {
     use rand::seq::SliceRandom as _;
 
     use super::*;
-    use crate::util::BroadcastMode;
+    use crate::{udp::GroupId, util::BroadcastMode};
     type PT = monad_crypto::NopPubKey;
 
     const EPOCH: Epoch = Epoch(1);
@@ -1620,7 +1602,7 @@ mod test {
                 // these fields are never touched in this module
                 recipient_hash: HexBytes([0; 20]),
                 message: Bytes::new(),
-                group_id: EPOCH.0,
+                group_id: GroupId::Primary(EPOCH),
                 unix_ts_ms,
             };
             messages.push(message);
@@ -1633,7 +1615,7 @@ mod test {
         let app_message = Bytes::from(vec![1u8; APP_MESSAGE_LEN]);
         let author = node_id(0);
         let symbols = make_symbols(&app_message, author, UNIX_TS_MS);
-        let context = DecodingContext::new(None, UNIX_TS_MS, EPOCH);
+        let context = DecodingContext::new(None, UNIX_TS_MS);
 
         for n in 0..MIN_DECODABLE_SYMBOLS {
             let mut cache = make_cache(10, 10, 10);
@@ -1710,7 +1692,7 @@ mod test {
         // single slot per tier is enough
         let mut cache = make_cache(1, 1, 1);
 
-        let context = DecodingContext::new(Some(&validator_set), UNIX_TS_MS, EPOCH);
+        let context = DecodingContext::new(Some(&validator_set), UNIX_TS_MS);
         let res = try_decode_all(&mut cache, &context, all_symbols.iter())
             .expect("Decoding should succeed");
 
@@ -1723,7 +1705,7 @@ mod test {
         let app_message = Bytes::from(vec![1u8; APP_MESSAGE_LEN]);
         let author = node_id(0);
         let symbols = make_symbols(&app_message, author, UNIX_TS_MS);
-        let context = DecodingContext::new(None, UNIX_TS_MS, EPOCH);
+        let context = DecodingContext::new(None, UNIX_TS_MS);
         let mut cache = make_cache(10, 10, 10);
 
         // Decode a message completely.
@@ -1746,7 +1728,7 @@ mod test {
         let app_message = Bytes::from(vec![1u8; APP_MESSAGE_LEN]);
         let author = node_id(0);
         let symbols = make_symbols(&app_message, author, old_ts);
-        let context = DecodingContext::new(None, old_ts, EPOCH);
+        let context = DecodingContext::new(None, old_ts);
 
         // Insert an old message.
         let _ = cache.try_decode(&symbols[0], &context);
@@ -1757,7 +1739,7 @@ mod test {
             let new_app_message = Bytes::from(vec![2u8; APP_MESSAGE_LEN]);
             let new_author = node_id(i);
             let new_symbols = make_symbols(&new_app_message, new_author, UNIX_TS_MS);
-            let new_context = DecodingContext::new(None, UNIX_TS_MS, EPOCH);
+            let new_context = DecodingContext::new(None, UNIX_TS_MS);
             let _ = cache.try_decode(&new_symbols[0], &new_context);
             assert!(cache.consistency_breaches().is_empty());
         }
@@ -1801,7 +1783,7 @@ mod test {
         config.validator_tier.min_slots_per_validator = Some(2);
 
         let mut cache = DecoderCache::new(config);
-        let context = DecodingContext::new(Some(&validator_set), UNIX_TS_MS, EPOCH);
+        let context = DecodingContext::new(Some(&validator_set), UNIX_TS_MS);
         let res = try_decode_all(&mut cache, &context, all_symbols_part_1.iter())
             .expect("Decoding should succeed");
         assert!(cache.consistency_breaches().is_empty());
@@ -1883,7 +1865,7 @@ mod test {
         }
 
         let mut cache = DecoderCache::new(config);
-        let context = DecodingContext::new(Some(&validator_set), UNIX_TS_MS, EPOCH);
+        let context = DecodingContext::new(Some(&validator_set), UNIX_TS_MS);
         let res = try_decode_all(&mut cache, &context, all_symbols_part_1.iter())
             .expect("Decoding should succeed");
         assert!(cache.consistency_breaches().is_empty());
@@ -1930,7 +1912,7 @@ mod test {
         config.p2p_tier.min_slots_per_author = 2; // each author gets at least 2 slots
 
         let mut cache = DecoderCache::new(config);
-        let context = DecodingContext::new(None, UNIX_TS_MS, EPOCH);
+        let context = DecodingContext::new(None, UNIX_TS_MS);
         let res = try_decode_all(&mut cache, &context, all_symbols_part_1.iter())
             .expect("Decoding should succeed");
         assert!(cache.consistency_breaches().is_empty());
@@ -1996,7 +1978,7 @@ mod test {
         }
 
         let mut cache = DecoderCache::new(config);
-        let context = DecodingContext::new(None, UNIX_TS_MS, EPOCH);
+        let context = DecodingContext::new(None, UNIX_TS_MS);
         let res = try_decode_all(&mut cache, &context, all_symbols_part_1.iter())
             .expect("Decoding should succeed");
         assert!(cache.consistency_breaches().is_empty());
@@ -2022,7 +2004,7 @@ mod test {
         let app_message = Bytes::from(vec![1u8; APP_MESSAGE_LEN]);
         let author = node_id(0);
         let symbols = make_symbols(&app_message, author, UNIX_TS_MS);
-        let context = DecodingContext::new(None, UNIX_TS_MS, EPOCH);
+        let context = DecodingContext::new(None, UNIX_TS_MS);
 
         // Insert a valid symbol first.
         let _ = cache.try_decode(&symbols[0], &context);
@@ -2056,7 +2038,7 @@ mod test {
         config.p2p_tier.min_slots_per_author = 2;
 
         let mut cache = DecoderCache::new(config);
-        let context = DecodingContext::new(None, UNIX_TS_MS, EPOCH);
+        let context = DecodingContext::new(None, UNIX_TS_MS);
 
         // Fill the cache.
         let app_message0 = Bytes::from(vec![0u8; APP_MESSAGE_LEN]);
@@ -2095,7 +2077,7 @@ mod test {
         config.p2p_tier.min_slots_per_author = 5;
 
         let mut cache = DecoderCache::new(config);
-        let context = DecodingContext::new(None, UNIX_TS_MS, EPOCH);
+        let context = DecodingContext::new(None, UNIX_TS_MS);
 
         // take a single symbol for a given message
         let partial_symbol = |msg: u8, ts: UnixTimestamp| {

monad-raptorcast/src/lib.rs

@@ -58,6 +58,7 @@ use monad_validator::signature_collection::SignatureCollection;
 use raptorcast_secondary::{group_message::FullNodesGroupMessage, SecondaryRaptorCastModeConfig};
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 use tracing::{debug, debug_span, error, trace, warn};
+use udp::GroupId;
 use util::{BuildTarget, EpochValidators, FullNodes, Group, ReBroadcastGroupMap, Redundancy};
 
 pub mod config;
@@ -152,7 +153,7 @@ where
         let message_builder =
             OwnedMessageBuilder::new(config.shared_key.clone(), peer_discovery_driver.clone())
                 .segment_size(segment_size_for_mtu(config.mtu))
-                .group_id(current_epoch)
+                .group_id(GroupId::Primary(current_epoch))
                 .redundancy(redundancy);
 
         Self {
@@ -323,7 +324,7 @@ where
                 if let Some(rc_chunks) = self
                     .message_builder
                     .prepare()
-                    .group_id(epoch)
+                    .group_id(GroupId::Primary(epoch))
                     .build_unicast_msg(&outbound_message, &build_target)
                 {
                     self.dataplane_writer
@@ -473,7 +474,7 @@ where
                         }
 
                         self.current_epoch = epoch;
-                        self.message_builder.set_group_id(epoch);
+                        self.message_builder.set_group_id(GroupId::Primary(epoch));
 
                         while let Some(entry) = self.epoch_validators.first_entry() {
                             if *entry.key() + Epoch(1) < self.current_epoch {
@@ -577,7 +578,7 @@ where
                         if let Some(rc_chunks) = self
                             .message_builder
                             .prepare_with_peer_lookup(&node_addrs)
-                            .group_id(epoch)
+                            .group_id(GroupId::Primary(epoch))
                             .build_unicast_msg(&outbound_message, &build_target)
                         {
                             self.dataplane_writer.udp_write_unicast(rc_chunks);
@@ -709,7 +710,6 @@ where
             let decoded_app_messages = {
                 // FIXME: pass dataplane as arg to handle_message
                 this.udp_state.handle_message(
-                    this.current_epoch,
                     &this.rebroadcast_map, // contains the NodeIds for all the RC participants for each epoch
                     &this.epoch_validators,
                     |targets, payload, bcast_stride| {

monad-raptorcast/src/packet/assembler.rs

@@ -30,7 +30,7 @@ use super::{
     assigner::{ChunkAssignment, ChunkOrder},
     BuildError, Collector, PeerAddrLookup, Result, UdpMessage,
 };
-use crate::{util::Redundancy, SIGNATURE_SIZE};
+use crate::{udp::GroupId, util::Redundancy, SIGNATURE_SIZE};
 
 #[derive(Default, Clone, Copy, PartialEq, Eq)]
 pub enum AssembleMode {
@@ -577,7 +577,7 @@ pub(crate) fn build_header(
     version: u16,
     broadcast_type: BroadcastType,
     merkle_tree_depth: u8,
-    group_id: u64,
+    group_id: GroupId,
     unix_ts_ms: u64,
     app_message: &[u8],
 ) -> Result<Bytes> {
@@ -610,8 +610,9 @@ pub(crate) fn build_header(
     broadcast_byte |= merkle_tree_depth & 0b0000_1111;
     cursor_broadcast_merkle_depth[0] = broadcast_byte;
 
-    let (cursor_epoch_no, cursor) = cursor.split_at_mut_checked(8).expect("header too short");
-    cursor_epoch_no.copy_from_slice(&group_id.to_le_bytes());
+    let group_id: u64 = group_id.into();
+    let (cursor_group_id, cursor) = cursor.split_at_mut_checked(8).expect("header too short");
+    cursor_group_id.copy_from_slice(&group_id.to_le_bytes());
 
     let (cursor_unix_ts_ms, cursor) = cursor.split_at_mut_checked(8).expect("header too short");
     cursor_unix_ts_ms.copy_from_slice(&unix_ts_ms.to_le_bytes());