[consensus] Add optimistic randomness share verification

config/src/config/consensus_config.rs

@@ -98,6 +98,7 @@ pub struct ConsensusConfig {
     pub enable_pre_commit: bool,
     pub max_pending_rounds_in_commit_vote_cache: u64,
     pub optimistic_sig_verification: bool,
+    pub optimistic_rand_share_verification: bool,
     pub enable_round_timeout_msg: bool,
     pub enable_optimistic_proposal_rx: bool,
     pub enable_optimistic_proposal_tx: bool,
@@ -383,6 +384,7 @@ impl Default for ConsensusConfig {
             enable_pre_commit: true,
             max_pending_rounds_in_commit_vote_cache: 100,
             optimistic_sig_verification: true,
+            optimistic_rand_share_verification: true,
             enable_round_timeout_msg: true,
             enable_optimistic_proposal_rx: true,
             enable_optimistic_proposal_tx: true,

consensus/src/epoch_manager.rs

@@ -1086,6 +1086,10 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
             })
             .collect::<Vec<_>>();
 
+        // Get aggregate public keys from the transcript for batch verification
+        let aggregate_pk_main = transcript.main.get_dealt_public_key();
+        let aggregate_pk_fast = transcript.fast.as_ref().map(|t| t.get_dealt_public_key());
+
         // Recover existing augmented key pair or generate a new one
         let (augmented_key_pair, fast_augmented_key_pair) = if let Some((_, key_pair)) = self
             .rand_storage
@@ -1133,6 +1137,8 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
             vuf_pp.clone(),
             keys,
             dkg_pub_params.pvss_config.wconfig.clone(),
+            aggregate_pk_main,
+            self.config.optimistic_rand_share_verification,
         );
 
         let fast_rand_config = if let (Some((ask, apk)), Some(trx), Some(wconfig)) = (
@@ -1154,6 +1160,8 @@ impl<P: OnChainConfigProvider> EpochManager<P> {
                 vuf_pp,
                 fast_keys,
                 fast_wconfig,
+                aggregate_pk_fast.expect("Fast PK must exist when fast path is enabled"),
+                self.config.optimistic_rand_share_verification,
             ))
         } else {
             None

consensus/src/rand/rand_gen/network_messages.rs

@@ -43,15 +43,15 @@ impl<S: TShare, D: TAugmentedData> RandMessage<S, D> {
         ensure!(self.epoch() == epoch_state.epoch);
         match self {
             RandMessage::RequestShare(_) => Ok(()),
-            RandMessage::Share(share) => share.verify(rand_config),
+            RandMessage::Share(share) => share.optimistic_verify(rand_config),
             RandMessage::AugData(aug_data) => {
                 aug_data.verify(rand_config, fast_rand_config, sender)
             },
             RandMessage::CertifiedAugData(certified_aug_data) => {
                 certified_aug_data.verify(&epoch_state.verifier)
             },
             RandMessage::FastShare(share) => {
-                share.share.verify(fast_rand_config.as_ref().ok_or_else(|| {
+                share.optimistic_verify(fast_rand_config.as_ref().ok_or_else(|| {
                     anyhow::anyhow!("[RandMessage] rand config for fast path not found")
                 })?)
             },

consensus/src/rand/rand_gen/rand_store.rs

@@ -39,14 +39,29 @@ impl<S: TShare> ShareAggregator<S> {
     }
 
     pub fn try_aggregate(
-        self,
+        mut self,
         rand_config: &RandConfig,
         rand_metadata: FullRandMetadata,
         decision_tx: Sender<Randomness>,
     ) -> Either<Self, RandShare<S>> {
         if self.total_weight < rand_config.threshold() {
             return Either::Left(self);
         }
+
+        // Pre-verify shares before spawning to ensure aggregation will succeed.
+        let bad_authors =
+            S::pre_aggregate_verify(self.shares.values(), rand_config, &rand_metadata.metadata);
+        for author in &bad_authors {
+            if self.shares.remove(author).is_some() {
+                self.total_weight = self
+                    .total_weight
+                    .saturating_sub(rand_config.get_peer_weight(author));
+            }
+        }
+        if self.total_weight < rand_config.threshold() {
+            return Either::Left(self);
+        }
+
         match self.path_type {
             PathType::Fast => {
                 observe_block(
@@ -440,6 +455,7 @@ mod tests {
                 .collect::<Vec<_>>();
             let vuf_pub_params = WvufPP::from(&dkg_pub_params.pvss_config.pp);
 
+            let aggregate_pk = transcript.main.get_dealt_public_key();
             let (ask, apk) = WVUF::augment_key_pair(&vuf_pub_params, sk.main, pk.main, &mut rng);
 
             let rand_keys = RandKeys::new(ask, apk, pk_shares, num_validators);
@@ -453,6 +469,8 @@ mod tests {
                 vuf_pub_params,
                 rand_keys,
                 weighted_config,
+                aggregate_pk,
+                false,
             );
 
             Self {

consensus/src/rand/rand_gen/reliable_broadcast_state.rs

@@ -136,7 +136,7 @@ impl<S: TShare, D: TAugmentedData> BroadcastStatus<RandMessage<S, D>, RandMessag
             self.rand_metadata,
             share.metadata()
         );
-        share.verify(&self.rand_config)?;
+        share.optimistic_verify(&self.rand_config)?;
         info!(LogSchema::new(LogEvent::ReceiveReactiveRandShare)
             .epoch(share.epoch())
             .round(share.metadata().round)