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 in chunk header. Secondary publishing
+    // embeds round 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 {
@@ -102,9 +106,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;
         }
 
@@ -1655,7 +1619,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,
                 secondary_broadcast: false,
             };

monad-raptorcast/src/lib.rs

@@ -567,7 +567,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");

monad-raptorcast/src/raptorcast_secondary/mod.rs

@@ -163,7 +163,7 @@ where
     }
 
     fn udp_build(
-        epoch: &Epoch,
+        group_id: u64,
         build_target: BuildTarget<ST>,
         outbound_message: Bytes,
         mtu: u16,
@@ -197,7 +197,7 @@ where
             segment_size,
             outbound_message,
             redundancy,
-            epoch.0,
+            group_id,
             unix_ts_ms,
             build_target,
             known_addresses,
@@ -230,7 +230,7 @@ where
             }
         };
         let udp_messages = Self::udp_build(
-            &self.curr_epoch,
+            self.curr_epoch.0,
             BuildTarget::<ST>::PointToPoint(dest_node),
             msg_bytes,
             self.mtu,
@@ -399,7 +399,11 @@ where
                     }
                 },
 
-                Self::Command::PublishToFullNodes { epoch, message } => {
+                Self::Command::PublishToFullNodes {
+                    epoch: _,
+                    round,
+                    message,
+                } => {
                     let _timer = DropTimer::start(Duration::from_millis(20), |elapsed| {
                         warn!(?elapsed, "long time to publish message")
                     });
@@ -451,7 +455,7 @@ where
                     // Split outbound_message into raptorcast chunks that we can
                     // send to full nodes.
                     let rc_chunks: UnicastMsg = Self::udp_build(
-                        &epoch,
+                        round.0,
                         build_target,
                         outbound_message,
                         self.mtu,

monad-raptorcast/src/udp.rs

@@ -189,8 +189,9 @@ impl<ST: CertificateSignatureRecoverable> UdpState<ST> {
             // chunks ASAP, before changing `recently_decoded_state`.
             if let Some(broadcast_mode) = maybe_broadcast_mode {
                 if !group_map.check_source(
-                    Epoch(parsed_message.epoch),
+                    parsed_message.group_id,
                     &parsed_message.author,
+                    &message.src_addr,
                     broadcast_mode,
                 ) {
                     continue;
@@ -221,7 +222,7 @@ impl<ST: CertificateSignatureRecoverable> UdpState<ST> {
                 if let Some(broadcast_mode) = maybe_broadcast_mode {
                     if self_hash == parsed_message.recipient_hash {
                         let maybe_targets = group_map.iterate_rebroadcast_peers(
-                            Epoch(parsed_message.epoch),
+                            parsed_message.group_id,
                             &parsed_message.author,
                             broadcast_mode,
                         );
@@ -237,9 +238,12 @@ impl<ST: CertificateSignatureRecoverable> UdpState<ST> {
                 }
             };
 
-            let validator_set = epoch_validators
-                .get(&Epoch(parsed_message.epoch))
-                .map(|ev| &ev.validators);
+            let validator_set = match maybe_broadcast_mode {
+                Some(BroadcastMode::Primary) | None => epoch_validators
+                    .get(&Epoch(parsed_message.group_id))
+                    .map(|ev| &ev.validators),
+                Some(BroadcastMode::Secondary) => None, // full-node raptorcast or point-to-point
+            };
 
             let decoding_context =
                 DecodingContext::new(validator_set, unix_ts_ms_now(), current_epoch);
@@ -360,7 +364,7 @@ pub fn build_messages<ST>(
     segment_size: u16, // Each chunk in the returned Vec (Bytes element of the tuple) will be limited to this size
     app_message: Bytes, // This is the actual message that gets raptor-10 encoded and split into UDP chunks
     redundancy: Redundancy,
-    epoch_no: u64,
+    group_id: u64,
     unix_ts_ms: u64,
     build_target: BuildTarget<ST>,
     known_addresses: &HashMap<NodeId<CertificateSignaturePubKey<ST>>, SocketAddr>,
@@ -376,7 +380,7 @@ where
         app_message,
         app_message_len,
         redundancy,
-        epoch_no,
+        group_id,
         unix_ts_ms,
         build_target,
         known_addresses,
@@ -431,7 +435,7 @@ pub fn build_messages_with_length<ST>(
     app_message: Bytes,
     app_message_len: u32,
     redundancy: Redundancy,
-    epoch_no: u64,
+    group_id: u64,
     unix_ts_ms: u64,
     build_target: BuildTarget<ST>,
     known_addresses: &HashMap<NodeId<CertificateSignaturePubKey<ST>>, SocketAddr>,
@@ -760,7 +764,6 @@ where
     // At this point, everything BELOW chunk_merkle_leaf_idx is populated
     // populate merkle trees/roots/leaf_idx + signatures (cached)
     let version: u16 = 0;
-    let epoch_no: u64 = epoch_no;
     let unix_ts_ms: u64 = unix_ts_ms;
     message
         // .par_chunks_mut(segment_size as usize * chunks_per_merkle_batch)
@@ -795,8 +798,9 @@ where
                 cursor_broadcast_merkle_depth[0] = ((is_raptor_broadcast as u8) << 7)
                     | ((is_secondary_raptor_broadcast as u8) << 6)
                     | (tree_depth & 0b0000_1111); // tree_depth max 4 bits
-                let (cursor_epoch_no, cursor) = cursor.split_at_mut(8);
-                cursor_epoch_no.copy_from_slice(&epoch_no.to_le_bytes());
+
+                let (cursor_group_id, cursor) = cursor.split_at_mut(8);
+                cursor_group_id.copy_from_slice(&group_id.to_le_bytes());
                 let (cursor_unix_ts_ms, cursor) = cursor.split_at_mut(8);
                 cursor_unix_ts_ms.copy_from_slice(&unix_ts_ms.to_le_bytes());
                 let (cursor_app_message_hash, cursor) = cursor.split_at_mut(20);
@@ -860,7 +864,10 @@ where
     // This applies to both validator-to-validator and validator-to-full-node
     // raptorcasting.
     pub author: NodeId<PT>,
-    pub epoch: u64,
+    // group_id is set to
+    // - epoch number for validator-to-validator raptorcast
+    // - round number for validator-to-fullnode raptorcast
+    pub group_id: u64,
     pub unix_ts_ms: u64,
     pub app_message_hash: AppMessageHash,
     pub app_message_len: u32,
@@ -951,8 +958,8 @@ where
         return Err(MessageValidationError::InvalidTreeDepth);
     }
 
-    let cursor_epoch = split_off(8)?;
-    let epoch = u64::from_le_bytes(cursor_epoch.as_ref().try_into().expect("u64 is 8 bytes"));
+    let cursor_group_id = split_off(8)?;
+    let group_id = u64::from_le_bytes(cursor_group_id.as_ref().try_into().expect("u64 is 8 bytes"));
 
     let cursor_unix_ts_ms = split_off(8)?;
     let unix_ts_ms = u64::from_le_bytes(
@@ -1047,7 +1054,7 @@ where
     Ok(ValidatedMessage {
         message,
         author,
-        epoch,
+        group_id,
         unix_ts_ms,
         app_message_hash,
         app_message_len,