From a47087b35e8175840e78a67865689340ac11b920 Mon Sep 17 00:00:00 2001 From: dicethedev Date: Mon, 21 Sep 2026 14:40:10 +0100 Subject: [PATCH 1/2] fix(sync): prevent validator duties on stale heads --- crates/blockchain/src/lib.rs | 6 +-- crates/blockchain/src/sync_status.rs | 8 +++ crates/net/p2p/src/req_resp/handlers.rs | 72 ++++++++++++++++++++++++- crates/storage/src/store.rs | 47 +++++++++++++++- 4 files changed, 126 insertions(+), 7 deletions(-) diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 7de7b898..191402ba 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -1215,11 +1215,7 @@ impl BlockChainServer { fn update_sync_status(&mut self, current_slot: u64) { let head_slot = self.store.head_slot(); - let max_seen_slot = self - .store - .max_live_chain_slot() - .expect("max live chain slot exists") - .unwrap_or(head_slot); + let max_seen_slot = self.store.latest_known_block_slot(); let status = self .sync_status .update(current_slot, head_slot, max_seen_slot); diff --git a/crates/blockchain/src/sync_status.rs b/crates/blockchain/src/sync_status.rs index 5e968a5c..66a96046 100644 --- a/crates/blockchain/src/sync_status.rs +++ b/crates/blockchain/src/sync_status.rs @@ -158,6 +158,14 @@ mod tests { ); } + #[test] + fn fresh_peer_head_blocks_duties_during_long_range_sync() { + let mut tracker = SyncStatusTracker::default(); + + assert_eq!(tracker.update(550, 0, 550), SyncStatus::Syncing); + assert!(!tracker.duties_allowed()); + } + #[test] fn sync_status_treats_stale_known_blocks_as_network_stall() { let mut tracker = SyncStatusTracker::default(); diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 56913ad0..1a993034 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -1,11 +1,11 @@ use std::collections::HashSet; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use ethlambda_network_api::BlockSource; use ethlambda_storage::Store; use libp2p::{PeerId, request_response}; use rand::seq::SliceRandom; use spawned_concurrency::tasks::{Context, send_after}; -use std::time::Duration; use tracing::{debug, error, trace, warn}; use ethlambda_types::checkpoint::Checkpoint; @@ -161,6 +161,7 @@ async fn handle_status_request( peer: PeerId, ) { trace!(finalized_slot=%request.finalized.slot, head_slot=%request.head.slot, "Received status request from peer {peer}"); + observe_peer_head(&server.store, &request, peer); let our_status = build_status(&server.store); let response = Response::success(ResponsePayload::Status(our_status)); server.swarm_handle.send_response(channel, response); @@ -169,6 +170,10 @@ async fn handle_status_request( async fn handle_status_response(server: &mut P2PServer, status: Status, peer: PeerId) { trace!(finalized_slot=%status.finalized.slot, head_slot=%status.head.slot, "Received status response from peer {peer}"); + if !observe_peer_head(&server.store, &status, peer) { + return; + } + let our_head_slot = server.store.head_slot(); if status.head.slot <= our_head_slot { return; @@ -200,6 +205,31 @@ async fn handle_status_response(server: &mut P2PServer, status: Status, peer: Pe trace!(%peer, start_slot, gap, "Long-range sync: using BlocksByRange"); } +/// Record a peer's head when it is not implausibly ahead of the local wall clock. +fn observe_peer_head(store: &Store, status: &Status, peer: PeerId) -> bool { + let config = store.config(); + let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_millis() as u64) + .unwrap_or_else(|_| config.genesis_time_ms()); + let current_slot = + now_ms.saturating_sub(config.genesis_time_ms()) / config.milliseconds_per_slot; + let max_head_slot = current_slot.saturating_add(1); + + if status.head.slot > max_head_slot { + warn!( + %peer, + peer_head_slot = status.head.slot, + current_slot, + "Ignoring peer status with a future head" + ); + return false; + } + + store.observe_block_slot(status.head.slot); + true +} + async fn handle_blocks_by_root_request( server: &mut P2PServer, request: BlocksByRootRequest, @@ -609,6 +639,46 @@ mod tests { } } + #[test] + fn peer_status_advances_latest_known_block_slot() { + let backend = Arc::new(InMemoryBackend::new()); + let store = Store::from_anchor_state( + backend, + State::from_genesis(0, vec![]), + DEFAULT_MILLISECONDS_PER_SLOT, + ); + let status = Status { + finalized: Checkpoint::default(), + head: Checkpoint { + root: H256::ZERO, + slot: 550, + }, + }; + + assert!(observe_peer_head(&store, &status, PeerId::random())); + assert_eq!(store.latest_known_block_slot(), 550); + } + + #[test] + fn peer_status_rejects_implausibly_future_head() { + let backend = Arc::new(InMemoryBackend::new()); + let store = Store::from_anchor_state( + backend, + State::from_genesis(0, vec![]), + DEFAULT_MILLISECONDS_PER_SLOT, + ); + let status = Status { + finalized: Checkpoint::default(), + head: Checkpoint { + root: H256::ZERO, + slot: u64::MAX, + }, + }; + + assert!(!observe_peer_head(&store, &status, PeerId::random())); + assert_eq!(store.latest_known_block_slot(), 0); + } + #[test] fn blocks_by_range_returns_canonical_blocks_in_requested_order() { let backend = Arc::new(InMemoryBackend::new()); diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index b92aaad2..cc818dee 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -1,5 +1,6 @@ use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::num::NonZeroUsize; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, LazyLock, Mutex}; use lru::LruCache; @@ -563,6 +564,11 @@ pub struct Store { /// LRU memoization of states by block root, shared across `Store` clones. /// Avoids reconstructing recent states from diffs on every read. state_cache: Arc>>, + /// Highest plausible block slot observed locally or advertised by a peer. + /// + /// This runtime-only high-water mark is shared across `Store` clones so the + /// networking actor can inform the blockchain actor's sync-duty gate. + latest_known_block_slot: Arc, } /// Build an empty state cache sized to [`STATE_CACHE_CAPACITY`]. @@ -682,8 +688,11 @@ impl Store { GOSSIP_SIGNATURE_CAP, ))), state_cache: new_state_cache(), + latest_known_block_slot: Arc::new(AtomicU64::new(0)), }; + store.observe_block_slot(store.head_slot()); + // Also compare against the finalized state: the persisted config // carries no validator registry, so the check above cannot catch a // chain that shares our genesis time and cadence but not our validator @@ -819,6 +828,9 @@ impl Store { GOSSIP_SIGNATURE_CAP, ))), state_cache: new_state_cache(), + latest_known_block_slot: Arc::new(AtomicU64::new( + anchor_state.latest_block_header.slot, + )), }) } @@ -863,6 +875,20 @@ impl Store { self.time().expect("store time exists") / INTERVALS_PER_SLOT } + /// Record evidence that a block exists at `slot`. + /// + /// The marker is monotonic for the lifetime of the process and shared by + /// all clones of this store. + pub fn observe_block_slot(&self, slot: u64) { + self.latest_known_block_slot + .fetch_max(slot, Ordering::Relaxed); + } + + /// Return the highest block slot observed during this process. + pub fn latest_known_block_slot(&self) -> u64 { + self.latest_known_block_slot.load(Ordering::Relaxed) + } + // ============ Config ============ /// Returns the chain configuration. @@ -1210,8 +1236,9 @@ impl Store { signed_block: SignedBlock, ) -> Result<(), Error> { let mut batch = self.backend.begin_write().expect("write batch"); - write_signed_block(batch.as_mut(), &root, signed_block); + let block = write_signed_block(batch.as_mut(), &root, signed_block); batch.commit().expect("commit"); + self.observe_block_slot(block.slot); Ok(()) } @@ -1239,6 +1266,7 @@ impl Store { .expect("put non-finalized chain index"); batch.commit().expect("commit"); + self.observe_block_slot(block.slot); self.record_known_attestation_votes(&block.body.attestations); Ok(()) } @@ -1996,6 +2024,7 @@ mod tests { GOSSIP_SIGNATURE_CAP, ))), state_cache: new_state_cache(), + latest_known_block_slot: Default::default(), } } @@ -2012,12 +2041,27 @@ mod tests { GOSSIP_SIGNATURE_CAP, ))), state_cache: new_state_cache(), + latest_known_block_slot: Default::default(), } } } // ============ Block Signature Pruning Tests ============ + #[test] + fn observed_block_slot_is_shared_and_monotonic() { + let mut store = Store::test_store(); + let reader = store.clone(); + + store + .insert_pending_block(root(42), signed_block(42, H256::ZERO)) + .expect("insert pending block"); + assert_eq!(reader.latest_known_block_slot(), 42); + + store.observe_block_slot(7); + assert_eq!(reader.latest_known_block_slot(), 42); + } + #[test] fn block_root_index_tracks_canonical_chain_across_reorgs() { let backend = Arc::new(InMemoryBackend::new()); @@ -2096,6 +2140,7 @@ mod tests { .expect("get blocks by slot range"); assert_eq!(blocks.len(), 1); assert_eq!(blocks[0].message.hash_tree_root(), block_root); + assert_eq!(restored.latest_known_block_slot(), 1); } #[test] From 24a8c6b5ee17de82bcfa724c934e401e87312aad Mon Sep 17 00:00:00 2001 From: dicethedev Date: Thu, 24 Sep 2026 14:14:37 +0100 Subject: [PATCH 2/2] fix(sync): address duty gate review feedback --- bin/ethlambda/src/main.rs | 10 +-- crates/blockchain/src/lib.rs | 75 +++++++++++++----- crates/blockchain/src/sync_status.rs | 8 -- crates/common/types/src/constants.rs | 8 ++ crates/net/p2p/src/req_resp/handlers.rs | 101 ++++++++++++++---------- crates/net/rpc/src/node.rs | 7 +- crates/storage/src/store.rs | 22 +++--- 7 files changed, 135 insertions(+), 96 deletions(-) diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index 7510bc1d..9afb2bbb 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -29,7 +29,6 @@ use std::{ net::{IpAddr, SocketAddr}, path::{Path, PathBuf}, sync::Arc, - time::SystemTime, }; use tokio_util::sync::CancellationToken; @@ -772,12 +771,7 @@ async fn fetch_initial_state( // without `--checkpoint-sync-url` keeps the chain instead of writing a // slot-0 anchor over it. if let Some(store) = Store::from_db_state(backend.clone(), genesis)? { - let now_ms = SystemTime::UNIX_EPOCH - .elapsed() - .expect("already past the unix epoch") - .as_millis() as u64; - let current_slot = - now_ms.saturating_sub(genesis.genesis_time * 1000) / genesis.milliseconds_per_slot; + let current_slot = store.wall_clock_slot(); let head_slot = store.head_slot(); let gap = current_slot.saturating_sub(head_slot); if gap <= MAX_RESUMABLE_DB_STATE_AGE { @@ -1070,7 +1064,7 @@ validators: const UNREACHABLE_CHECKPOINT_URL: &str = "http://127.0.0.1:1"; fn now_secs() -> u64 { - SystemTime::UNIX_EPOCH + std::time::SystemTime::UNIX_EPOCH .elapsed() .expect("already past the unix epoch") .as_secs() diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 191402ba..2c326cdb 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -80,17 +80,10 @@ pub struct BlockChainConfig { // derives slots from `store.time()` and must not carry a second copy of a // consensus-critical constant. pub use ethlambda_types::block::MAX_ATTESTATIONS_DATA; -pub use ethlambda_types::constants::{DEFAULT_MILLISECONDS_PER_SLOT, INTERVALS_PER_SLOT}; +pub use ethlambda_types::constants::{ + DEFAULT_MILLISECONDS_PER_SLOT, GOSSIP_DISPARITY_INTERVALS, INTERVALS_PER_SLOT, +}; pub use sync_status::SyncStatusController; -/// Future-slot tolerance for gossip attestations, expressed in intervals. -/// -/// Bounds the clock skew the time check is willing to absorb when admitting a -/// vote whose slot has not yet started locally. One interval is a fifth of the -/// configured slot, the lean analogue of mainnet's -/// `MAXIMUM_GOSSIP_CLOCK_DISPARITY`. -/// -/// See: leanSpec PR #682. -pub const GOSSIP_DISPARITY_INTERVALS: u64 = 1; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum SlotInterval { @@ -188,9 +181,7 @@ impl BlockChain { // Catch XMSS keys up to the current slot before the first tick // store.time() doesn't work here: after an offline gap it lags wall-clock by // exactly the gap we need to catch up through - let now_ms = unix_now_ms(); - let current_slot = (now_ms.saturating_sub(time_config.genesis_time_ms()) - / time_config.milliseconds_per_slot) as u32; + let current_slot = store.wall_clock_slot() as u32; key_manager.advance_keys_to(current_slot); let handle = BlockChainServer { @@ -313,6 +304,18 @@ pub struct BlockChainServer { events: EventBus, } +fn sync_status_from_store( + tracker: &mut SyncStatusTracker, + store: &Store, + current_slot: u64, +) -> metrics::SyncStatus { + tracker.update( + current_slot, + store.head_slot(), + store.latest_known_block_slot(), + ) +} + impl BlockChainServer { async fn on_tick(&mut self, timestamp_ms: u64, ctx: &Context) { let time_config = *self.store.config(); @@ -890,9 +893,7 @@ impl BlockChainServer { } // Block import has no ready-made "now" slot like `on_tick`'s, so // compute the wall-clock slot fresh for the head-recency gate. - let time_config = *self.store.config(); - let wall_clock_slot = unix_now_ms().saturating_sub(time_config.genesis_time_ms()) - / time_config.milliseconds_per_slot; + let wall_clock_slot = self.store.wall_clock_slot(); pre_import.diff_and_emit(&self.store, &self.events, wall_clock_slot); metrics::update_head_slot(self.store.head_slot()); @@ -1214,11 +1215,7 @@ impl BlockChainServer { } fn update_sync_status(&mut self, current_slot: u64) { - let head_slot = self.store.head_slot(); - let max_seen_slot = self.store.latest_known_block_slot(); - let status = self - .sync_status - .update(current_slot, head_slot, max_seen_slot); + let status = sync_status_from_store(&mut self.sync_status, &self.store, current_slot); metrics::set_node_sync_status(status); self.sync_status_controller.set(status); } @@ -1486,6 +1483,11 @@ impl Handler for BlockChainServer { #[cfg(test)] mod tests { use super::*; + use ethlambda_storage::backend::InMemoryBackend; + use ethlambda_types::{ + block::{Block, BlockBody, MultiMessageAggregate}, + state::State, + }; const GENESIS_TIME: u64 = 1_000; @@ -1493,6 +1495,37 @@ mod tests { ChainConfig::new(GENESIS_TIME, milliseconds_per_slot) } + #[test] + fn pending_block_marks_node_syncing_and_blocks_duties() { + let backend = std::sync::Arc::new(InMemoryBackend::new()); + let mut store = Store::from_anchor_state( + backend, + State::from_genesis(0, vec![]), + DEFAULT_MILLISECONDS_PER_SLOT, + ); + let pending = SignedBlock { + message: Block { + slot: 550, + proposer_index: 0, + parent_root: H256::ZERO, + state_root: H256::ZERO, + body: BlockBody::default(), + }, + proof: MultiMessageAggregate::default(), + }; + let root = pending.message.hash_tree_root(); + store + .insert_pending_block(root, pending) + .expect("insert pending block"); + let mut tracker = SyncStatusTracker::default(); + + assert_eq!( + sync_status_from_store(&mut tracker, &store, 550), + metrics::SyncStatus::Syncing + ); + assert!(!tracker.duties_allowed()); + } + #[test] fn interval_boundaries_scale_with_the_slot_duration() { let default = config(DEFAULT_MILLISECONDS_PER_SLOT); diff --git a/crates/blockchain/src/sync_status.rs b/crates/blockchain/src/sync_status.rs index 66a96046..5e968a5c 100644 --- a/crates/blockchain/src/sync_status.rs +++ b/crates/blockchain/src/sync_status.rs @@ -158,14 +158,6 @@ mod tests { ); } - #[test] - fn fresh_peer_head_blocks_duties_during_long_range_sync() { - let mut tracker = SyncStatusTracker::default(); - - assert_eq!(tracker.update(550, 0, 550), SyncStatus::Syncing); - assert!(!tracker.duties_allowed()); - } - #[test] fn sync_status_treats_stale_known_blocks_as_network_stall() { let mut tracker = SyncStatusTracker::default(); diff --git a/crates/common/types/src/constants.rs b/crates/common/types/src/constants.rs index 8b74ec4a..f5c79df8 100644 --- a/crates/common/types/src/constants.rs +++ b/crates/common/types/src/constants.rs @@ -17,6 +17,14 @@ pub const FORK_DIGEST: &str = "12345678"; /// see [`crate::chain_config::ChainConfig`]. pub const INTERVALS_PER_SLOT: u64 = 5; +/// Future-slot tolerance for gossip objects, expressed in intervals. +/// +/// One interval is a fifth of the configured slot, the lean analogue of +/// mainnet's `MAXIMUM_GOSSIP_CLOCK_DISPARITY`. +/// +/// See: leanSpec PR #682. +pub const GOSSIP_DISPARITY_INTERVALS: u64 = 1; + /// Slot duration used when the network's config file omits /// `MILLISECONDS_PER_SLOT`. /// diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 1a993034..e34da020 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -1,5 +1,5 @@ use std::collections::HashSet; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::Duration; use ethlambda_network_api::BlockSource; use ethlambda_storage::Store; @@ -9,6 +9,7 @@ use spawned_concurrency::tasks::{Context, send_after}; use tracing::{debug, error, trace, warn}; use ethlambda_types::checkpoint::Checkpoint; +use ethlambda_types::constants::{GOSSIP_DISPARITY_INTERVALS, INTERVALS_PER_SLOT}; use ethlambda_types::primitives::HashTreeRoot as _; use ethlambda_types::{block::SignedBlock, primitives::H256}; @@ -170,30 +171,65 @@ async fn handle_status_request( async fn handle_status_response(server: &mut P2PServer, status: Status, peer: PeerId) { trace!(finalized_slot=%status.finalized.slot, head_slot=%status.head.slot, "Received status response from peer {peer}"); - if !observe_peer_head(&server.store, &status, peer) { + let Some((start_slot, gap)) = + prepare_range_sync(&server.store, &mut server.range_sync_state, &status, peer) + else { return; + }; + + request_next_range_batch(server).await; + trace!(%peer, start_slot, gap, "Long-range sync: using BlocksByRange"); +} + +/// Record a peer's head when it satisfies the same future bound as a block. +fn observe_peer_head(store: &Store, status: &Status, peer: PeerId) -> bool { + let head_start_interval = status.head.slot.saturating_mul(INTERVALS_PER_SLOT); + let store_time = store.time().expect("store time exists"); + + if head_start_interval > store_time.saturating_add(GOSSIP_DISPARITY_INTERVALS) { + warn!( + %peer, + peer_head_slot = status.head.slot, + store_time, + "Ignoring peer status with a future head" + ); + return false; } - let our_head_slot = server.store.head_slot(); - if status.head.slot <= our_head_slot { - return; + store.observe_block_slot(status.head.slot); + true +} + +fn prepare_range_sync( + store: &Store, + range_sync_state: &mut Option, + status: &Status, + peer: PeerId, +) -> Option<(u64, u64)> { + if !observe_peer_head(store, status, peer) { + return None; + } + + let local_head_slot = store.head_slot(); + if status.head.slot <= local_head_slot { + return None; } - let gap = status.head.slot - our_head_slot; + + let gap = status.head.slot - local_head_slot; debug!( %peer, peer_head_slot = status.head.slot, - local_head_slot = our_head_slot, + local_head_slot, slot_gap = gap, "Peer status head is ahead of local head" ); - let start_slot = our_head_slot.saturating_add(1); + let start_slot = local_head_slot.saturating_add(1); let end_exclusive = start_slot.saturating_add(gap.min(MAX_SYNC_RANGE)); - - match &mut server.range_sync_state { + match range_sync_state { Some(state) => state.merge_peer(peer, status.head.slot, end_exclusive), None => { - server.range_sync_state = Some(RangeSyncState::new( + *range_sync_state = Some(RangeSyncState::new( start_slot..end_exclusive, peer, status.head.slot, @@ -201,33 +237,7 @@ async fn handle_status_response(server: &mut P2PServer, status: Status, peer: Pe } } - request_next_range_batch(server).await; - trace!(%peer, start_slot, gap, "Long-range sync: using BlocksByRange"); -} - -/// Record a peer's head when it is not implausibly ahead of the local wall clock. -fn observe_peer_head(store: &Store, status: &Status, peer: PeerId) -> bool { - let config = store.config(); - let now_ms = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|duration| duration.as_millis() as u64) - .unwrap_or_else(|_| config.genesis_time_ms()); - let current_slot = - now_ms.saturating_sub(config.genesis_time_ms()) / config.milliseconds_per_slot; - let max_head_slot = current_slot.saturating_add(1); - - if status.head.slot > max_head_slot { - warn!( - %peer, - peer_head_slot = status.head.slot, - current_slot, - "Ignoring peer status with a future head" - ); - return false; - } - - store.observe_block_slot(status.head.slot); - true + Some((start_slot, gap)) } async fn handle_blocks_by_root_request( @@ -642,11 +652,14 @@ mod tests { #[test] fn peer_status_advances_latest_known_block_slot() { let backend = Arc::new(InMemoryBackend::new()); - let store = Store::from_anchor_state( + let mut store = Store::from_anchor_state( backend, State::from_genesis(0, vec![]), DEFAULT_MILLISECONDS_PER_SLOT, ); + store + .set_time(550 * INTERVALS_PER_SLOT) + .expect("set store time"); let status = Status { finalized: Checkpoint::default(), head: Checkpoint { @@ -660,7 +673,7 @@ mod tests { } #[test] - fn peer_status_rejects_implausibly_future_head() { + fn future_peer_status_does_not_start_range_sync() { let backend = Arc::new(InMemoryBackend::new()); let store = Store::from_anchor_state( backend, @@ -671,11 +684,15 @@ mod tests { finalized: Checkpoint::default(), head: Checkpoint { root: H256::ZERO, - slot: u64::MAX, + slot: 1, }, }; + let mut range_sync_state = None; - assert!(!observe_peer_head(&store, &status, PeerId::random())); + assert!( + prepare_range_sync(&store, &mut range_sync_state, &status, PeerId::random()).is_none() + ); + assert!(range_sync_state.is_none()); assert_eq!(store.latest_known_block_slot(), 0); } diff --git a/crates/net/rpc/src/node.rs b/crates/net/rpc/src/node.rs index ab0999cc..fce3bf77 100644 --- a/crates/net/rpc/src/node.rs +++ b/crates/net/rpc/src/node.rs @@ -38,12 +38,7 @@ async fn get_syncing( State(store): State, Extension(sync_status): Extension, ) -> impl IntoResponse { - let genesis_ms = store.config().genesis_time_ms(); - let now_ms = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_millis() as u64) - .unwrap_or(genesis_ms); - let wall_slot = now_ms.saturating_sub(genesis_ms) / store.config().milliseconds_per_slot; + let wall_slot = store.wall_clock_slot(); let head_slot = store.head_slot(); let sync_distance = wall_slot.saturating_sub(head_slot); let finalized_slot = store diff --git a/crates/storage/src/store.rs b/crates/storage/src/store.rs index cc818dee..d9ec507d 100644 --- a/crates/storage/src/store.rs +++ b/crates/storage/src/store.rs @@ -2,6 +2,7 @@ use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::num::NonZeroUsize; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, LazyLock, Mutex}; +use std::time::SystemTime; use lru::LruCache; @@ -875,6 +876,16 @@ impl Store { self.time().expect("store time exists") / INTERVALS_PER_SLOT } + /// Return the slot containing the current wall-clock time. + pub fn wall_clock_slot(&self) -> u64 { + let genesis_ms = self.config.genesis_time_ms(); + let now_ms = SystemTime::UNIX_EPOCH + .elapsed() + .map(|duration| duration.as_millis() as u64) + .unwrap_or(genesis_ms); + now_ms.saturating_sub(genesis_ms) / self.config.milliseconds_per_slot + } + /// Record evidence that a block exists at `slot`. /// /// The marker is monotonic for the lifetime of the process and shared by @@ -1079,17 +1090,6 @@ impl Store { .collect()) } - /// Return the highest slot in the live chain. - pub fn max_live_chain_slot(&self) -> Result, Error> { - let view = self.backend.begin_read().expect("read view"); - Ok(view - .prefix_iterator(Table::LiveChain, &[]) - .expect("iterator") - .filter_map(Result::ok) - .map(|(key, _)| decode_slot_root_key(&key).0) - .max()) - } - /// Get all known block roots as HashSet. /// /// Useful for checking block existence without deserializing.