fix(sync): prevent validator duties on stale heads

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()

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<Self>) {
         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,15 +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 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);
     }
@@ -1490,13 +1483,49 @@ impl Handler<AggregationDeadline> 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;
 
     fn config(milliseconds_per_slot: u64) -> ChainConfig {
         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);

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`.
 ///

crates/net/p2p/src/req_resp/handlers.rs

@@ -1,14 +1,15 @@
 use std::collections::HashSet;
+use std::time::Duration;
 
 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;
+use ethlambda_types::constants::{GOSSIP_DISPARITY_INTERVALS, INTERVALS_PER_SLOT};
 use ethlambda_types::primitives::HashTreeRoot as _;
 use ethlambda_types::{block::SignedBlock, primitives::H256};
 
@@ -161,6 +162,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,35 +171,73 @@ 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}");
 
-    let our_head_slot = server.store.head_slot();
-    if status.head.slot <= our_head_slot {
+    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;
+    }
+
+    store.observe_block_slot(status.head.slot);
+    true
+}
+
+fn prepare_range_sync(
+    store: &Store,
+    range_sync_state: &mut Option<RangeSyncState>,
+    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,
             ));
         }
     }
 
-    request_next_range_batch(server).await;
-    trace!(%peer, start_slot, gap, "Long-range sync: using BlocksByRange");
+    Some((start_slot, gap))
 }
 
 async fn handle_blocks_by_root_request(
@@ -609,6 +649,53 @@ mod tests {
         }
     }
 
+    #[test]
+    fn peer_status_advances_latest_known_block_slot() {
+        let backend = Arc::new(InMemoryBackend::new());
+        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 {
+                root: H256::ZERO,
+                slot: 550,
+            },
+        };
+
+        assert!(observe_peer_head(&store, &status, PeerId::random()));
+        assert_eq!(store.latest_known_block_slot(), 550);
+    }
+
+    #[test]
+    fn future_peer_status_does_not_start_range_sync() {
+        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: 1,
+            },
+        };
+        let mut range_sync_state = None;
+
+        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);
+    }
+
     #[test]
     fn blocks_by_range_returns_canonical_blocks_in_requested_order() {
         let backend = Arc::new(InMemoryBackend::new());

crates/net/rpc/src/node.rs

@@ -38,12 +38,7 @@ async fn get_syncing(
     State(store): State<Store>,
     Extension(sync_status): Extension<SyncStatusController>,
 ) -> 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