Overload Raptor Chunk Header Epoch

A full node may join new RaptorCast groups from the same validator while
it's syncing or stuck. In such cases, it may rebroadcast to an expired
group because its local round hasn't advanced, and expired groups are
not garbage-collected. As a result, many full nodes end up receiving
unsolicited Raptor chunks from groups they no longer belong to.

The root cause is that the node selects the rebroadcast group based on
its local round. This PR resolves the issue by overloading the epoch
field in the Raptor chunk header setting it to the round number for
secondary raptorcast. This allows syncing or stuck full nodes to use
the round number to identify and service the correct group

Author: michael-yxchen

monad-consensus-state/src/command.rs

@@ -61,6 +61,7 @@ where
     },
     PublishToFullNodes {
         epoch: Epoch,
+        round: Round,
         message: Verified<ST, Validated<ConsensusMessage<ST, SCT, EPT>>>,
     },
     /// Schedule a timeout event for `round` to be emitted in `duration`
@@ -137,6 +138,7 @@ where
                 cmds.push(ConsensusCommand::EnterRound(epoch, round));
                 cmds.push(ConsensusCommand::PublishToFullNodes {
                     epoch,
+                    round: high_certificate.round(),
                     message: ConsensusMessage {
                         version,
                         message: ProtocolMessage::AdvanceRound(AdvanceRoundMessage {

monad-executor-glue/src/lib.rs

@@ -66,8 +66,12 @@ pub enum RouterCommand<ST: CertificateSignatureRecoverable, OM> {
         message: OM,
         priority: UdpPriority,
     },
+    // Primary publishing embeds epoch as gropu_id in chunk header. Secondary
+    // publishing embeds round as group_id in chunk header, as rebroadcasting
+    // periods are defined in rounds
     PublishToFullNodes {
-        epoch: Epoch, // Epoch gets embedded into the raptorcast message
+        epoch: Epoch,
+        round: Round,
         message: OM,
     },
     AddEpochValidatorSet {
@@ -103,9 +107,14 @@ impl<ST: CertificateSignatureRecoverable, OM> Debug for RouterCommand<ST, OM> {
                 .field("target", target)
                 .field("priority", priority)
                 .finish(),
-            Self::PublishToFullNodes { epoch, message: _ } => f
+            Self::PublishToFullNodes {
+                epoch,
+                round,
+                message: _,
+            } => f
                 .debug_struct("PublishToFullNodes")
                 .field("epoch", epoch)
+                .field("round", round)
                 .finish(),
             Self::AddEpochValidatorSet {
                 epoch,

monad-raptorcast/src/decoding.rs

@@ -762,12 +762,7 @@ impl<PT: PubKey> AuthorIndex<PT> {
 
         let message_size = message.app_message_len as MessageSize;
 
-        index.insert(
-            cache_key.clone(),
-            message.unix_ts_ms,
-            Epoch(message.epoch),
-            message_size,
-        );
+        index.insert(cache_key.clone(), message.unix_ts_ms, message_size);
         self.used_size += message_size;
 
         if index.is_overquota() {
@@ -1000,8 +995,7 @@ struct PerAuthorIndex {
     quota: Quota,
     used_size: MessageSize,
     time_index: BTreeSet<(UnixTimestamp, CacheKey)>,
-    epoch_index: BTreeSet<(Epoch, CacheKey)>,
-    reverse_index: HashMap<CacheKey, (UnixTimestamp, Epoch, MessageSize)>,
+    reverse_index: HashMap<CacheKey, (UnixTimestamp, MessageSize)>,
 }
 
 impl PerAuthorIndex {
@@ -1010,7 +1004,6 @@ impl PerAuthorIndex {
             quota,
             used_size: 0,
             time_index: BTreeSet::new(),
-            epoch_index: BTreeSet::new(),
             reverse_index: HashMap::new(),
         }
     }
@@ -1028,12 +1021,11 @@ impl PerAuthorIndex {
     }
 
     pub fn remove(&mut self, key: &CacheKey) -> PrunedKeys {
-        let Some((unix_ts_ms, epoch, size)) = self.reverse_index.remove(key) else {
+        let Some((unix_ts_ms, size)) = self.reverse_index.remove(key) else {
             return PrunedKeys::empty();
         };
 
         self.time_index.remove(&(unix_ts_ms, key.clone()));
-        self.epoch_index.remove(&(epoch, key.clone()));
         self.used_size -= size;
         PrunedKeys::singleton(key.clone(), size)
     }
@@ -1054,17 +1046,9 @@ impl PerAuthorIndex {
             .collect()
     }
 
-    pub fn insert(
-        &mut self,
-        cache_key: CacheKey,
-        unix_ts_ms: UnixTimestamp,
-        epoch: Epoch,
-        size: MessageSize,
-    ) {
+    pub fn insert(&mut self, cache_key: CacheKey, unix_ts_ms: UnixTimestamp, size: MessageSize) {
         self.time_index.insert((unix_ts_ms, cache_key.clone()));
-        self.epoch_index.insert((epoch, cache_key.clone()));
-        self.reverse_index
-            .insert(cache_key, (unix_ts_ms, epoch, size));
+        self.reverse_index.insert(cache_key, (unix_ts_ms, size));
         self.used_size += size;
     }
 
@@ -1079,9 +1063,6 @@ impl PerAuthorIndex {
         if let Some(threshold) = unix_ts_threshold {
             evicted_keys.extend(self.prune_by_time(threshold));
         }
-        if let Some(threshold) = epoch_threshold {
-            evicted_keys.extend(self.prune_by_epoch(threshold));
-        }
         evicted_keys
     }
 
@@ -1115,17 +1096,6 @@ impl PerAuthorIndex {
         self.remove_many(&to_prune_keys)
     }
 
-    fn prune_by_epoch(&mut self, epoch_threshold: Epoch) -> PrunedKeys {
-        let mut to_prune_keys = vec![];
-        for (epoch, key) in &self.epoch_index {
-            if *epoch >= epoch_threshold {
-                break;
-            }
-            to_prune_keys.push(key.clone());
-        }
-        self.remove_many(&to_prune_keys)
-    }
-
     fn prune_by_slots(&mut self, target_len: usize) -> PrunedKeys {
         let slots_to_free_up = self.len().saturating_sub(target_len);
         if slots_to_free_up == 0 {
@@ -1150,21 +1120,15 @@ impl PerAuthorIndex {
     #[cfg(test)]
     fn consistency_breaches(&self, prefix: &str) -> Vec<String> {
         let mut breaches = vec![];
-        if self.epoch_index.len() != self.reverse_index.len() {
-            breaches.push(format!("{prefix}.epoch-index-size-mismatch"));
-        }
         if self.time_index.len() != self.reverse_index.len() {
             breaches.push(format!("{prefix}.time-index-size-mismatch"));
         }
 
         let mut used_size = self.used_size;
-        for (key, (unix_ts, epoch, _size)) in &self.reverse_index {
+        for (key, (unix_ts, _size)) in &self.reverse_index {
             if !self.time_index.contains(&(*unix_ts, key.clone())) {
                 breaches.push(format!("{prefix}.time-index-missing-key"));
             }
-            if !self.epoch_index.contains(&(*epoch, key.clone())) {
-                breaches.push(format!("{prefix}.epoch-index-missing-key"));
-            }
             used_size -= *_size;
         }
 
@@ -1656,7 +1620,7 @@ mod test {
                 // these fields are never touched in this module
                 recipient_hash: HexBytes([0; 20]),
                 message: Bytes::new(),
-                epoch: EPOCH.0,
+                group_id: EPOCH.0,
                 unix_ts_ms,
             };
             messages.push(message);

monad-raptorcast/src/lib.rs

@@ -152,7 +152,7 @@ where
         let message_builder =
             OwnedMessageBuilder::new(config.shared_key.clone(), peer_discovery_driver.clone())
                 .segment_size(segment_size_for_mtu(config.mtu))
-                .epoch_no(current_epoch)
+                .group_id(current_epoch)
                 .redundancy(redundancy);
 
         Self {
@@ -323,7 +323,7 @@ where
                 if let Some(rc_chunks) = self
                     .message_builder
                     .prepare()
-                    .epoch_no(epoch)
+                    .group_id(epoch)
                     .build_unicast_msg(&outbound_message, &build_target)
                 {
                     self.dataplane_writer
@@ -473,7 +473,7 @@ where
                         }
 
                         self.current_epoch = epoch;
-                        self.message_builder.set_epoch_no(epoch);
+                        self.message_builder.set_group_id(epoch);
 
                         while let Some(entry) = self.epoch_validators.first_entry() {
                             if *entry.key() + Epoch(1) < self.current_epoch {
@@ -532,7 +532,11 @@ where
                 } => {
                     self.handle_publish(target, message, priority, self_id);
                 }
-                RouterCommand::PublishToFullNodes { epoch, message } => {
+                RouterCommand::PublishToFullNodes {
+                    epoch,
+                    round: _,
+                    message,
+                } => {
                     let full_nodes_view = self.dedicated_full_nodes.view();
                     if self.is_dynamic_fullnode {
                         debug!("self is dynamic full node, skipping publishing to full nodes");
@@ -573,7 +577,7 @@ where
                         if let Some(rc_chunks) = self
                             .message_builder
                             .prepare_with_peer_lookup(&node_addrs)
-                            .epoch_no(epoch)
+                            .group_id(epoch)
                             .build_unicast_msg(&outbound_message, &build_target)
                         {
                             self.dataplane_writer.udp_write_unicast(rc_chunks);

monad-raptorcast/src/packet/assembler.rs

@@ -577,7 +577,7 @@ pub(crate) fn build_header(
     version: u16,
     broadcast_type: BroadcastType,
     merkle_tree_depth: u8,
-    epoch_no: u64,
+    group_id: u64,
     unix_ts_ms: u64,
     app_message: &[u8],
 ) -> Result<Bytes> {
@@ -586,7 +586,7 @@ pub(crate) fn build_header(
     //       Secondary broadcast bit,
     //       2 unused bits,
     //       4 bits for Merkle Tree Depth
-    // 8  // Epoch #
+    // 8  // Group id
     // 8  // Unix timestamp
     // 20 // AppMessage hash
     // 4  // AppMessage length
@@ -611,7 +611,7 @@ pub(crate) fn build_header(
     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(&epoch_no.to_le_bytes());
+    cursor_epoch_no.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());

monad-raptorcast/src/packet/builder.rs

@@ -92,7 +92,7 @@ where
     peer_lookup: PL,
 
     // required fields
-    epoch_no: Option<u64>,
+    group_id: Option<u64>,
     redundancy: Option<Redundancy>,
 
     // optional fields
@@ -111,7 +111,7 @@ where
         Self {
             key: self.key.clone(),
             peer_lookup: self.peer_lookup.clone(),
-            epoch_no: self.epoch_no,
+            group_id: self.group_id,
             redundancy: self.redundancy,
             unix_ts_ms: self.unix_ts_ms,
             segment_size: self.segment_size,
@@ -141,7 +141,7 @@ where
 
             // default fields
             redundancy: None,
-            epoch_no: None,
+            group_id: None,
             unix_ts_ms: TimestampMode::RealTime,
 
             // optional fields
@@ -162,8 +162,8 @@ where
         self
     }
 
-    pub fn epoch_no(mut self, epoch_no: impl Into<u64>) -> Self {
-        self.epoch_no = Some(epoch_no.into());
+    pub fn group_id(mut self, group_id: impl Into<u64>) -> Self {
+        self.group_id = Some(group_id.into());
         self
     }
 
@@ -187,16 +187,16 @@ where
     }
 
     // ----- Convenience methods for modifying the builder -----
-    pub fn set_epoch_no(&mut self, epoch_no: impl Into<u64>) {
-        self.epoch_no = Some(epoch_no.into());
+    pub fn set_group_id(&mut self, group_id: impl Into<u64>) {
+        self.group_id = Some(group_id.into());
     }
 
     // ----- Prepare override builder -----
     pub fn prepare(&self) -> PreparedMessageBuilder<'_, 'key, ST, PL, PL> {
         PreparedMessageBuilder {
             base: self,
             peer_lookup: None,
-            epoch_no: None,
+            group_id: None,
         }
     }
 
@@ -210,7 +210,7 @@ where
         PreparedMessageBuilder {
             base: self,
             peer_lookup: Some(peer_lookup),
-            epoch_no: None,
+            group_id: None,
         }
     }
 
@@ -256,7 +256,7 @@ where
 
     // Add extra override fields as needed
     peer_lookup: Option<PL2>,
-    epoch_no: Option<u64>,
+    group_id: Option<u64>,
 }
 
 impl<'base, 'key, ST, PL, PL2> PreparedMessageBuilder<'base, 'key, ST, PL, PL2>
@@ -266,21 +266,21 @@ where
     PL2: PeerAddrLookup<CertificateSignaturePubKey<ST>>,
 {
     // ----- Setters for overrides -----
-    pub fn epoch_no(mut self, epoch_no: impl Into<u64>) -> Self {
-        self.epoch_no = Some(epoch_no.into());
+    pub fn group_id(mut self, group_id: impl Into<u64>) -> Self {
+        self.group_id = Some(group_id.into());
         self
     }
 
     // ----- Parameter validation methods -----
-    fn unwrap_epoch_no(&self) -> Result<u64> {
-        if let Some(epoch_no) = self.epoch_no {
-            return Ok(epoch_no);
+    fn unwrap_group_id(&self) -> Result<u64> {
+        if let Some(group_id) = self.group_id {
+            return Ok(group_id);
         }
-        let epoch_no = self
+        let group_id = self
             .base
-            .epoch_no
-            .expect("epoch_no must be set before building");
-        Ok(epoch_no)
+            .group_id
+            .expect("group_id must be set before building");
+        Ok(group_id)
     }
     fn unwrap_unix_ts_ms(&self) -> Result<u64> {
         let unix_ts_ms = match self.base.unix_ts_ms {
@@ -366,14 +366,14 @@ where
         broadcast_type: BroadcastType,
         app_message: &[u8],
     ) -> Result<Bytes> {
-        let epoch_no = self.unwrap_epoch_no()?;
+        let group_id = self.unwrap_group_id()?;
         let unix_ts_ms = self.unwrap_unix_ts_ms()?;
 
         let header_buf = build_header(
             0, // version
             broadcast_type,
             merkle_tree_depth,
-            epoch_no,
+            group_id,
             unix_ts_ms,
             app_message,
         )?;

monad-raptorcast/src/packet/mod.rs

@@ -80,7 +80,7 @@ pub fn build_messages<ST>(
     segment_size: u16,
     app_message: Bytes,
     redundancy: Redundancy,
-    epoch_no: u64,
+    group_id: u64,
     unix_ts_ms: u64,
     build_target: BuildTarget<ST>,
     known_addresses: &HashMap<NodeId<CertificateSignaturePubKey<ST>>, SocketAddr>,
@@ -90,7 +90,7 @@ where
 {
     let builder = MessageBuilder::new(key, known_addresses)
         .segment_size(segment_size)
-        .epoch_no(epoch_no)
+        .group_id(group_id)
         .unix_ts_ms(unix_ts_ms)
         .redundancy(redundancy);