feat(collator): mempool to slasher bridge

Author: Mododo

collator/src/collator/anchors_cache.rs

@@ -2,6 +2,8 @@ use std::collections::VecDeque;
 use std::sync::Arc;
 
 use tycho_network::PeerId;
+use tycho_slasher_traits::{AnchorPeerStats, AnchorStats, AnchorStatsRange, CollatedAnchorStats};
+use tycho_util::FastHashMap;
 
 use crate::collator::messages_reader::state::ext::ExternalKey;
 use crate::mempool::{MempoolAnchor, MempoolAnchorId};
@@ -15,6 +17,7 @@ pub struct AnchorInfo {
     #[allow(dead_code)]
     pub our_exts_count: usize,
     pub author: PeerId,
+    pub stats: AnchorStats,
 }
 
 impl AnchorInfo {
@@ -25,6 +28,7 @@ impl AnchorInfo {
             all_exts_count: anchor.externals.len(),
             our_exts_count,
             author: anchor.author,
+            stats: anchor.stats.clone(),
         }
     }
 }
@@ -311,6 +315,62 @@ impl<'a> AnchorsCacheTransaction<'a> {
         self.cache.last_imported_anchor_info()
     }
 
+    pub fn collect_imported_anchor_stats(
+        &self,
+        prev_imported_chain_time: u64,
+        next_chain_time: u64,
+    ) -> Option<CollatedAnchorStats> {
+        fn merge_anchor_stats(acc: &mut FastHashMap<u16, AnchorPeerStats>, stats: &AnchorStats) {
+            for (validator_idx, peer_stats) in stats.0.iter() {
+                acc.entry(*validator_idx)
+                    .and_modify(|exists| {
+                        exists.points_proven =
+                            (exists.points_proven).saturating_add(peer_stats.points_proven);
+                    })
+                    .or_insert(AnchorPeerStats {
+                        points_proven: peer_stats.points_proven,
+                    });
+            }
+        }
+
+        if prev_imported_chain_time >= next_chain_time {
+            return None;
+        }
+
+        let mut prev_top_anchor = None;
+        let mut top_anchor = None;
+        let mut acc = None;
+
+        for info in &self.cache.imported_anchors_info_history {
+            if info.ct <= prev_imported_chain_time {
+                prev_top_anchor = Some(info.id);
+                continue;
+            }
+            if info.ct > next_chain_time {
+                break;
+            }
+
+            top_anchor = Some(info.id);
+
+            match &mut acc {
+                None => acc = Some(info.stats.0.as_ref().clone()),
+                Some(acc) => merge_anchor_stats(acc, &info.stats),
+            };
+        }
+
+        if prev_top_anchor >= top_anchor {
+            return None;
+        }
+
+        Some(CollatedAnchorStats {
+            range: AnchorStatsRange {
+                prev_top_anchor: prev_top_anchor?,
+                top_anchor: top_anchor?,
+            },
+            stats: AnchorStats(Arc::new(acc?)),
+        })
+    }
+
     pub fn get(&self, index: usize) -> Option<(MempoolAnchorId, Arc<MempoolAnchor>)> {
         self.cache.get(index)
     }
