fix(consensus): mempool stats across subset change

because stats need data from previous round

Author: Mododo

collator/src/collator/anchors_cache.rs

@@ -2,7 +2,9 @@ use std::collections::VecDeque;
 use std::sync::Arc;
 
 use tycho_network::PeerId;
-use tycho_slasher_traits::{AnchorPeerStats, AnchorStats, AnchorStatsRange, CollatedAnchorStats};
+use tycho_slasher_traits::{
+    AnchorPeerStats, AnchorStats, AnchorStatsRange, CollatedAnchorStats, PeerIdInner,
+};
 use tycho_util::FastHashMap;
 
 use crate::collator::messages_reader::state::ext::ExternalKey;
@@ -320,7 +322,10 @@ impl<'a> AnchorsCacheTransaction<'a> {
         prev_imported_chain_time: u64,
         next_chain_time: u64,
     ) -> Option<CollatedAnchorStats> {
-        fn merge_anchor_stats(acc: &mut FastHashMap<u16, AnchorPeerStats>, stats: &AnchorStats) {
+        fn merge_anchor_stats(
+            acc: &mut FastHashMap<PeerIdInner, AnchorPeerStats>,
+            stats: &AnchorStats,
+        ) {
             for (validator_idx, peer_stats) in stats.0.iter() {
                 acc.entry(*validator_idx)
                     .and_modify(|exists| {
@@ -405,7 +410,7 @@ mod tests {
     fn make_anchor_with_stats(
         id: MempoolAnchorId,
         chain_time: u64,
-        validator_idx_stats: &[(u16, u16)],
+        peer_id_stats: &[(PeerIdInner, u16)],
     ) -> Arc<MempoolAnchor> {
         Arc::new(MempoolAnchor {
             id,
@@ -414,13 +419,13 @@ mod tests {
             author: PeerId([0; 32]),
             externals: Default::default(),
             stats: AnchorStats(Arc::new(
-                validator_idx_stats
+                peer_id_stats
                     .iter()
-                    .map(|(validator_idx, points_proven)| {
+                    .map(|(peer_id, points_proven)| {
                         let stats = AnchorPeerStats {
                             points_proven: *points_proven,
                         };
-                        (*validator_idx, stats)
+                        (*peer_id, stats)
                     })
                     .collect(),
             )),
@@ -449,10 +454,13 @@ mod tests {
     #[test]
     fn test_collect_imported_anchor_stats_uses_trimmed_interval() {
         let mut cache = AnchorsCache::default();
-        cache.add(make_anchor_with_stats(10, 100, &[(1, 1)]), 0);
-        cache.add(make_anchor_with_stats(11, 200, &[(1, 2), (2, 3)]), 0);
-        cache.add(make_anchor_with_stats(12, 300, &[(1, 4)]), 0);
-        cache.add(make_anchor_with_stats(13, 400, &[(2, 5)]), 0);
+        let p1 = [1; _];
+        let p2 = [2; _];
+
+        cache.add(make_anchor_with_stats(10, 100, &[(p1, 1)]), 0);
+        cache.add(make_anchor_with_stats(11, 200, &[(p1, 2), (p2, 3)]), 0);
+        cache.add(make_anchor_with_stats(12, 300, &[(p1, 4)]), 0);
+        cache.add(make_anchor_with_stats(13, 400, &[(p2, 5)]), 0);
 
         let mut tx = AnchorsCacheTransaction::new(&mut cache);
         tx.remove_last_imported_above(320);
@@ -464,8 +472,8 @@ mod tests {
             top_anchor: 12,
         });
         assert_eq!(collated_stats.stats.0.len(), 2);
-        assert_eq!(collated_stats.stats.0.get(&1).unwrap().points_proven, 6);
-        assert_eq!(collated_stats.stats.0.get(&2).unwrap().points_proven, 3);
+        assert_eq!(collated_stats.stats.0.get(&p1).unwrap().points_proven, 6);
+        assert_eq!(collated_stats.stats.0.get(&p2).unwrap().points_proven, 3);
     }
 
     #[test]

collator/src/validator/impls/std_impl/session.rs

@@ -264,7 +264,7 @@ impl ValidatorSession {
         );
 
         // Notify listeners about the own signature
-        events_scope.receive_signature(self.inner.own_validator_idx, true);
+        events_scope.receive_signature(self.inner.peer_id.as_bytes(), true);
 
         let mut total_weight = self.inner.own_weight;
 
@@ -430,7 +430,6 @@ impl ValidatorSession {
                     };
 
                     let validator_info = validators.get(peer_id).expect("peer info out of sync");
-                    let validator_idx = validator_info.validator_idx;
 
                     if !validator_info.public_key.verify_raw(&data, &signature) {
                         tracing::warn!(
@@ -448,14 +447,14 @@ impl ValidatorSession {
                         metrics::counter!(METRIC_INVALID_SIGNATURES_CACHED_TOTAL).increment(1);
 
                         if let Some(scope) = events_scope.upgrade() {
-                            scope.receive_signature(validator_idx, false);
+                            scope.receive_signature(peer_id.as_bytes(), false);
                         }
 
                         break 'stored Default::default();
                     }
 
                     if let Some(scope) = events_scope.upgrade() {
-                        scope.receive_signature(validator_idx, true);
+                        scope.receive_signature(peer_id.as_bytes(), true);
                     }
 
                     total_weight += validator_info.weight;
@@ -708,7 +707,7 @@ impl SessionState {
             metrics::counter!(METRIC_INVALID_SIGNATURES_IN_TOTAL).increment(1);
 
             if let Some(scope) = block.events_scope.upgrade() {
-                scope.receive_signature(validator_info.validator_idx, false);
+                scope.receive_signature(peer_id.as_bytes(), false);
             }
 
             return Err(ValidationError::InvalidSignature);
@@ -749,7 +748,7 @@ impl SessionState {
             && !was_sealed
             && let Some(scope) = block.events_scope.upgrade()
         {
-            scope.receive_signature(validator_info.validator_idx, true);
+            scope.receive_signature(peer_id.as_bytes(), true);
         }
 
         Ok(())

consensus/src/dag/commit/inspector.rs

@@ -9,7 +9,7 @@ use tycho_util::FastHashMap;
 use crate::dag::DagRound;
 use crate::effects::{AltFormat, TaskResult};
 use crate::engine::MempoolConfig;
-use crate::intercom::{PeerSchedule, SubsetPeer};
+use crate::intercom::PeerSchedule;
 use crate::models::{
     CounterU16, DagPoint, Digest, IllFormedPoint, InvalidPoint, MempoolPeerCounters,
     MempoolPeerStats, PointId, PointInfo, Round, TransInvalidPoint,
@@ -39,7 +39,7 @@ struct NotReferenced {
 }
 
 pub struct RoundInspector {
-    stats: FastHashMap<SubsetPeer, MempoolPeerStats>,
+    stats: FastHashMap<PeerId, MempoolPeerStats>,
     trans_invalid: Vec<TransInvalidPoint>,
     invalid: Vec<InvalidPoint>,
     ill_formed: Vec<IllFormedPoint>,
@@ -64,7 +64,7 @@ impl Default for RoundInspector {
 }
 
 impl RoundInspector {
-    pub fn take_stats(&mut self) -> FastHashMap<SubsetPeer, MempoolPeerStats> {
+    pub fn take_stats(&mut self) -> FastHashMap<PeerId, MempoolPeerStats> {
         let capacity = if self.stats.capacity() >= self.stats.len() * 4 {
             self.stats.len() * 2
         } else {
@@ -107,16 +107,13 @@ impl RoundInspector {
         peer_schedule: &PeerSchedule,
         conf: &MempoolConfig,
     ) -> TaskResult<()> {
-        let (prev_v_idxs, curr_v_idxs, emit_stats) = {
+        let emit_stats = {
             let guard = peer_schedule.atomic();
-            let prev = guard.validator_idx_map_for(r_0.round().prev()).clone();
-            let current = guard.validator_idx_map_for(r_0.round()).clone();
             // collator is run on another (sub)set than mempool, so slasher stats cannot be merged
             let no_stats_range = guard
                 .next_epoch_start
                 .map(|r| r - conf.consensus.max_consensus_lag_rounds.get()..r);
-            let emit_stats = no_stats_range.is_none_or(|r| !r.contains(&r_0.round()));
-            (prev, current, emit_stats)
+            no_stats_range.is_none_or(|r| !r.contains(&r_0.round()))
         };
 
         let leader_used = r_0
@@ -197,23 +194,18 @@ impl RoundInspector {
                 }
             }
 
-            if let Some(&validator_idx) = curr_v_idxs.get(author) {
-                match (self.stats)
-                    .entry(SubsetPeer {
-                        id: *author,
-                        validator_idx,
-                    })
-                    .or_insert_with(|| MempoolPeerStats::new(r_0.round().0))
-                    .add_in_order(&author_counters)
-                {
-                    Ok(()) => {}
-                    Err(err) => tracing::error!(
-                        author = display(author.alt()),
-                        error = display(err),
-                        "cannot report peer stats, skipping"
-                    ),
-                };
-            }
+            match (self.stats)
+                .entry(*author)
+                .or_insert_with(|| MempoolPeerStats::new(r_0.round().0))
+                .add_in_order(&author_counters)
+            {
+                Ok(()) => {}
+                Err(err) => tracing::error!(
+                    author = display(author.alt()),
+                    error = display(err),
+                    "cannot report peer stats, skipping"
+                ),
+            };
         }
 
         // check references against evidences
@@ -247,13 +239,7 @@ impl RoundInspector {
                     // will proceed with peers that made two consecutive points, forgive others
                     continue;
                 }
-                let Some(&validator_idx) = prev_v_idxs.get(&signer) else {
-                    continue;
-                };
-                let Some(signer_stats) = self.stats.get_mut(&SubsetPeer {
-                    id: signer,
-                    validator_idx,
-                }) else {
+                let Some(signer_stats) = self.stats.get_mut(&signer) else {
                     // this map contains full new v_set, but prev v_set is not guaranteed
                     continue;
                 };

consensus/src/dag/commit/mod.rs

@@ -5,6 +5,7 @@ mod inspector;
 use std::ops::RangeInclusive;
 use std::sync::atomic::Ordering;
 
+use tycho_network::PeerId;
 use tycho_util::FastHashMap;
 
 use crate::dag::DagRound;
@@ -13,7 +14,7 @@ use crate::dag::commit::back::DagBack;
 use crate::dag::commit::inspector::RoundInspector;
 use crate::effects::{AltFmt, AltFormat, Cancelled, TaskResult};
 use crate::engine::MempoolConfig;
-use crate::intercom::{PeerSchedule, SubsetPeer};
+use crate::intercom::PeerSchedule;
 use crate::models::{AnchorData, AnchorStageRole, MempoolPeerStats, Round};
 use crate::moderator::JournalEvent;
 
@@ -126,7 +127,7 @@ impl Committer {
         last_anchor: Round,
         peer_schedule: &PeerSchedule,
         conf: &MempoolConfig,
-    ) -> TaskResult<(FastHashMap<SubsetPeer, MempoolPeerStats>, Vec<JournalEvent>)> {
+    ) -> TaskResult<(FastHashMap<PeerId, MempoolPeerStats>, Vec<JournalEvent>)> {
         // in case previous anchor was triggered directly - rounds are already dropped
         let drained =
             (self.dag).drain_upto(last_anchor - conf.consensus.commit_history_rounds.get());

consensus/src/engine/committer_task.rs

@@ -18,7 +18,7 @@ use crate::engine::round_watch::{RoundWatch, TopKnownAnchor};
 use crate::engine::{
     ConsensusConfigExt, DAG_ROUNDS_TO_DROP, EngineResult, MempoolConfig, NodeConfig,
 };
-use crate::intercom::{PeerSchedule, SubsetPeer};
+use crate::intercom::PeerSchedule;
 use crate::models::{
     AnchorData, MempoolOutput, MempoolPeerStats, MempoolStatsMergeError, PointInfo, Round,
 };
@@ -187,28 +187,19 @@ impl State {
                         &inner.peer_schedule,
                         round_ctx.conf(),
                     )?;
-                    let mut stats_by_idx =
-                        FastHashMap::<u16, AnchorPeerStats>::with_capacity(stat_map.len());
-                    for (subset_peer, stats) in &stat_map {
+                    let mut anchor_stats = FastHashMap::with_capacity(stat_map.len());
+                    for (peer_id, stats) in &stat_map {
                         let Some(counters) = stats.counters() else {
                             continue;
                         };
-                        // zero counters not emitted to keep away `validator_idx` collisions
-                        // in slasher (receiver side) at vset switch boundary
                         if counters.points_proved.inner() == 0 {
                             continue;
                         }
-                        stats_by_idx
-                            .entry(subset_peer.validator_idx)
-                            .and_modify(|exist| {
-                                exist.points_proven = (exist.points_proven)
-                                    .saturating_add(counters.points_proved.inner());
-                            })
-                            .or_insert(AnchorPeerStats {
-                                points_proven: counters.points_proved.inner(),
-                            });
+                        anchor_stats.insert(peer_id.0, AnchorPeerStats {
+                            points_proven: counters.points_proved.inner(),
+                        });
                     }
-                    adata.stats.0 = Arc::new(stats_by_idx);
+                    adata.stats.0 = Arc::new(anchor_stats);
                     all_stats.push(stat_map);
 
                     round_ctx.commit_metrics(&adata.anchor);
@@ -279,14 +270,10 @@ impl RoundCtx {
         metrics::counter!("tycho_mempool_stats_merge_errors", "kind" => kind).increment(1);
     }
 
-    fn meter_stats(&self, all_stats: Vec<FastHashMap<SubsetPeer, MempoolPeerStats>>) {
-        let acc = match all_stats.first() {
-            None => return,
-            Some(m) => FastHashMap::<PeerId, _>::with_capacity(m.len() * 2),
-        };
-        let reduced = all_stats.into_iter().fold(acc, |mut acc, other_map| {
+    fn meter_stats(&self, all_stats: Vec<FastHashMap<PeerId, MempoolPeerStats>>) {
+        let reduced = all_stats.into_iter().reduce(|mut acc, other_map| {
             for (peer, stats) in other_map {
-                acc.entry(peer.id)
+                acc.entry(peer)
                     .and_modify(|occupied| self.merge_stats(occupied, &stats))
                     .or_insert(stats);
             }
@@ -297,6 +284,10 @@ impl RoundCtx {
             return; // reduce anyway to catch merge error
         }
 
+        let Some(reduced) = reduced else {
+            return;
+        };
+
         macro_rules! emit_counters {
             ($prefix:literal, $stats:expr, $labels:expr, [ $($field:ident),* $(,)? ]) => {
                 $(

consensus/src/intercom/peer_schedule/stateless.rs

@@ -71,24 +71,6 @@ impl PeerScheduleStateless {
         self.epoch_starts[2]
     }
 
-    pub fn validator_idx_map_for(&self, round: Round) -> &Arc<FastHashMap<PeerId, u16>> {
-        let result = if self.next_epoch_start.is_some_and(|r| round >= r) {
-            &self.validator_idx_maps[3]
-        } else if round >= self.epoch_starts[2] {
-            &self.validator_idx_maps[2]
-        } else if round >= self.epoch_starts[1] {
-            &self.validator_idx_maps[1]
-        } else if round >= self.epoch_starts[0] {
-            &self.validator_idx_maps[0]
-        } else {
-            &self.empty_validator_idx_map
-        };
-        if result.is_empty() {
-            tracing::error!("empty validator idx remap for {round:?}: {:?}", self.alt());
-        }
-        result
-    }
-
     pub fn peers_ordered_for(&self, round: Round) -> &Arc<Vec<SubsetPeer>> {
         let result = if self.next_epoch_start.is_some_and(|r| round >= r) {
             &self.peer_vecs[3]

slasher-traits/src/lib.rs

@@ -6,3 +6,5 @@ pub use self::validator::{
 
 mod mempool;
 mod validator;
+
+pub type PeerIdInner = [u8; 32];

slasher-traits/src/mempool.rs

@@ -2,6 +2,8 @@ use std::sync::Arc;
 
 use tycho_util::FastHashMap;
 
+use crate::PeerIdInner;
+
 #[derive(Debug, Clone)]
 pub struct CollatedAnchorStats {
     pub range: AnchorStatsRange,
@@ -23,7 +25,7 @@ pub struct AnchorStatsRange {
 }
 
 #[derive(Debug, Default, Clone)]
-pub struct AnchorStats(pub Arc<FastHashMap<u16, AnchorPeerStats>>);
+pub struct AnchorStats(pub Arc<FastHashMap<PeerIdInner, AnchorPeerStats>>);
 
 #[derive(Debug, Clone)]
 pub struct AnchorPeerStats {