diff --git a/Cargo.lock b/Cargo.lock index 2e3f8cca1..9210a6fe5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4271,6 +4271,7 @@ dependencies = [ "tracing-subscriber", "tycho-crypto", "tycho-network", + "tycho-slasher-traits", "tycho-storage", "tycho-types", "tycho-util", diff --git a/cli/src/node/mod.rs b/cli/src/node/mod.rs index 6ba72c669..f949dfcc1 100644 --- a/cli/src/node/mod.rs +++ b/cli/src/node/mod.rs @@ -242,6 +242,7 @@ impl Node { self.slasher_config, &mc_state, ) + .await .context("failed to create slasher")?; let validator = ValidatorStdImpl::new( @@ -280,6 +281,7 @@ impl Node { CollatorStdImplFactory { wu_tuner_event_sender: Some(wu_tuner.event_sender.clone()), }, + slasher.validator_events_listener(), self.mempool_config_override.clone(), ); let collator_subcriber = CollatorStateSubscriber { diff --git a/collator/src/collator/anchors_cache.rs b/collator/src/collator/anchors_cache.rs index 38eadfd9e..992dec74b 100644 --- a/collator/src/collator/anchors_cache.rs +++ b/collator/src/collator/anchors_cache.rs @@ -343,6 +343,7 @@ mod tests { chain_time, author: PeerId([0; 32]), externals: Default::default(), + stats: None, }) } diff --git a/collator/src/collator/mod.rs b/collator/src/collator/mod.rs index 05e7ea63d..1163b677c 100644 --- a/collator/src/collator/mod.rs +++ b/collator/src/collator/mod.rs @@ -72,6 +72,7 @@ pub(super) mod tests; #[cfg(test)] pub(crate) use messages_reader::tests::{TestInternalMessage, TestMessageFactory}; +use tycho_slasher_traits::ValidatorEventsListener; // FACTORY @@ -82,6 +83,7 @@ pub struct CollatorContext { pub config: Arc, pub collation_session: Arc, pub zerostate_id: ZerostateId, + pub stats_recorder: Arc, pub shard_id: ShardIdent, pub prev_blocks_ids: Vec, @@ -165,6 +167,7 @@ pub struct CollatorStdImpl { collation_session: Arc, zerostate_id: ZerostateId, + stats_recorder: Arc, mq_adapter: Arc>, mpool_adapter: Arc, state_node_adapter: Arc, @@ -263,6 +266,7 @@ impl CollatorStdImpl { config, collation_session, zerostate_id, + stats_recorder, shard_id, prev_blocks_ids, mempool_config_override, @@ -285,6 +289,7 @@ impl CollatorStdImpl { config, collation_session, zerostate_id, + stats_recorder, mq_adapter, mpool_adapter, state_node_adapter, @@ -1564,6 +1569,20 @@ impl CollatorStdImpl { ) .record(elapsed_from_prev_anchor); + if let Some(stats) = &next_anchor.stats { + let (catchain_seqno, vset_switch_round) = + self.collation_session.get_validation_session_id(); + + self.stats_recorder.on_anchor_import( + tycho_slasher_traits::ValidationSessionId { + catchain_seqno, + vset_switch_round, + }, + &self.next_block_info, + stats.clone(), + ); + } + working_state.wu_used_from_last_anchor = 0; // time between anchors @@ -1770,6 +1789,20 @@ impl CollatorStdImpl { "imported next anchor, will notify collation manager", ); + if let Some(stats) = &next_anchor.stats { + let (catchain_seqno, vset_switch_round) = + self.collation_session.get_validation_session_id(); + + self.stats_recorder.on_anchor_import( + tycho_slasher_traits::ValidationSessionId { + catchain_seqno, + vset_switch_round, + }, + &self.next_block_info, + stats.clone(), + ); + } + // this may start master block collation or cause next anchor import let res = CollatorResult::skipped( working_state.mc_data.block_id, diff --git a/collator/src/collator/tests/messages_reader_tests.rs b/collator/src/collator/tests/messages_reader_tests.rs index ee8a7241e..c3b4376aa 100644 --- a/collator/src/collator/tests/messages_reader_tests.rs +++ b/collator/src/collator/tests/messages_reader_tests.rs @@ -1816,6 +1816,7 @@ where author: PeerId(Default::default()), chain_time: anchor_ct, externals, + stats: None, }); self.mempool.insert(anchor_id, anchor.clone()); diff --git a/collator/src/manager/mod.rs b/collator/src/manager/mod.rs index 88b59a690..c878f0403 100644 --- a/collator/src/manager/mod.rs +++ b/collator/src/manager/mod.rs @@ -7,6 +7,7 @@ use parking_lot::{Mutex, RwLock}; use tycho_block_util::state::ShardStateStuff; use tycho_core::global_config::MempoolGlobalConfig; use tycho_crypto::ed25519::KeyPair; +use tycho_slasher_traits::ValidatorEventsListener; use tycho_types::models::{BlockId, BlockIdShort, CollationConfig, ProcessedUptoInfo, ShardIdent}; use tycho_util::futures::JoinTask; use tycho_util::metrics::HistogramGuard; @@ -77,6 +78,7 @@ where validator: Arc, cancel_validation_runner: Mutex, + stats_recorder: Arc, active_collation_sessions: RwLock>>, active_collators: FastDashMap>>, @@ -136,6 +138,7 @@ where mpool_adapter_factory: MPF, validator: V, collator_factory: CF, + stats_recorder: Arc, mempool_config_override: Option, ) -> Arc> where @@ -170,6 +173,7 @@ where validator, cancel_validation_runner: Default::default(), + stats_recorder, active_collation_sessions: Default::default(), active_collators: Default::default(), diff --git a/collator/src/manager/state_update_handler.rs b/collator/src/manager/state_update_handler.rs index c6f46ece7..6bde490b9 100644 --- a/collator/src/manager/state_update_handler.rs +++ b/collator/src/manager/state_update_handler.rs @@ -336,6 +336,20 @@ where ); let next_block_id_short = calc_next_block_id_short(&prev_blocks_ids); + let add_validator_session = || { + // need to add validation session only for masterchain blocks, + // shard blocks are not being validated + if shard_id.is_masterchain() { + self.validator.add_session(AddSession { + shard_ident: shard_id, + session_id: new_session_info.get_validation_session_id(), + start_block_seqno: next_block_id_short.seqno, + vset_hash: raw_validators_set.repr_hash(), + validators: &new_session_info.collators().validators, + })?; + } + anyhow::Ok(()) + }; match self.active_collators.entry(shard_id) { DashMapEntry::Occupied(_) => { @@ -344,6 +358,7 @@ where "Active collator exists for collation session {:?}. Will resume it", new_session_info.id(), ); + add_validator_session()?; sessions_to_keep.push((shard_id, new_session_info.clone(), prev_blocks_ids)); } DashMapEntry::Vacant(entry) => { @@ -364,6 +379,7 @@ where config: self.config.clone(), collation_session: new_session_info.clone(), zerostate_id: *self.state_node_adapter.zerostate_id(), + stats_recorder: self.stats_recorder.clone(), shard_id, prev_blocks_ids: prev_blocks_ids.clone(), mempool_config_override: self.mempool_config_override.clone(), @@ -382,6 +398,7 @@ where } Ok(collator) => { let collator = Box::new(collator); + add_validator_session()?; // init collator let collator = @@ -409,17 +426,6 @@ where } } - // need to add validation session only for masterchain blocks, shard blocks are not being validated - if shard_id.is_masterchain() { - self.validator.add_session(AddSession { - shard_ident: shard_id, - session_id: new_session_info.get_validation_session_id(), - start_block_seqno: next_block_id_short.seqno, - vset_hash: raw_validators_set.repr_hash(), - validators: &new_session_info.collators().validators, - })?; - } - self.active_collation_sessions .write() .insert(shard_id, new_session_info); diff --git a/collator/src/manager/tests/manager_tests.rs b/collator/src/manager/tests/manager_tests.rs index 59b0f7a59..26b18254a 100644 --- a/collator/src/manager/tests/manager_tests.rs +++ b/collator/src/manager/tests/manager_tests.rs @@ -14,6 +14,7 @@ use tycho_block_util::state::{MinRefMcStateTracker, ShardStateStuff}; use tycho_core::global_config::ZerostateId; use tycho_core::storage::{BlockHandle, LoadStateHint, NewBlockMeta, StoreStateHint}; use tycho_crypto::ed25519::{KeyPair, SecretKey}; +use tycho_slasher_traits::NoopValidatorEventsRecorder; use tycho_types::boc::Boc; use tycho_types::cell::{Cell, CellBuilder, CellFamily, HashBytes, Lazy}; use tycho_types::merkle::MerkleUpdate; @@ -1914,6 +1915,7 @@ async fn test_should_not_sync_without_applied_range_after_known_sync() { }, validator: Arc::new(NoopValidator), cancel_validation_runner: Default::default(), + stats_recorder: Arc::new(NoopValidatorEventsRecorder), active_collation_sessions: Default::default(), active_collators: Default::default(), blocks_cache: BlocksCache::new(&Default::default()), @@ -3764,6 +3766,7 @@ async fn test_skipped_validation_does_not_commit_collated_master() { }, validator: Arc::new(NoopValidator), cancel_validation_runner: Default::default(), + stats_recorder: Arc::new(NoopValidatorEventsRecorder), active_collation_sessions: Default::default(), active_collators: Default::default(), blocks_cache: test_adapter.blocks_cache.clone(), @@ -3977,6 +3980,7 @@ async fn test_cache_gc_on_skipped_sync() { CollatorStdImplFactory { wu_tuner_event_sender: None, }, + Arc::new(NoopValidatorEventsRecorder), None, ); Arc::get_mut(&mut manager) diff --git a/collator/src/mempool/impls/common/cache.rs b/collator/src/mempool/impls/common/cache.rs index d8a7fdb13..6a2b8976e 100644 --- a/collator/src/mempool/impls/common/cache.rs +++ b/collator/src/mempool/impls/common/cache.rs @@ -442,6 +442,7 @@ mod tests { author: PeerId(Default::default()), chain_time: id as u64, externals: vec![], + stats: None, }) } diff --git a/collator/src/mempool/impls/common/shuttle.rs b/collator/src/mempool/impls/common/shuttle.rs index 91faa60e2..fae776a7a 100644 --- a/collator/src/mempool/impls/common/shuttle.rs +++ b/collator/src/mempool/impls/common/shuttle.rs @@ -56,6 +56,7 @@ impl Shuttle { chain_time, author: *committed.anchor.author(), externals: unique_messages, + stats: committed.stats.clone(), }); metrics::counter!("tycho_mempool_msgs_unique_count") diff --git a/collator/src/mempool/impls/dump_anchors.rs b/collator/src/mempool/impls/dump_anchors.rs index efaf2f518..0b2cb45a0 100644 --- a/collator/src/mempool/impls/dump_anchors.rs +++ b/collator/src/mempool/impls/dump_anchors.rs @@ -45,6 +45,7 @@ impl TryFrom for MempoolAnchor { author: value.author, chain_time: value.chain_time, externals, + stats: None, }) } } diff --git a/collator/src/mempool/impls/single_node_impl/anchor_handler.rs b/collator/src/mempool/impls/single_node_impl/anchor_handler.rs index 20af83b60..cc1ec41e8 100644 --- a/collator/src/mempool/impls/single_node_impl/anchor_handler.rs +++ b/collator/src/mempool/impls/single_node_impl/anchor_handler.rs @@ -60,6 +60,7 @@ impl SingleNodeAnchorHandler { chain_time, author: self.peer_id, externals: unique_messages, + stats: None, })) .expect("push new anchor"); diff --git a/collator/src/mempool/impls/stub_impl.rs b/collator/src/mempool/impls/stub_impl.rs index 46dea9df3..797968bee 100644 --- a/collator/src/mempool/impls/stub_impl.rs +++ b/collator/src/mempool/impls/stub_impl.rs @@ -443,6 +443,7 @@ pub(crate) fn make_empty_anchor( author: PeerId(Default::default()), chain_time, externals: vec![], + stats: None, }) } @@ -467,6 +468,7 @@ pub(crate) fn make_stub_anchor(id: MempoolAnchorId, prev_id: MempoolAnchorId) -> author: PeerId(Default::default()), chain_time, externals, + stats: None, } } @@ -530,6 +532,7 @@ pub(crate) fn make_anchor_from_file( author: PeerId(Default::default()), chain_time, externals, + stats: None, })) } diff --git a/collator/src/mempool/mod.rs b/collator/src/mempool/mod.rs index ae8fa230f..4375f0578 100644 --- a/collator/src/mempool/mod.rs +++ b/collator/src/mempool/mod.rs @@ -6,6 +6,7 @@ use anyhow::Result; use async_trait::async_trait; use bytes::Bytes; use tycho_network::PeerId; +use tycho_slasher_traits::AnchorStats; use tycho_types::models::*; use tycho_types::prelude::*; @@ -126,6 +127,7 @@ pub struct MempoolAnchor { pub author: PeerId, pub chain_time: u64, pub externals: Vec>, + pub stats: Option, } impl MempoolAnchor { diff --git a/collator/src/validator/impls/std_impl/mod.rs b/collator/src/validator/impls/std_impl/mod.rs index d32c4ad72..98be36aa7 100644 --- a/collator/src/validator/impls/std_impl/mod.rs +++ b/collator/src/validator/impls/std_impl/mod.rs @@ -77,7 +77,7 @@ impl ValidatorStdImpl { net_context: ValidatorNetworkContext, keypair: Arc, config: ValidatorStdImplConfig, - recorder: Arc, + stats_recorder: Arc, ) -> Self { Self { inner: Arc::new(Inner { @@ -85,7 +85,7 @@ impl ValidatorStdImpl { keypair, sessions: Default::default(), config, - events: ValidatorEvents::new(recorder), + events: ValidatorEvents::new(stats_recorder), }), } } diff --git a/collator/tests/collation_tests.rs b/collator/tests/collation_tests.rs index 7d4c5e0d9..b740d8efb 100644 --- a/collator/tests/collation_tests.rs +++ b/collator/tests/collation_tests.rs @@ -26,6 +26,7 @@ use tycho_core::global_config::ZerostateId; use tycho_core::node::NodeKeys; use tycho_core::storage::CoreStorage; use tycho_crypto::ed25519; +use tycho_slasher_traits::NoopValidatorEventsRecorder; use tycho_storage::StorageContext; use tycho_types::models::{BlockId, BlockIdShort, ShardIdent}; @@ -219,6 +220,8 @@ fn start_collation_manager( ctx: &DumpCollationContext, dumped_anchors: Vec, ) -> RunningCollationManager { + let recorder = Arc::new(NoopValidatorEventsRecorder); + CollationManager::create( ctx.keypair.clone(), ctx.config.clone(), @@ -240,6 +243,7 @@ fn start_collation_manager( CollatorStdImplFactory { wu_tuner_event_sender: None, }, + recorder, None, ) } diff --git a/consensus/Cargo.toml b/consensus/Cargo.toml index dbd4cf031..01a53c067 100644 --- a/consensus/Cargo.toml +++ b/consensus/Cargo.toml @@ -50,8 +50,9 @@ weedb = { workspace = true } # local deps tycho-network = { workspace = true } -tycho-util = { workspace = true } +tycho-slasher-traits = { workspace = true } tycho-storage = { workspace = true } +tycho-util = { workspace = true } [dev-dependencies] humantime = { workspace = true } diff --git a/consensus/src/dag/commit/mod.rs b/consensus/src/dag/commit/mod.rs index b5752de85..60bd0c1b5 100644 --- a/consensus/src/dag/commit/mod.rs +++ b/consensus/src/dag/commit/mod.rs @@ -300,6 +300,7 @@ impl Committer { proof_key: next.proof.key(), history: committed, is_executable, + stats: None, // may be set later }) } } diff --git a/consensus/src/engine/committer_task.rs b/consensus/src/engine/committer_task.rs index e6c13575a..9d3c9d09b 100644 --- a/consensus/src/engine/committer_task.rs +++ b/consensus/src/engine/committer_task.rs @@ -1,8 +1,10 @@ use std::mem; +use std::sync::Arc; use itertools::Itertools; use tokio::sync::mpsc; use tycho_network::PeerId; +use tycho_slasher_traits::{AnchorPeerStats, AnchorStats}; use tycho_util::FastHashMap; use tycho_util::metrics::HistogramGuard; @@ -165,7 +167,7 @@ impl State { let stats_ranges = inner.peer_schedule.atomic().stats_ranges(); - for adata in committed { + for mut adata in committed { let anchor_round = adata.anchor.round(); let (stat_map, events) = (inner.committer).remove_committed( @@ -173,6 +175,9 @@ impl State { &stats_ranges, round_ctx.conf(), )?; + if let Some(stats_slot_map) = stats_ranges.stats_slot_map(anchor_round) { + adata.stats = build_dense_anchor_stats(&stat_map, stats_slot_map); + } all_stats.push(stat_map); round_ctx.commit_metrics(&adata.anchor); @@ -196,6 +201,54 @@ impl State { } } +fn build_dense_anchor_stats( + stat_map: &FastHashMap, + stats_slot_map: &FastHashMap, +) -> Option { + if stats_slot_map.is_empty() { + return None; + } + + let mut dense = (0..stats_slot_map.len()) + .map(|_| AnchorPeerStats { points_proven: 0 }) + .collect::>(); + + #[allow(clippy::reversed_empty_ranges)] + let mut round_range = u32::MAX..=0; + let mut filled_rounds = 0; + + for (peer_id, stats) in stat_map { + let Some(counters) = stats.counters() else { + continue; + }; + + round_range = *round_range.start().min(&stats.first_round) + ..=*round_range.end().max(&counters.last_round); + + filled_rounds = filled_rounds.max(stats.filled_rounds()); + + if counters.points_proved.inner() == 0 { + continue; + } + let Some(slot_id) = stats_slot_map.get(peer_id) else { + continue; + }; + let Some(slot) = dense.get_mut(*slot_id) else { + continue; + }; + + slot.points_proven = slot + .points_proven + .saturating_add(counters.points_proved.inner()); + } + + (!round_range.is_empty()).then(|| AnchorStats { + round_range, + filled_rounds, + data: Arc::from(dense), + }) +} + impl RoundCtx { fn commit_metrics(&self, anchor: &PointInfo) { metrics::counter!("tycho_mempool_commit_anchors").increment(1); diff --git a/consensus/src/models/output.rs b/consensus/src/models/output.rs index ebcc40143..6e37047bf 100644 --- a/consensus/src/models/output.rs +++ b/consensus/src/models/output.rs @@ -1,3 +1,5 @@ +use tycho_slasher_traits::AnchorStats; + use crate::models::{PointInfo, PointKey, Round}; pub struct AnchorData { @@ -7,6 +9,7 @@ pub struct AnchorData { pub prev_anchor: Option, pub history: Vec, pub is_executable: bool, + pub stats: Option, } pub enum MempoolOutput { diff --git a/consensus/src/models/stats.rs b/consensus/src/models/stats.rs index 9b5aaf689..a33eef5f7 100644 --- a/consensus/src/models/stats.rs +++ b/consensus/src/models/stats.rs @@ -11,7 +11,7 @@ pub enum MempoolStatsMergeError { #[derive(Debug)] pub struct MempoolPeerStats { - first_round: u32, + pub first_round: u32, filled_rounds: u32, cumulative: MempoolPeerCounters, } diff --git a/consensus/src/storage/adapter_store.rs b/consensus/src/storage/adapter_store.rs index 39d639847..2080057d3 100644 --- a/consensus/src/storage/adapter_store.rs +++ b/consensus/src/storage/adapter_store.rs @@ -324,6 +324,7 @@ impl MempoolAdapterStore { .map(|r| r.prev()), history: keyed_vec.into_iter().map(|(_, info)| info).collect(), is_executable: false, // define later + stats: None, // not needed }); } Ok(result) diff --git a/contracts/src/slasher.tolk b/contracts/src/slasher.tolk index 6f5f29672..f9f3c64f0 100644 --- a/contracts/src/slasher.tolk +++ b/contracts/src/slasher.tolk @@ -195,6 +195,16 @@ fun validateBlocksBatch(batch: slice, params: ValidateBlocksBatchParams): int? { val nextSeqno = startSeqno + params.batchSize; + val start_anchor = batch.loadUint(32); + val end_anchor = batch.loadUint(32); + if (end_anchor < start_anchor) { + return null; + } + val filled_rounds = batch.loadUint(32); + if (filled_rounds > end_anchor - start_anchor + 1) { + return null; + } + batch.skipBits(params.batchSize); val history = batch.loadRef() as dict; if (!batch.isEmpty()) { @@ -211,7 +221,10 @@ fun validateBlocksBatch(batch: slice, params: ValidateBlocksBatchParams): int? { return null; } - val (csBits, csRefs) = cs!.remainingBitsAndRefsCount(); + var cs = cs!; + cs.skipBits(16); + + val (csBits, csRefs) = cs.remainingBitsAndRefsCount(); if (csBits != params.batchSize * 2 || csRefs != 0) { return null; } diff --git a/contracts/tests/Slasher.spec.ts b/contracts/tests/Slasher.spec.ts index bbb700885..6a29e017d 100644 --- a/contracts/tests/Slasher.spec.ts +++ b/contracts/tests/Slasher.spec.ts @@ -36,7 +36,7 @@ const SLASHER_ADDR = address( const BLOCKS_BATCH_SIZE = 10; const SAMPLE_BLOCKS_BATCH = Cell.fromBase64( - "te6ccgEBCAEAMAABCwAAAObYYAECAswFAgIBIAQDAAfRCgDAAAdpRQBgAgEgBwYAB2UFAGAAB/SKAMA=", + "te6ccgEBCAEARAABIwAAAOYAAAAAAAAAAAAAAADYYAECAswFAgIBIAQDAAvQAASKAMAAC2gAAIUAYAIBIAcGAAtkAAFFAGAAC/AAAgoAwA==", ); const FIRST_MC_SEQNO = 241; diff --git a/slasher-traits/src/lib.rs b/slasher-traits/src/lib.rs index 7effcbd34..02018e8c6 100644 --- a/slasher-traits/src/lib.rs +++ b/slasher-traits/src/lib.rs @@ -1,6 +1,8 @@ +pub use self::mempool::*; pub use self::validator::{ BlockValidationScope, NoopValidatorEventsRecorder, ReceivedSignature, ValidationSessionId, ValidatorEvents, ValidatorEventsListener, ValidatorSessionScope, }; +mod mempool; mod validator; diff --git a/slasher-traits/src/mempool.rs b/slasher-traits/src/mempool.rs new file mode 100644 index 000000000..818839a2e --- /dev/null +++ b/slasher-traits/src/mempool.rs @@ -0,0 +1,14 @@ +use std::ops::RangeInclusive; +use std::sync::Arc; + +#[derive(Debug, Clone)] +pub struct AnchorStats { + pub round_range: RangeInclusive, + pub filled_rounds: u32, + pub data: Arc<[AnchorPeerStats]>, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AnchorPeerStats { + pub points_proven: u16, +} diff --git a/slasher-traits/src/validator.rs b/slasher-traits/src/validator.rs index 5ba2284e8..f4b386175 100644 --- a/slasher-traits/src/validator.rs +++ b/slasher-traits/src/validator.rs @@ -4,9 +4,11 @@ use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use indexmap::IndexMap; use tycho_types::cell::HashBytes; -use tycho_types::models::{BlockId, IndexedValidatorDescription, McStateExtra}; +use tycho_types::models::{BlockId, BlockIdShort, IndexedValidatorDescription, McStateExtra}; use tycho_util::FastHasherState; +use crate::AnchorStats; + // TODO: Decide how to be with this collator-defined type #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct ValidationSessionId { @@ -81,8 +83,15 @@ impl ValidatorEvents { validators.len(), Default::default(), ); - for (i, validator) in validators.iter().enumerate() { - remap.insert(validator.validator_idx, i as u16); + + let mut validator_idxs = validators + .iter() + .map(|v| v.validator_idx) + .collect::>(); + validator_idxs.sort_unstable(); + + for (i, validator_idx) in validator_idxs.iter().enumerate() { + remap.insert(*validator_idx, i as u16); } ValidatorSessionScope { @@ -247,6 +256,14 @@ pub trait ValidatorEventsListener: Send + Sync + 'static { /// Called when the session is complete. fn on_session_finished(&self, session_id: ValidationSessionId); + /// Called when anchor is imported + fn on_anchor_import( + &self, + session_id: ValidationSessionId, + block_id: &BlockIdShort, + anchor_stats: AnchorStats, + ); + /// Called when validation is complete for a block. fn on_block_validated( &self, @@ -275,6 +292,14 @@ impl ValidatorEventsListener for NoopValidatorEventsRecorder { fn on_session_finished(&self, _session_id: ValidationSessionId) {} + fn on_anchor_import( + &self, + _session_id: ValidationSessionId, + _block_id: &BlockIdShort, + _anchor_stats: AnchorStats, + ) { + } + fn on_block_validated( &self, _session_id: ValidationSessionId, @@ -313,13 +338,31 @@ macro_rules! impl_recorder_for_tuples { $(self.$n.on_session_finished(session_id);)+ } + fn on_anchor_import(&self, + session_id: ValidationSessionId, + block_id: &BlockIdShort, + anchor_stats: AnchorStats, + ) { + impl_recorder_for_tuples!(@call_on_anchor_import self, + session_id, + block_id, + anchor_stats, + $($n)+ + ); + } + fn on_block_validated( &self, session_id: ValidationSessionId, block_id: &BlockId, signatures: Arc<[ReceivedSignature]>, ) { - impl_recorder_for_tuples!(@call_on_validated self, session_id, block_id, signatures, $($n)+); + impl_recorder_for_tuples!(@call_on_validated self, + session_id, + block_id, + signatures, + $($n)+ + ); } fn on_block_skipped(&self, session_id: ValidationSessionId, block_id: &BlockId) { @@ -328,10 +371,17 @@ macro_rules! impl_recorder_for_tuples { })* }; + (@call_on_anchor_import $self:ident, $sid:ident, $block_id:ident, $anchor_stats:ident, $n:tt $($rest:tt)+) => { + $self.$n.on_anchor_import($sid, $block_id, $anchor_stats.clone()); + impl_recorder_for_tuples!(@call_on_anchor_import $self, $sid, $block_id, $anchor_stats, $($rest)+) + }; (@call_on_validated $self:ident, $sid:ident, $block_id:ident, $signatures:ident, $n:tt $($rest:tt)+) => { $self.$n.on_block_validated($sid, $block_id, $signatures.clone()); impl_recorder_for_tuples!(@call_on_validated $self, $sid, $block_id, $signatures, $($rest)+) }; + (@call_on_anchor_import $self:ident, $sid:ident, $block_id:ident, $anchor_stats:ident, $n:tt) => { + $self.$n.on_anchor_import($sid, $block_id, $anchor_stats); + }; (@call_on_validated $self:ident, $sid:ident, $block_id:ident, $signatures:ident, $n:tt) => { $self.$n.on_block_validated($sid, $block_id, $signatures); }; diff --git a/slasher/src/analyzer.rs b/slasher/src/analyzer.rs index 3f28b5b1b..ed4663d88 100644 --- a/slasher/src/analyzer.rs +++ b/slasher/src/analyzer.rs @@ -56,9 +56,17 @@ pub fn analyze_vset( weight_per_block.resize(block_count, 0u64); let mut malformed = false; + let has_bad_filled_rounds = has_bad_filled_rounds(&batch); + if has_bad_filled_rounds { + tracing::warn!( + observer, + reason = "has_bad_filled_rounds", + "malformed batch" + ); + } // Count weight of valid signatures per committed column (block). - for history in &batch.signatures_history { + for history in &batch.observed { let other = history.validator_idx as usize; if other >= n { malformed = true; @@ -103,8 +111,12 @@ pub fn analyze_vset( } } + if !has_bad_filled_rounds { + observed[observer].filled_rounds += batch.filled_rounds as u64; + } + // Update scores. - for history in &batch.signatures_history { + for history in &batch.observed { let other = history.validator_idx as usize; if other >= n || other == observer { continue; @@ -119,81 +131,71 @@ pub fn analyze_vset( scores.invalid_signatures += history.bits.get(block * 2) as u64; scores.valid_signatures += history.bits.get(block * 2 + 1) as u64; } + + if !has_bad_filled_rounds && batch.filled_rounds > 0 { + scores[observer][other].points_proven += history.points_proven as u64; + } } // Update malformed - if malformed { + if malformed || has_bad_filled_rounds { observed[observer].malformed += 1; } } // Finally reduce scores and observations into accusations. - let mut accusation_weights = vec![0; n]; + let mut signature_accusation_weights = vec![0; n]; + let mut proven_accusation_weights = vec![0; n]; let mut rates = Vec::with_capacity(n - 1); - for (observer, (observed, scores)) in std::iter::zip(&observed, &scores).enumerate() { - if observed.samples < config.block_samples_threshold { - continue; + for observer in 0..n { + if observed[observer].samples >= config.block_samples_threshold.get() { + apply_signature_votes( + observer, + vset.list[observer].weight, + &vset.list[observer].public_key, + own_validator_idx, + observed[observer].samples, + &scores[observer], + config.slow_node_factor, + &mut rates, + &mut signature_accusation_weights, + ); } - let observer_weight = vset.list[observer].weight; - - // Compute the rate of valid signatures from other nodes. - rates.clear(); - rates.extend( - scores - .iter() - .enumerate() - .filter(|(i, _)| *i != observer) - .map(|(_, score)| score.valid_signatures as f64 / observed.samples as f64), - ); - rates.sort_by(|a, b| a.total_cmp(b)); - - let baseline = rates[rates.len() / 2]; - let slow_threshold = baseline * config.slow_node_factor; - tracing::debug!( - observer, - baseline, - slow_threshold, - "computed valid signature rates" - ); - - for (other, score) in scores.iter().enumerate() { - if other == observer { - continue; - } - - let rate = wilson_upper_bound(score.valid_signatures, observed.samples, Z_95); - if rate <= slow_threshold { - tracing::debug!(observer, other, "accusation found"); - if other == own_validator_idx { - tracing::warn!( - own_validator_idx, - observer, - "our node may be accused by {}", - vset.list[observer].public_key - ); - } - accusation_weights[other] += observer_weight; - } + if observed[observer].filled_rounds >= config.filled_rounds_threshold.get() { + // those with `has_bad_filled_rounds` have zero filled rounds and will be skipped too + apply_mempool_votes( + observer, + vset.list[observer].weight, + &vset.list[observer].public_key, + own_validator_idx, + observed[observer].filled_rounds, + &scores[observer], + config.proven_node_factor, + &mut rates, + &mut proven_accusation_weights, + ); } } tracing::debug!( %vset_hash, - ?accusation_weights, + ?signature_accusation_weights, + ?proven_accusation_weights, weight_threshold, "computed accusation weights" ); - let accusations = std::iter::zip(observed, accusation_weights) - .enumerate() - .filter_map(|(idx, (observed, weight))| { - let should_accuse = weight >= weight_threshold - || observed.malformed >= config.malformed_samples_threshold; + let accusations = (0..n) + .filter_map(|idx| { + let should_accuse = signature_accusation_weights[idx] >= weight_threshold + || proven_accusation_weights[idx] >= weight_threshold + || observed[idx].malformed >= config.malformed_samples_threshold; if should_accuse && idx == own_validator_idx { tracing::warn!( own_validator_idx, - weight, + signature_weight = signature_accusation_weights[idx], + proven_weight = proven_accusation_weights[idx], weight_threshold, "our node is considered bad" ); @@ -211,14 +213,146 @@ pub fn analyze_vset( struct Score { valid_signatures: u64, invalid_signatures: u64, + points_proven: u64, } #[derive(Default, Clone, Copy)] struct Observed { samples: u64, + filled_rounds: u64, malformed: u64, } +#[allow(clippy::too_many_arguments)] +fn apply_signature_votes( + observer: usize, + observer_weight: u64, + observer_public_key: &impl std::fmt::Display, + own_validator_idx: usize, + samples: u64, + scores: &[Score], + slow_node_factor: f64, + rates: &mut Vec, + accusation_weights: &mut [u64], +) { + // Compute the rate of valid signatures from other nodes. + rates.clear(); + rates.extend( + scores + .iter() + .enumerate() + .filter(|(i, _)| *i != observer) + .map(|(_, score)| score.valid_signatures as f64 / samples as f64), + ); + rates.sort_by(|a, b| a.total_cmp(b)); + + let baseline = rates[rates.len() / 2]; + let slow_threshold = baseline * slow_node_factor; + + tracing::debug!( + observer, + baseline, + slow_threshold, + "computed valid signature rates" + ); + + for (other, score) in scores.iter().enumerate() { + if other == observer { + continue; + } + + let rate = wilson_upper_bound(score.valid_signatures, samples, Z_95); + if rate <= slow_threshold { + if other == own_validator_idx { + tracing::warn!( + own_validator_idx, + observer, + reason = "slow_signatures", + "our node may be accused by {}", + observer_public_key + ); + } else { + tracing::debug!( + observer, + other, + reason = "slow_signatures", + "accusation found" + ); + } + accusation_weights[other] += observer_weight; + } + } +} + +#[allow(clippy::too_many_arguments)] +fn apply_mempool_votes( + observer: usize, + observer_weight: u64, + observer_public_key: &impl std::fmt::Display, + own_validator_idx: usize, + filled_rounds: u64, + scores: &[Score], + proven_node_factor: f64, + rates: &mut Vec, + accusation_weights: &mut [u64], +) { + rates.clear(); + rates.extend( + scores + .iter() + .enumerate() + .filter(|(target, _)| *target != observer) + .map(|(_, score)| score.points_proven as f64 / filled_rounds as f64), + ); + rates.sort_by(|a, b| a.total_cmp(b)); + + let baseline = rates[rates.len() / 2]; + let low_proven_threshold = baseline * proven_node_factor; + + tracing::debug!( + observer, + baseline, + low_proven_threshold, + "computed proven point rates" + ); + + for (other, score) in scores.iter().enumerate() { + if other == observer { + continue; + } + + let rate = wilson_upper_bound(score.points_proven, filled_rounds, Z_95); + if rate <= low_proven_threshold { + if other == own_validator_idx { + tracing::warn!( + own_validator_idx, + observer, + reason = "low_proven_points", + "our node may be accused by {}", + observer_public_key + ); + } else { + tracing::debug!( + observer, + other, + reason = "low_proven_points", + "accusation found" + ); + } + accusation_weights[other] += observer_weight; + } + } +} + +fn has_bad_filled_rounds(batch: &crate::BlocksBatch) -> bool { + let range_len = (batch.round_range.as_ref()) + .filter(|r| !r.is_empty()) + .map(|r| (*r.end() - *r.start()) as u64 + 1) + .unwrap_or_default(); + (batch.filled_rounds as u64 > range_len) + || (batch.observed.iter()).any(|history| history.points_proven as u32 > batch.filled_rounds) +} + // NOTE: We don't really need an exact determenism in decisions so we can // use floating point math here. If its a concert, this can be rewritten // to fixed point. @@ -241,3 +375,95 @@ fn wilson_upper_bound(hits: u64, samples: u64, z: f64) -> f64 { ((center + margin) / denom).clamp(0.0, 1.0) } + +#[cfg(test)] +mod tests { + use std::num::NonZeroU32; + + use super::*; + use crate::BlocksBatch; + + #[test] + fn proven_votes_use_observed_filled_rounds() { + let scores = vec![ + Score::default(), + Score { + points_proven: 100, + ..Default::default() + }, + Score { + points_proven: 30, + ..Default::default() + }, + ]; + let mut rates = Vec::new(); + let mut accusation_weights = vec![0; 3]; + + apply_mempool_votes( + 0, + 10, + &"observer", + usize::MAX, + 100, + &scores, + 0.67, + &mut rates, + &mut accusation_weights, + ); + + assert_eq!(accusation_weights, [0, 0, 10]); + } + + #[test] + fn signature_votes_still_use_observed_samples() { + let scores = vec![ + Score::default(), + Score { + valid_signatures: 100, + ..Default::default() + }, + Score { + valid_signatures: 30, + ..Default::default() + }, + ]; + let mut rates = Vec::new(); + let mut accusation_weights = vec![0; 3]; + + apply_signature_votes( + 0, + 10, + &"observer", + usize::MAX, + 100, + &scores, + 0.5, + &mut rates, + &mut accusation_weights, + ); + + assert_eq!(accusation_weights, [0, 0, 10]); + } + + #[test] + fn detects_impossible_proof_stats() { + let mut batch = BlocksBatch::new(0, NonZeroU32::new(1).unwrap(), &[0, 1]); + batch.round_range = Some(10..=10); + batch.filled_rounds = 1; + batch.observed[0].points_proven = 1; + batch.observed[1].points_proven = 0; + assert!(!has_bad_filled_rounds(&batch)); + + batch.filled_rounds = 2; + assert!(has_bad_filled_rounds(&batch)); + + batch.filled_rounds = 1; + batch.observed[0].points_proven = 2; + assert!(has_bad_filled_rounds(&batch)); + + batch.filled_rounds = 1; + batch.observed[0].points_proven = 1; + batch.round_range = None; + assert!(has_bad_filled_rounds(&batch)); + } +} diff --git a/slasher/src/bc/contract.rs b/slasher/src/bc/contract.rs index ca39cba48..9629d2ec9 100644 --- a/slasher/src/bc/contract.rs +++ b/slasher/src/bc/contract.rs @@ -9,7 +9,7 @@ use tycho_types::models::{ use tycho_types::prelude::*; use super::{ - BlocksBatch, SignatureHistory, SignedMessage, SlasherContract, SlasherContractEvent, + BlocksBatch, ObservedHistory, SignedMessage, SlasherContract, SlasherContractEvent, SlasherParams, SubmitBlocksBatch, }; use crate::util::BitSet; @@ -160,30 +160,34 @@ impl BlocksBatchBc { impl<'a> Load<'a> for BlocksBatchBc { fn load_from(slice: &mut CellSlice<'a>) -> Result { let start_seqno = slice.load_u32()?; + let round_range = + Some(slice.load_u32()?..=slice.load_u32()?).filter(|range| *range.end() > 0); + let filled_rounds = slice.load_u32()?; let block_count = slice.size_bits() as usize; let committed_blocks = BitSet::load_from_cs(block_count, slice)?; - let mut signatures_history = Vec::new(); + let mut observed = Vec::new(); let dict = Dict::>::from_raw(Some(slice.load_reference_cloned()?)); for entry in dict.iter() { let (validator_idx, mut cs) = entry?; - let bits = BitSet::load_from_cs(block_count * 2, &mut cs)?; + observed.push(ObservedHistory { + validator_idx, + points_proven: cs.load_u16()?, + bits: BitSet::load_from_cs(block_count * 2, &mut cs)?, + }); if !cs.is_empty() { return Err(tycho_types::error::Error::CellOverflow); } - - signatures_history.push(SignatureHistory { - validator_idx, - bits, - }); } Ok(Self(BlocksBatch { start_seqno, + round_range, + filled_rounds, committed_blocks, - signatures_history: signatures_history.into_boxed_slice(), + observed: observed.into_boxed_slice(), })) } } @@ -197,18 +201,31 @@ impl Store for BlocksBatchBc { let batch = &self.0; builder.store_u32(batch.start_seqno)?; + if let Some(anchor_range) = self.0.round_range.as_ref() { + builder.store_u32(*anchor_range.start())?; + builder.store_u32(*anchor_range.end())?; + } else { + builder.store_u32(0)?; + builder.store_u32(0)?; + }; + builder.store_u32(self.0.filled_rounds)?; + batch.committed_blocks.store_into(builder, context)?; // A subset contains items in no particular order, // so we need to sort by them to simplify remapping to vset. - let mut entries = batch - .signatures_history + debug_assert!( + batch.observed.is_sorted_by_key(|a| a.validator_idx), + "batch entries must be sorted by validator_idx" + ); + + let history = batch + .observed .iter() - .map(|item| (item.validator_idx, &item.bits)) + .map(|item| (item.validator_idx, (item.points_proven, &item.bits))) .collect::>(); - entries.sort_unstable_by_key(|(a, _)| *a); - let Some(dict_root) = dict::build_dict_from_sorted_iter(entries, context)? else { + let Some(dict_root) = dict::build_dict_from_sorted_iter(history, context)? else { // Subset must not be empty. return Err(tycho_types::error::Error::InvalidData); }; diff --git a/slasher/src/bc/mod.rs b/slasher/src/bc/mod.rs index 4ed0ff0c5..e8df27305 100644 --- a/slasher/src/bc/mod.rs +++ b/slasher/src/bc/mod.rs @@ -1,10 +1,11 @@ use std::num::NonZeroU32; +use std::ops::{Range, RangeInclusive}; use std::time::Duration; use anyhow::Result; use tokio::sync::oneshot; use tycho_crypto::ed25519; -use tycho_slasher_traits::{ReceivedSignature, ValidationSessionId}; +use tycho_slasher_traits::{AnchorStats, ReceivedSignature, ValidationSessionId}; use tycho_types::cell::{HashBytes, Lazy}; use tycho_types::models::{ BlockchainConfigParams, OwnedMessage, SignatureContext, StdAddr, Transaction, @@ -157,29 +158,38 @@ pub struct SubmitBlocksBatch { #[derive(Debug, PartialEq, Eq)] pub struct BlocksBatch { pub start_seqno: u32, + pub round_range: Option>, + pub filled_rounds: u32, pub committed_blocks: BitSet, - pub signatures_history: Box<[SignatureHistory]>, + /// sorted by `validator_idx` + pub observed: Box<[ObservedHistory]>, } impl BlocksBatch { pub fn new(start_seqno: u32, len: NonZeroU32, map_ids: &[u16]) -> Self { let len = len.get() as usize; + let mut observed = map_ids + .iter() + .map(|validator_idx| ObservedHistory { + validator_idx: *validator_idx, + points_proven: 0, + bits: BitSet::with_capacity(len * 2), + }) + .collect::>(); + observed.sort_unstable_by_key(|a| a.validator_idx); + Self { start_seqno, + round_range: None, + filled_rounds: 0, committed_blocks: BitSet::with_capacity(len), - signatures_history: map_ids - .iter() - .map(|validator_idx| SignatureHistory { - validator_idx: *validator_idx, - bits: BitSet::with_capacity(len * 2), - }) - .collect::>(), + observed: observed.into_boxed_slice(), } } pub fn is_empty(&self) -> bool { - self.committed_blocks.is_zero() + self.committed_blocks.is_zero() && self.round_range.is_none() } pub fn start_seqno(&self) -> u32 { @@ -191,6 +201,14 @@ impl BlocksBatch { .saturating_add(self.committed_blocks.len() as u32) } + pub fn seqno_range(&self) -> Range { + self.start_seqno..self.seqno_after() + } + + pub fn contains_seqno(&self, seqno: u32) -> bool { + self.seqno_range().contains(&seqno) + } + pub fn committed_block_count(&self) -> usize { (0..self.committed_blocks.len()) .filter(|offset| self.committed_blocks.get(*offset)) @@ -198,22 +216,60 @@ impl BlocksBatch { } pub fn validator_count(&self) -> usize { - self.signatures_history.len() + self.observed.len() } - pub fn contains_seqno(&self, seqno: u32) -> bool { - (self.start_seqno..self.seqno_after()).contains(&seqno) + pub fn push_anchor_stats(&mut self, anchor_stats: &AnchorStats) -> bool { + if anchor_stats.round_range.is_empty() + || anchor_stats.filled_rounds == 0 + || anchor_stats.data.len() != self.observed.len() + { + return false; + } + + let round_span = + (*anchor_stats.round_range.end() - *anchor_stats.round_range.start()) as u64 + 1; + if anchor_stats.filled_rounds as u64 > round_span + || (anchor_stats.data.iter()) + .any(|stats| stats.points_proven as u32 > anchor_stats.filled_rounds) + { + return false; + } + + if (self.round_range.as_ref()).is_some_and(|r| { + r.contains(anchor_stats.round_range.start()) + || r.contains(anchor_stats.round_range.end()) + || anchor_stats.round_range.contains(r.start()) + || anchor_stats.round_range.contains(r.end()) + }) { + return false; + } + + match &mut self.round_range { + None => self.round_range = Some(anchor_stats.round_range.clone()), + Some(exist) => { + *exist = *exist.start().min(anchor_stats.round_range.start()) + ..=*exist.end().max(anchor_stats.round_range.end()); + } + }; + self.filled_rounds = (self.filled_rounds).saturating_add(anchor_stats.filled_rounds); + + for (history, received) in std::iter::zip(&mut self.observed, &*anchor_stats.data) { + history.points_proven = (history.points_proven).saturating_add(received.points_proven); + } + + true } pub fn commit_signatures(&mut self, mut seqno: u32, signatures: &[ReceivedSignature]) -> bool { - if !self.contains_seqno(seqno) || signatures.len() != self.signatures_history.len() { + if !self.contains_seqno(seqno) || signatures.len() != self.observed.len() { return false; } seqno -= self.start_seqno; self.committed_blocks.set(seqno as usize, true); - for (history, received) in std::iter::zip(&mut self.signatures_history, signatures) { + for (history, received) in std::iter::zip(&mut self.observed, signatures) { let idx = (seqno as usize) * 2; history.bits.set(idx, received.has_invalid_signature()); history.bits.set(idx + 1, received.has_valid_signature()); @@ -224,7 +280,8 @@ impl BlocksBatch { } #[derive(Debug, PartialEq, Eq)] -pub struct SignatureHistory { +pub struct ObservedHistory { pub validator_idx: u16, + pub points_proven: u16, pub bits: BitSet, } diff --git a/slasher/src/collector/validator_events.rs b/slasher/src/collector/validator_events.rs index 4fa734507..7237d5339 100644 --- a/slasher/src/collector/validator_events.rs +++ b/slasher/src/collector/validator_events.rs @@ -7,8 +7,10 @@ use anyhow::Result; use tokio::sync::mpsc; use tracing::instrument; use tycho_crypto::ed25519; -use tycho_slasher_traits::{ReceivedSignature, ValidationSessionId, ValidatorEventsListener}; -use tycho_types::models::{BlockId, IndexedValidatorDescription}; +use tycho_slasher_traits::{ + AnchorStats, ReceivedSignature, ValidationSessionId, ValidatorEventsListener, +}; +use tycho_types::models::{BlockId, BlockIdShort, IndexedValidatorDescription}; use tycho_types::prelude::*; use tycho_util::{DashMapEntry, FastDashMap}; @@ -43,6 +45,8 @@ struct SessionState { /// Maps each subset item with its original vset index. validator_indices: Box<[u16]>, current_batch: BlocksBatch, + /// Imported anchors can arrive one batch window before block callbacks rotate the session. + next_batch: BlocksBatch, first_seqno: u32, next_expected_seqno: u32, complete_batches: Option>, @@ -99,14 +103,11 @@ impl ValidatorEventsCollector { .store(batch_size.get(), Ordering::Release); for mut session in self.sessions.iter_mut() { - // TODO: Split or grow the previous batch to not discard events. + // TODO: Split or grow the previous batches to not discard events. if session.batch_size != batch_size { session.batch_size = batch_size; - session.current_batch = BlocksBatch::new( - session.align_seqno(session.next_expected_seqno), - batch_size, - &session.validator_indices, - ); + let next_expected_seqno = session.next_expected_seqno; + session.reset_batches(next_expected_seqno); } } } @@ -121,15 +122,12 @@ impl ValidatorEventsCollector { return false; }; - // Reset the current batch if its size has changed. - // TODO: Split or grow the previous batch to not discard events. + // Reset the live batches if their size has changed. + // TODO: Split or grow the previous batches to not discard events. if session.batch_size != batch_size { session.batch_size = batch_size; - session.current_batch = BlocksBatch::new( - session.align_seqno(session.next_expected_seqno), - batch_size, - &session.validator_indices, - ); + let next_expected_seqno = session.next_expected_seqno; + session.reset_batches(next_expected_seqno); } session.complete_batches = Some(complete_batches); @@ -165,6 +163,8 @@ impl ValidatorEventsListener for ValidatorEventsCollector { let batch_size = NonZeroU32::new(self.default_batch_size.load(Ordering::Acquire)).unwrap(); let current_batch = BlocksBatch::new(first_mc_seqno, batch_size, &validator_indices); + let next_batch = + BlocksBatch::new(current_batch.seqno_after(), batch_size, &validator_indices); let validators = Arc::<[IndexedValidatorDescription]>::from(validators); @@ -173,6 +173,7 @@ impl ValidatorEventsListener for ValidatorEventsCollector { batch_size, validator_indices, current_batch, + next_batch, first_seqno: first_mc_seqno, next_expected_seqno: first_mc_seqno, // Will be initialized later via `init_session`. @@ -203,6 +204,29 @@ impl ValidatorEventsListener for ValidatorEventsCollector { metrics::gauge!(METRIC_ACTIVE_SESSIONS).set(self.sessions.len() as f64); } + #[instrument(skip_all, fields(session_id = ?session_id))] + fn on_anchor_import( + &self, + session_id: ValidationSessionId, + block_id: &BlockIdShort, + anchor_stats: AnchorStats, + ) { + if !block_id.is_masterchain() { + // Ignore for non-masterchain blocks (just in case). + return; + } + + tracing::debug!( + %block_id, + "on_anchor_import" + ); + let Some(mut session) = self.sessions.get_mut(&session_id) else { + tracing::warn!("session not found, ignoring on_anchor_import event"); + return; + }; + session.push_anchor_stats(block_id.seqno, &anchor_stats); + } + #[instrument(skip_all, fields(session_id = ?session_id))] fn on_block_validated( &self, @@ -258,11 +282,27 @@ impl ValidatorSessionInfo { // === Session state impl === impl SessionState { + fn push_anchor_stats(&mut self, seqno: u32, anchor_stats: &AnchorStats) -> bool { + if self.current_batch.contains_seqno(seqno) { + (self.current_batch).push_anchor_stats(anchor_stats) + } else if self.next_batch.contains_seqno(seqno) { + (self.next_batch).push_anchor_stats(anchor_stats) + } else { + tracing::warn!( + seqno, + current_batch_seqnos = ?self.current_batch.seqno_range(), + next_batch_seqnos = ?self.next_batch.seqno_range(), + "anchor import outside block batches" + ); + false + } + } + fn handle_block(&mut self, seqno: u32, signatures: Option<&[ReceivedSignature]>) -> bool { - let to_commit = match self.try_advance_current_batch(seqno) { + let batches = match self.try_advance_current_batch(seqno) { AdvanceBlockStatus::TooOld => return false, - AdvanceBlockStatus::Unchanged => None, - AdvanceBlockStatus::Replaced(batch) => Some(batch), + AdvanceBlockStatus::Unchanged => [None, None], + AdvanceBlockStatus::Rotated { first, second } => [Some(first), second], }; let event_type = match signatures { @@ -273,10 +313,10 @@ impl SessionState { None => "skipped", }; - if let Some(batch) = to_commit - && let Err(e) = self.commit_batch(batch) - { - tracing::error!(event_type, "failed to commit blocks batch: {e:?}"); + for (batch, ith) in batches.into_iter().flatten().zip(["1st", "2nd"]) { + if let Err(e) = self.commit_batch(batch) { + tracing::error!(event_type, "{ith} blocks batch failed to commit: {e:?}"); + } } true } @@ -288,14 +328,19 @@ impl SessionState { return AdvanceBlockStatus::Unchanged; } - let start_seqno = self.align_seqno(seqno); - let prev_batch = std::mem::replace( - &mut self.current_batch, - BlocksBatch::new(start_seqno, self.batch_size, &self.validator_indices), - ); self.next_expected_seqno = seqno + 1; - AdvanceBlockStatus::Replaced(prev_batch) + if self.next_batch.contains_seqno(seqno) { + let next = self.make_batch(self.next_batch.seqno_after()); + let current = std::mem::replace(&mut self.next_batch, next); + + AdvanceBlockStatus::Rotated { + first: std::mem::replace(&mut self.current_batch, current), + second: None, + } + } else { + self.reset_batches(seqno) + } } fn commit_batch(&self, batch: BlocksBatch) -> Result<()> { @@ -303,7 +348,8 @@ impl SessionState { } fn commit_final_batch(self) -> Result<()> { - Self::commit_batch_impl(&self.complete_batches, self.current_batch) + Self::commit_batch_impl(&self.complete_batches, self.current_batch)?; + Self::commit_batch_impl(&self.complete_batches, self.next_batch) } fn commit_batch_impl( @@ -325,6 +371,20 @@ impl SessionState { Ok(()) } + fn reset_batches(&mut self, seqno: u32) -> AdvanceBlockStatus { + let current = self.make_batch(self.align_seqno(seqno)); + let next = self.make_batch(current.seqno_after()); + + AdvanceBlockStatus::Rotated { + first: std::mem::replace(&mut self.current_batch, current), + second: Some(std::mem::replace(&mut self.next_batch, next)), + } + } + + fn make_batch(&self, start_seqno: u32) -> BlocksBatch { + BlocksBatch::new(start_seqno, self.batch_size, &self.validator_indices) + } + fn align_seqno(&self, seqno: u32) -> u32 { assert!(seqno >= self.first_seqno); @@ -340,5 +400,247 @@ impl SessionState { enum AdvanceBlockStatus { TooOld, Unchanged, - Replaced(BlocksBatch), + Rotated { + first: BlocksBatch, + second: Option, + }, +} + +#[cfg(test)] +mod tests { + use std::ops::RangeInclusive; + use std::sync::Arc; + + use tycho_slasher_traits::AnchorPeerStats; + use tycho_types::models::{ShardIdent, ValidatorDescription}; + + use super::*; + + const FIRST_SEQNO: u32 = 100; + const BATCH_SIZE: u32 = 3; + + #[test] + fn next_window_anchor_import_survives_rotation() { + let (collector, mut rx, session_id) = make_collector(); + + collector.on_anchor_import( + session_id, + &master_short(FIRST_SEQNO + BATCH_SIZE), + stats(10..=10, &[1, 0, 1]), + ); + assert!(drain_batches(&mut rx).is_empty()); + + collector.on_block_skipped(session_id, &master(FIRST_SEQNO + BATCH_SIZE)); + assert!(drain_batches(&mut rx).is_empty()); + + collector.on_session_finished(session_id); + let batches = drain_batches(&mut rx); + + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].start_seqno, FIRST_SEQNO + BATCH_SIZE); + assert_eq!(batches[0].round_range, Some(10..=10)); + assert_eq!(points(&batches[0]), vec![(0, 1), (1, 0), (2, 1)]); + } + + #[test] + fn final_flush_commits_current_and_next_anchor_batches() { + let (collector, mut rx, session_id) = make_collector(); + + collector.on_anchor_import( + session_id, + &master_short(FIRST_SEQNO), + stats(10..=10, &[1, 0, 1]), + ); + collector.on_anchor_import( + session_id, + &master_short(FIRST_SEQNO + BATCH_SIZE), + stats(11..=11, &[0, 1, 0]), + ); + + collector.on_session_finished(session_id); + let batches = drain_batches(&mut rx); + + assert_eq!(batches.len(), 2); + assert_eq!(batches[0].start_seqno, FIRST_SEQNO); + assert_eq!(batches[0].round_range, Some(10..=10)); + assert_eq!(points(&batches[0]), vec![(0, 1), (1, 0), (2, 1)]); + assert_eq!(batches[1].start_seqno, FIRST_SEQNO + BATCH_SIZE); + assert_eq!(batches[1].round_range, Some(11..=11)); + assert_eq!(points(&batches[1]), vec![(0, 0), (1, 1), (2, 0)]); + } + + #[test] + fn far_jump_flushes_old_next_batch_before_reset() { + let (collector, mut rx, session_id) = make_collector(); + + collector.on_anchor_import( + session_id, + &master_short(FIRST_SEQNO + BATCH_SIZE), + stats(10..=10, &[1, 0, 1]), + ); + + collector.on_block_skipped(session_id, &master(FIRST_SEQNO + 3 * BATCH_SIZE)); + let batches = drain_batches(&mut rx); + + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].start_seqno, FIRST_SEQNO + BATCH_SIZE); + assert_eq!(batches[0].round_range, Some(10..=10)); + assert_eq!(points(&batches[0]), vec![(0, 1), (1, 0), (2, 1)]); + } + + #[test] + fn duplicate_anchor_import_is_not_counted_twice() { + let (collector, mut rx, session_id) = make_collector(); + + collector.on_anchor_import( + session_id, + &master_short(FIRST_SEQNO), + stats(10..=10, &[1, 0, 1]), + ); + collector.on_anchor_import( + session_id, + &master_short(FIRST_SEQNO), + stats(10..=10, &[0, 1, 0]), + ); + + collector.on_session_finished(session_id); + let batches = drain_batches(&mut rx); + + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].round_range, Some(10..=10)); + assert_eq!(points(&batches[0]), vec![(0, 1), (1, 0), (2, 1)]); + } + + #[test] + fn anchor_import_outside_live_windows_is_rejected() { + let (collector, mut rx, session_id) = make_collector(); + + collector.on_anchor_import( + session_id, + &master_short(FIRST_SEQNO + 2 * BATCH_SIZE), + stats(10..=10, &[1, 0, 1]), + ); + + collector.on_session_finished(session_id); + + assert!(drain_batches(&mut rx).is_empty()); + } + + #[test] + fn batch_size_update_resets_lookahead_batch() { + let (collector, mut rx, session_id) = make_collector(); + + collector.on_anchor_import( + session_id, + &master_short(FIRST_SEQNO + BATCH_SIZE), + stats(10..=10, &[1, 0, 1]), + ); + assert!(drain_batches(&mut rx).is_empty()); + + let new_batch_size = BATCH_SIZE + 1; + collector.set_default_batch_size(NonZeroU32::new(new_batch_size).unwrap()); + + collector.on_anchor_import( + session_id, + &master_short(FIRST_SEQNO + new_batch_size), + stats(11..=11, &[0, 1, 0]), + ); + + collector.on_session_finished(session_id); + let batches = drain_batches(&mut rx); + + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].start_seqno, FIRST_SEQNO + new_batch_size); + assert_eq!(batches[0].committed_blocks.len(), new_batch_size as usize); + assert_eq!(batches[0].round_range, Some(11..=11)); + assert_eq!(points(&batches[0]), vec![(0, 0), (1, 1), (2, 0)]); + } + + fn master(seqno: u32) -> BlockId { + BlockId { + shard: ShardIdent::MASTERCHAIN, + seqno, + root_hash: HashBytes::ZERO, + file_hash: HashBytes::ZERO, + } + } + + fn master_short(seqno: u32) -> BlockIdShort { + BlockIdShort { + shard: ShardIdent::MASTERCHAIN, + seqno, + } + } + + fn stats(round_range: RangeInclusive, points: &[u16]) -> AnchorStats { + let filled_rounds = *round_range.end() - *round_range.start() + 1; + assert!( + (points.iter()).all(|points_proven| u32::from(*points_proven) <= filled_rounds), + "test stats must keep points_proven <= filled_rounds" + ); + + AnchorStats { + filled_rounds, + round_range, + data: Arc::from( + points + .iter() + .copied() + .map(|points_proven| AnchorPeerStats { points_proven }) + .collect::>(), + ), + } + } + + fn drain_batches(rx: &mut BlocksBatchRx) -> Vec { + let mut result = Vec::new(); + while let Ok(batch) = rx.try_recv() { + result.push(batch); + } + result + } + + fn points(batch: &BlocksBatch) -> Vec<(u16, u16)> { + (batch.observed) + .iter() + .map(|item| (item.validator_idx, item.points_proven)) + .collect() + } + + fn make_collector() -> (ValidatorEventsCollector, BlocksBatchRx, ValidationSessionId) { + let collector = ValidatorEventsCollector::new(NonZeroU32::new(BATCH_SIZE).unwrap()); + let session_id = ValidationSessionId { + catchain_seqno: 1, + vset_switch_round: 2, + }; + let validators = make_validators(); + + collector.on_session_started(session_id, FIRST_SEQNO, 0, &HashBytes::ZERO, &validators); + + let (tx, rx) = mpsc::unbounded_channel(); + assert!(collector.init_session(session_id, NonZeroU32::new(BATCH_SIZE).unwrap(), tx,)); + + (collector, rx, session_id) + } + + fn make_validators() -> Vec { + [2, 0, 1] + .into_iter() + .scan(0, |prev_total_weight, validator_idx| { + let desc = ValidatorDescription { + public_key: HashBytes([validator_idx as u8 + 1; 32]), + weight: 1, + adnl_addr: None, + mc_seqno_since: FIRST_SEQNO, + prev_total_weight: *prev_total_weight, + }; + *prev_total_weight += desc.weight; + + Some(IndexedValidatorDescription { + desc, + validator_idx, + }) + }) + .collect() + } } diff --git a/slasher/src/lib.rs b/slasher/src/lib.rs index 26beb3336..48a036642 100644 --- a/slasher/src/lib.rs +++ b/slasher/src/lib.rs @@ -1,3 +1,4 @@ +use std::num::NonZeroU64; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -24,8 +25,8 @@ use tycho_util::serde_helpers; use self::bc::SlasherParams; pub use self::bc::{ - BlocksBatch, ContractSubscription, EncodeBlocksBatchMessage, MessageDelivered, - SignatureHistory, SignedMessage, SlasherContract, StdSlasherContract, + BlocksBatch, ContractSubscription, EncodeBlocksBatchMessage, MessageDelivered, SignedMessage, + SlasherContract, StdSlasherContract, }; use self::collector::{ValidatorEventsCollector, ValidatorSessionInfo}; use self::storage::SlasherStorage; @@ -82,10 +83,15 @@ pub struct SlasherConfig { /// Default: `1000` pub vset_len_threshold: u32, - // At least this number of block samples must be collected to accuse someone. - // - // Default: `100` - pub block_samples_threshold: u64, + /// At least this number of block samples must be collected to accuse someone. + /// + /// Default: `100` + pub block_samples_threshold: NonZeroU64, + + /// At least this number of rounds must be filled to accuse someone. + /// + /// Default: `1000` + pub filled_rounds_threshold: NonZeroU64, /// At least this number of malformed batches must be collected to accuse someone. /// @@ -96,6 +102,12 @@ pub struct SlasherConfig { /// /// Default: `0.5` pub slow_node_factor: f64, + + /// We treat the node as low-proven if its proven-point rate is this times + /// the median rate. + /// + /// Default: `0.67` + pub proven_node_factor: f64, } impl Default for SlasherConfig { @@ -105,9 +117,11 @@ impl Default for SlasherConfig { message_retry_interval: Duration::from_secs(1), prev_delivery_timeout: Some(Duration::from_secs(5)), vset_len_threshold: 1000, - block_samples_threshold: 100, + block_samples_threshold: 100.try_into().unwrap(), + filled_rounds_threshold: 1000.try_into().unwrap(), malformed_samples_threshold: 5, slow_node_factor: 0.5, + proven_node_factor: 0.67, } } } @@ -119,7 +133,7 @@ pub struct Slasher { } impl Slasher { - pub fn new( + pub async fn new( node_keys: Arc, contract: C, blockchain_rpc_client: BlockchainRpcClient, @@ -138,8 +152,9 @@ impl Slasher { let known_session_id = tycho_slasher_traits::ValidationSessionId::from(state_extra); let blockchain_config = &state_extra.config; - let storage = - SlasherStorage::open(storage_context).context("failed to open slasher storage")?; + let storage = SlasherStorage::open(storage_context) + .await + .context("failed to open slasher storage")?; let current_vset = Arc::new(ParsedVset::from_raw( blockchain_config.get_current_validator_set_raw()?, diff --git a/slasher/src/proto.tl b/slasher/src/proto.tl index 97caf47d2..a0956de10 100644 --- a/slasher/src/proto.tl +++ b/slasher/src/proto.tl @@ -7,18 +7,23 @@ */ slasher.blocksBatch start_seqno:int + min_anchor:int + max_anchor:int + filled_rounds:int committed_blocks:bitset - entries:(vector slasher.signatureHistory) + observed:(vector slasher.observedHistory) = slasher.BlocksBatch; /** * @param validator_idx validator index relative to the validator set +* @param points_proven amount of points with proofs at rounds DAG closed in a given anchor range * @param bits history bits (2 for each block) */ -slasher.signatureHistory +slasher.observedHistory validator_idx:int + points_proven:int bits:bitset - = slasher.SignatureHistory; + = slasher.ObservedHistory; slasher.storedVsetInfo prev_vset_hash:int256 diff --git a/slasher/src/storage/db.rs b/slasher/src/storage/db.rs index a68c22065..3adad7c92 100644 --- a/slasher/src/storage/db.rs +++ b/slasher/src/storage/db.rs @@ -1,8 +1,10 @@ +use futures_util::TryFutureExt; use tycho_storage::kv::{ Migrations, NamedTables, StateVersionProvider, TableContext, WithMigrations, }; use tycho_util::sync::CancellationFlag; -use weedb::{MigrationError, Semver, WeeDb}; +use weedb::rocksdb::{WaitForCompactOptions, WriteBatch}; +use weedb::{BoundedCfHandle, MigrationError, Semver, VersionProvider, WeeDb}; pub type SlasherDb = WeeDb; @@ -11,7 +13,7 @@ impl NamedTables for SlasherTables { } impl WithMigrations for SlasherTables { - const VERSION: Semver = [0, 1, 0]; + const VERSION: Semver = [0, 2, 0]; type VersionProvider = StateVersionProvider; @@ -20,9 +22,10 @@ impl WithMigrations for SlasherTables { } fn register_migrations( - _migrations: &mut Migrations, + migrations: &mut Migrations, _cancelled: CancellationFlag, ) -> Result<(), MigrationError> { + migrations.register([0, 1, 0], Self::VERSION, wipe_db)?; Ok(()) } } @@ -121,3 +124,83 @@ pub mod tables { } } } + +pub(super) trait SlasherDbExt { + fn normalize_version(&self) -> impl Future>; +} + +impl SlasherDbExt for SlasherDb { + fn normalize_version(&self) -> impl Future> { + let db = self.clone(); + tokio::task::spawn_blocking(move || normalize_version(&db)) + .unwrap_or_else(|e| Err(anyhow::anyhow!("normalize slasher db {e}"))) + } +} + +fn normalize_version(db: &SlasherDb) -> anyhow::Result<()> { + fn is_table_empty(table: &weedb::Table) -> anyhow::Result + where + T: weedb::ColumnFamily, + { + let mut iterator = table.raw_iterator(); + iterator.seek_to_first(); + iterator.status()?; + Ok(iterator.item().is_none()) + } + + let provider = SlasherTables::new_version_provider(); + + if provider.get_version(db.raw())?.is_some() { + return Ok(()); + } + + if is_table_empty(&db.vset_state)? + && is_table_empty(&db.vset_sessions)? + && is_table_empty(&db.block_batches)? + { + return Ok(()); + } + + tracing::warn!("normalizing DB version for slasher"); + provider.set_version(db.raw(), [0, 1, 0])?; + Ok(()) +} + +fn wipe_db(db: &SlasherDb) -> Result<(), MigrationError> { + tracing::warn!("wiping slasher DB for incompatible storage format change"); + + let mut batch = WriteBatch::default(); + + batch.delete_range_cf( + &db.vset_state.cf(), + [0; tables::VsetState::KEY_LEN], + [u8::MAX; tables::VsetState::KEY_LEN], + ); + + batch.delete_range_cf( + &db.vset_sessions.cf(), + [0; tables::VsetSessions::KEY_LEN], + [u8::MAX; tables::VsetSessions::KEY_LEN], + ); + batch.delete_range_cf( + &db.block_batches.cf(), + [0; tables::BlockBatches::KEY_LEN], + [u8::MAX; tables::BlockBatches::KEY_LEN], + ); + + let compact = |cf: &BoundedCfHandle<'_>| { + db.rocksdb() + .compact_range_cf::<&[u8], &[u8]>(cf, None, None); + }; + compact(&db.vset_state.cf()); + compact(&db.vset_sessions.cf()); + compact(&db.block_batches.cf()); + + db.rocksdb().write(batch)?; + + let mut opts = WaitForCompactOptions::default(); + opts.set_flush(true); + db.rocksdb().wait_for_compact(&opts)?; + + Ok(()) +} diff --git a/slasher/src/storage/mod.rs b/slasher/src/storage/mod.rs index f2fb092ff..44a617a72 100644 --- a/slasher/src/storage/mod.rs +++ b/slasher/src/storage/mod.rs @@ -3,12 +3,14 @@ use std::sync::Arc; use anyhow::Result; use tycho_slasher_traits::ValidationSessionId; use tycho_storage::StorageContext; +use tycho_storage::kv::ApplyMigrations; use tycho_types::cell::HashBytes; -use weedb::{OwnedSnapshot, rocksdb}; +use weedb::{OwnedSnapshot, WeeDb, rocksdb}; -use self::db::{SlasherDb, tables}; +use self::db::{SlasherDb, SlasherDbExt, tables}; use self::models::{StoredBlocksBatch, StoredVsetInfo, StoredVsetReport}; use crate::BlocksBatch; +use crate::storage::db::SlasherTables; pub mod db; pub mod models; @@ -22,8 +24,11 @@ pub struct SlasherStorage { } impl SlasherStorage { - pub fn open(ctx: &StorageContext) -> Result { - let db = ctx.open_preconfigured(SLASHER_DB_SUBDIR)?; + pub async fn open(ctx: &StorageContext) -> Result { + let db: WeeDb = ctx.open_preconfigured(SLASHER_DB_SUBDIR)?; + + db.normalize_version().await?; + db.apply_migrations().await?; Ok(Self { inner: Arc::new(Inner { db }), diff --git a/slasher/src/storage/models.rs b/slasher/src/storage/models.rs index 191a94b87..0da8a97c4 100644 --- a/slasher/src/storage/models.rs +++ b/slasher/src/storage/models.rs @@ -2,9 +2,9 @@ use tl_proto::{TlError, TlPacket, TlRead, TlResult, TlWrite}; use tycho_slasher_traits::ValidationSessionId; use tycho_util::FastHashSet; +use crate::BlocksBatch; +use crate::bc::ObservedHistory; use crate::util::BitSet; -use crate::{BlocksBatch, SignatureHistory}; - // === Vset State Stuff === #[derive(Debug, Clone, Copy, TlRead, TlWrite)] @@ -47,23 +47,33 @@ impl TlWrite for StoredBlocksBatch { fn max_size_hint(&self) -> usize { 4 + 4 + + (4 + 4) + + 4 + self.0.committed_blocks.max_size_hint() + 4 - + self - .0 - .signatures_history + + (self.0.observed) .iter() - .map(|item| 4 + item.bits.max_size_hint()) + .map(|item| 4 + 4 + item.bits.max_size_hint()) .sum::() } fn write_to(&self, packet: &mut P) { packet.write_u32(Self::TL_ID); packet.write_u32(self.0.start_seqno); + if let Some(anchor_range) = self.0.round_range.as_ref() { + packet.write_u32(*anchor_range.start()); + packet.write_u32(*anchor_range.end()); + } else { + packet.write_u32(0); + packet.write_u32(0); + }; + packet.write_u32(self.0.filled_rounds); + self.0.committed_blocks.write_to(packet); - packet.write_u32(self.0.signatures_history.len() as u32); - for item in &self.0.signatures_history { + packet.write_u32(self.0.observed.len() as u32); + for item in &self.0.observed { packet.write_u32(item.validator_idx as u32); + packet.write_u32(item.points_proven as u32); item.bits.write_to(packet); } } @@ -78,6 +88,10 @@ impl<'tl> TlRead<'tl> for StoredBlocksBatch { } let start_seqno = u32::read_from(packet)?; + let round_range = Some(u32::read_from(packet)?..=u32::read_from(packet)?) + .filter(|range| *range.end() > 0); + let filled_rounds = u32::read_from(packet)?; + let committed_blocks = BitSet::read_from(packet)?; let block_count = committed_blocks.len(); if start_seqno.checked_add(block_count as u32).is_none() { @@ -91,7 +105,7 @@ impl<'tl> TlRead<'tl> for StoredBlocksBatch { return Err(TlError::InvalidData); } - let mut signatures_history = Vec::with_capacity(history_count); + let mut observed = Vec::with_capacity(history_count); let mut unique_indices = FastHashSet::with_capacity_and_hasher(history_count, Default::default()); for _ in 0..history_count { @@ -101,20 +115,25 @@ impl<'tl> TlRead<'tl> for StoredBlocksBatch { if !unique_indices.insert(validator_idx) { return Err(TlError::InvalidData); } + let points_proven = + u16::try_from(u32::read_from(packet)?).map_err(|_e| TlError::InvalidData)?; let bits = BitSet::read_from(packet)?; if bits.len() != block_count * 2 { return Err(TlError::InvalidData); } - signatures_history.push(SignatureHistory { + observed.push(ObservedHistory { validator_idx, + points_proven, bits, }); } Ok(Self(BlocksBatch { start_seqno, + round_range, + filled_rounds, committed_blocks, - signatures_history: signatures_history.into_boxed_slice(), + observed: observed.into_boxed_slice(), })) } }