@@ -334,16 +394,36 @@ impl Drop for AnchorsCacheTransaction<'_> {
 
 #[cfg(test)]
 mod tests {
+    use tycho_slasher_traits::AnchorPeerStats;
+
     use super::*;
 
     fn make_anchor(id: MempoolAnchorId, chain_time: u64) -> Arc<MempoolAnchor> {
+        make_anchor_with_stats(id, chain_time, &[])
+    }
+
+    fn make_anchor_with_stats(
+        id: MempoolAnchorId,
+        chain_time: u64,
+        validator_idx_stats: &[(u16, u16)],
+    ) -> Arc<MempoolAnchor> {
         Arc::new(MempoolAnchor {
             id,
             prev_id: if id > 0 { Some(id - 1) } else { None },
             chain_time,
             author: PeerId([0; 32]),
             externals: Default::default(),
-            stats: Default::default(),
+            stats: AnchorStats(Arc::new(
+                validator_idx_stats
+                    .iter()
+                    .map(|(validator_idx, points_proven)| {
+                        let stats = AnchorPeerStats {
+                            points_proven: *points_proven,
+                        };
+                        (*validator_idx, stats)
+                    })
+                    .collect(),
+            )),
         })
     }
 
@@ -366,6 +446,28 @@ mod tests {
         assert_eq!(cache.get(0).unwrap().0, 2);
     }
 
+    #[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 mut tx = AnchorsCacheTransaction::new(&mut cache);
+        tx.remove_last_imported_above(320);
+
+        let collated_stats = tx.collect_imported_anchor_stats(100, 320).unwrap();
+
+        assert_eq!(collated_stats.range, AnchorStatsRange {
+            prev_top_anchor: 10,
+            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);
+    }
+
     #[test]
     fn test_pop_front_rollback() {
         let mut cache = AnchorsCache::default();

collator/src/collator/do_collate/mod.rs

@@ -11,6 +11,7 @@ use tycho_block_util::queue::{QueueDiffStuff, QueueKey, SerializedQueueDiff};
 use tycho_block_util::state::{RefMcStateHandle, ShardStateStuff};
 use tycho_core::global_config::ZerostateId;
 use tycho_core::storage::{NewBlockMeta, StoreStateHint};
+use tycho_slasher_traits::CollatedAnchorStats;
 use tycho_types::models::*;
 use tycho_types::num::Tokens;
 use tycho_types::prelude::*;
@@ -71,6 +72,7 @@ pub struct FinalizeCollationCtx {
     /// Do we have unprocessed messages in internals
     /// or externals queue after block collation
     pub has_unprocessed_messages: bool,
+    pub collated_anchor_stats: Option<CollatedAnchorStats>,
     pub finalized: FinalizeBlockResult,
     pub reader_state: ReaderState,
     pub ref_mc_state_handle: RefMcStateHandle,
@@ -103,6 +105,7 @@ impl CollatorStdImpl {
             next_block_id_short,
             mc_data,
             collation_config,
+            prev_imported_chain_time,
             wu_used_from_last_anchor,
             resume_collation_elapsed,
             prev_shard_data,
@@ -171,6 +174,7 @@ impl CollatorStdImpl {
             collation_data,
             mc_data,
             prev_shard_data,
+            prev_imported_chain_time,
             shard_id: self.shard_id,
             collation_is_cancelled: CancellationFlag::new(),
             is_first_block_after_prev_master,
@@ -331,6 +335,7 @@ impl CollatorStdImpl {
             finalized,
             execute_result,
             final_result,
+            collated_anchor_stats,
         } = match do_collate_res {
             Err(CollatorError::Cancelled(reason)) => {
                 // cancel collation
@@ -373,6 +378,7 @@ impl CollatorStdImpl {
         } = self
             .finalize_collation(FinalizeCollationCtx {
                 has_unprocessed_messages: final_result.has_unprocessed_messages,
+                collated_anchor_stats,
                 finalized,
                 reader_state,
                 ref_mc_state_handle,
@@ -483,6 +489,25 @@ impl CollatorStdImpl {
         // from shard and should use anchors only up to chosen next chain time.
         anchors_cache.remove_last_imported_above(next_chain_time);
 
+        let collated_anchor_stats = anchors_cache
+            .collect_imported_anchor_stats(state.prev_imported_chain_time, next_chain_time);
+        if let Some(collated_anchor_stats) = &collated_anchor_stats {
+            tracing::debug!(target: tracing_targets::COLLATOR,
+                prev_imported_chain_time = state.prev_imported_chain_time,
+                next_chain_time,
+                prev_top_anchor = collated_anchor_stats.range.prev_top_anchor,
+                top_anchor = collated_anchor_stats.range.top_anchor,
+                validators = collated_anchor_stats.stats.0.len(),
+                "aggregated imported anchor stats for block",
+            );
+        } else if state.prev_imported_chain_time < next_chain_time {
+            tracing::warn!(target: tracing_targets::COLLATOR,
+                prev_imported_chain_time = state.prev_imported_chain_time,
+                next_chain_time,
+                "unable to aggregate imported anchor stats for block",
+            );
+        }
+
         // Verify that anchor author matches created_by set before collation
         let Some(anchor_info) = anchors_cache.last_imported_anchor_info() else {
             return Err(CollatorError::Anyhow(anyhow::anyhow!(
@@ -691,6 +716,7 @@ impl CollatorStdImpl {
                 finalized,
                 execute_result,
                 final_result,
+                collated_anchor_stats,
             },
             prepared_queue_diff.tx,
         ))
@@ -1128,6 +1154,7 @@ impl CollatorStdImpl {
 
         let FinalizeCollationCtx {
             has_unprocessed_messages,
+            collated_anchor_stats,
             finalized,
             reader_state,
             ref_mc_state_handle,
@@ -1214,6 +1241,7 @@ impl CollatorStdImpl {
                 .on_block_candidate(BlockCollationResult {
                     collation_session_id: self.collation_session.id(),
                     candidate: finalized.block_candidate,
+                    collated_anchor_stats,
                     prev_mc_block_id: finalized.old_mc_data.block_id,
                     mc_data: finalized.mc_data.clone(),
                     collation_config: collation_config.clone(),

collator/src/collator/do_collate/phase.rs

@@ -20,6 +20,7 @@ pub struct ActualState {
     pub collation_data: Box<BlockCollationData>,
     pub mc_data: Arc<McData>,
     pub prev_shard_data: PrevData,
+    pub prev_imported_chain_time: u64,
     pub shard_id: ShardIdent,
     /// For graceful collation cancellation
     pub collation_is_cancelled: CancellationFlag,

collator/src/collator/mod.rs

@@ -538,6 +538,7 @@ impl CollatorStdImpl {
             //      collator will import next anchor for sure
             //      because processed_to anchor will be equal to last inported
             author: PeerId(HashBytes::ZERO.0),
+            stats: Default::default(),
         });
     }
 
@@ -949,6 +950,7 @@ impl CollatorStdImpl {
                 next_block_id_short,
                 mc_data: new_mc_data,
                 collation_config,
+                prev_imported_chain_time: prev_shard_data.gen_chain_time(),
                 wu_used_from_last_anchor: prev_shard_data.wu_used_from_last_anchor(),
                 resume_collation_elapsed,
                 prev_shard_data: Some(prev_shard_data),
@@ -1051,6 +1053,7 @@ impl CollatorStdImpl {
         Ok(Box::new(WorkingState {
             next_block_id_short,
             mc_data,
+            prev_imported_chain_time: prev_shard_data.gen_chain_time(),
             wu_used_from_last_anchor: prev_shard_data.wu_used_from_last_anchor(),
             resume_collation_elapsed: Duration::ZERO,
             reader_state: ReaderState::new(prev_shard_data.processed_upto()),

collator/src/collator/tests/collator_tests.rs

@@ -509,6 +509,10 @@ fn test_get_anchors_processing_info() {
         prev_gen_chain_time,
     );
     assert_eq!(anchors_proc_info.last_imported_in_block_id, prev_block_id);
+    assert_eq!(
+        anchors_proc_info.current_shard_last_imported_chain_time,
+        prev_gen_chain_time,
+    );
 
     //======
     // collated master block 1:968, it used the same shard block 0:17
@@ -563,6 +567,10 @@ fn test_get_anchors_processing_info() {
         prev_gen_chain_time,
     );
     assert_eq!(anchors_proc_info.last_imported_in_block_id, prev_block_id);
+    assert_eq!(
+        anchors_proc_info.current_shard_last_imported_chain_time,
+        prev_gen_chain_time,
+    );
 
     //======
     // collated master block 1:1005, it used the same shard block 0:17
@@ -623,6 +631,10 @@ fn test_get_anchors_processing_info() {
         anchors_proc_info.last_imported_in_block_id,
         mc_data.block_id,
     );
+    assert_eq!(
+        anchors_proc_info.current_shard_last_imported_chain_time,
+        prev_gen_chain_time,
+    );
 
     //======
     // when previous shard state has no processed_to and master has one, take from mc
@@ -651,6 +663,10 @@ fn test_get_anchors_processing_info() {
         anchors_proc_info.last_imported_in_block_id,
         mc_data.block_id,
     );
+    assert_eq!(
+        anchors_proc_info.current_shard_last_imported_chain_time,
+        prev_gen_chain_time,
+    );
 
     //======
     // if previous block seqno is ahead from top shard seqno, info from mc must be ignored
@@ -685,6 +701,10 @@ fn test_get_anchors_processing_info() {
         anchors_proc_info.last_imported_in_block_id,
         prev_block_id_with_other_seqno,
     );
+    assert_eq!(
+        anchors_proc_info.current_shard_last_imported_chain_time,
+        prev_gen_chain_time,
+    );
 
     //======
     // for masterchain info is always taken from previous data
@@ -722,4 +742,8 @@ fn test_get_anchors_processing_info() {
         anchors_proc_info.last_imported_in_block_id,
         master_prev_block_id,
     );
+    assert_eq!(
+        anchors_proc_info.current_shard_last_imported_chain_time,
+        master_prev_gen_chain_time,
+    );
 }

collator/src/collator/types.rs

@@ -12,6 +12,7 @@ use tycho_block_util::queue::QueuePartitionIdx;
 use tycho_block_util::state::{RefMcStateHandle, ShardStateStuff};
 use tycho_core::global_config::MempoolGlobalConfig;
 use tycho_executor::{AccountMeta, PublicLibraryChange, TransactionMeta};
+use tycho_slasher_traits::CollatedAnchorStats;
 use tycho_types::boc;
 use tycho_types::cell::{Cell, CellFamily, HashBytes, Lazy, UsageTree, UsageTreeMode};
 use tycho_types::dict::{self, Dict};
@@ -42,6 +43,7 @@ pub(super) struct WorkingState {
     pub next_block_id_short: BlockIdShort,
     pub mc_data: Arc<McData>,
     pub collation_config: Arc<CollationConfig>,
+    pub prev_imported_chain_time: u64,
     pub wu_used_from_last_anchor: u64,
     pub resume_collation_elapsed: Duration,
     pub prev_shard_data: Option<PrevData>,
@@ -1254,6 +1256,7 @@ pub struct CollationResult {
     pub final_result: FinalResult,
     pub finalized: FinalizeBlockResult,
     pub execute_result: ExecuteResult,
+    pub collated_anchor_stats: Option<CollatedAnchorStats>,
 }
 
 pub struct FinalResult {

collator/src/manager/mod.rs

@@ -1100,10 +1100,12 @@ where
             } else {
                 let validator = self.validator.clone();
                 let validation_session_id = session_info.get_validation_session_id();
+                let collated_anchor_stats = collation_result.collated_anchor_stats.clone();
                 let dispatcher = self.dispatcher.clone();
                 tokio::spawn(async move {
-                    let validation_result =
-                        validator.validate(validation_session_id, &block_id).await;
+                    let validation_result = validator
+                        .validate(validation_session_id, &block_id, collated_anchor_stats)
+                        .await;
 
                     match validation_result {
                         Ok(status) => {