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 group_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/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,
         )