Merge branch 'main' into feat/multi-server-devnet-skill

Author: pablodeymo

bin/ethlambda/src/checkpoint_sync.rs

@@ -1,6 +1,7 @@
 use std::time::Duration;
 
 use ethlambda_types::block::SignedBlock;
+use ethlambda_types::genesis::{GenesisMismatch, verify_state_genesis};
 use ethlambda_types::primitives::HashTreeRoot as _;
 use ethlambda_types::state::{State, Validator, anchor_pair_is_consistent};
 use libssz::{DecodeError, SszDecode};
@@ -45,20 +46,13 @@ pub enum CheckpointSyncError {
     SlotIsZero,
     #[error("checkpoint state has no validators")]
     NoValidators,
-    #[error("genesis time mismatch: expected {expected}, got {got}")]
-    GenesisTimeMismatch { expected: u64, got: u64 },
-    #[error("validator count mismatch: expected {expected}, got {got}")]
-    ValidatorCountMismatch { expected: usize, got: usize },
-    #[error(
-        "validator at position {position} has non-sequential index (expected {expected}, got {got})"
-    )]
-    NonSequentialValidatorIndex {
-        position: usize,
-        expected: u64,
-        got: u64,
-    },
-    #[error("validator {index} pubkey mismatch (attestation or proposal key)")]
-    ValidatorPubkeyMismatch { index: usize },
+    #[error("checkpoint state does not match the configured genesis: {0}")]
+    Genesis(#[from] GenesisMismatch),
+    /// Reading the persisted store failed, including the case where the data
+    /// directory holds another network's chain. Startup aborts: the operator
+    /// has to point at the right data directory or remove it.
+    #[error("failed to load persisted DB state: {0}")]
+    DbState(#[from] ethlambda_storage::Error),
     #[error("finalized slot cannot exceed state slot")]
     FinalizedExceedsStateSlot,
     #[error("justified slot cannot precede finalized slot")]
@@ -197,7 +191,8 @@ fn verify_checkpoint_state(
     expected_genesis_time: u64,
     expected_validators: &[Validator],
 ) -> Result<(), CheckpointSyncError> {
-    // Slot sanity check
+    // Slot sanity check. Checkpoint-specific: unlike a state loaded from our
+    // own data directory, a downloaded anchor at genesis is never legitimate.
     if state.slot == 0 {
         return Err(CheckpointSyncError::SlotIsZero);
     }
@@ -207,46 +202,10 @@ fn verify_checkpoint_state(
         return Err(CheckpointSyncError::NoValidators);
     }
 
-    // Genesis time matches
-    if state.config.genesis_time != expected_genesis_time {
-        return Err(CheckpointSyncError::GenesisTimeMismatch {
-            expected: expected_genesis_time,
-            got: state.config.genesis_time,
-        });
-    }
-
-    // Validator count matches
-    if state.validators.len() != expected_validators.len() {
-        return Err(CheckpointSyncError::ValidatorCountMismatch {
-            expected: expected_validators.len(),
-            got: state.validators.len(),
-        });
-    }
-
-    // Validator indices are sequential (0, 1, 2, ...)
-    for (position, validator) in state.validators.iter().enumerate() {
-        if validator.index != position as u64 {
-            return Err(CheckpointSyncError::NonSequentialValidatorIndex {
-                position,
-                expected: position as u64,
-                got: validator.index,
-            });
-        }
-    }
-
-    // Validator pubkeys match (critical security check)
-    for (i, (state_val, expected_val)) in state
-        .validators
-        .iter()
-        .zip(expected_validators.iter())
-        .enumerate()
-    {
-        if state_val.attestation_pubkey != expected_val.attestation_pubkey
-            || state_val.proposal_pubkey != expected_val.proposal_pubkey
-        {
-            return Err(CheckpointSyncError::ValidatorPubkeyMismatch { index: i });
-        }
-    }
+    // Genesis time and the full validator registry match our config. Shared
+    // with the resume-from-disk path so both entry points agree on what makes a
+    // state ours.
+    verify_state_genesis(state, expected_genesis_time, expected_validators)?;
 
     // Finalized slot sanity
     if state.latest_finalized.slot > state.slot {

bin/ethlambda/src/main.rs

@@ -688,7 +688,7 @@ async fn fetch_initial_state(
     // have. Tried before the checkpoint-sync and genesis paths so that a restart
     // without `--checkpoint-sync-url` keeps the chain instead of writing a
     // slot-0 anchor over it.
-    if let Ok(Some(store)) = Store::from_db_state(backend.clone(), genesis.genesis_time) {
+    if let Some(store) = Store::from_db_state(backend.clone(), genesis)? {
         let now_ms = SystemTime::UNIX_EPOCH
             .elapsed()
             .expect("already past the unix epoch")
@@ -978,26 +978,50 @@ validators:
         );
     }
 
-    /// A DB from another network is not resumable, so the no-URL path still
-    /// falls back to genesis.
-    ///
-    /// This pins current behavior, not a desired one. `from_db_state` treats a
-    /// `GENESIS_TIME` mismatch as an empty DB and only warns, so with no
-    /// checkpoint URL the node writes a genesis anchor over a populated
-    /// foreign-network directory: the same data loss the resume ordering
-    /// removes everywhere else. Left as-is deliberately, and this test is here
-    /// to make the change visible when someone fixes it.
+    /// A DB from another network aborts startup rather than being re-anchored:
+    /// writing genesis on top would leave the foreign blocks in place, and
+    /// slot-indexed reads would serve them to peers.
     #[tokio::test]
-    async fn initializes_from_genesis_when_db_genesis_time_differs() {
+    async fn fails_when_db_genesis_time_differs() {
         let seeded_genesis = test_genesis(now_secs());
         let backend = Arc::new(InMemoryBackend::default());
         seed_db(backend.clone(), &seeded_genesis);
 
         let other_genesis = test_genesis(seeded_genesis.genesis_time + 1);
-        let store = fetch_initial_state(&[], &other_genesis, backend)
-            .await
-            .unwrap();
+        // `Store` is not `Debug`, so unwrap the error by pattern.
+        let Err(err) = fetch_initial_state(&[], &other_genesis, backend.clone()).await else {
+            panic!("a foreign DB must not be silently re-anchored");
+        };
 
-        assert_eq!(store.head_slot(), 0);
+        assert!(
+            matches!(err, checkpoint_sync::CheckpointSyncError::DbState(_)),
+            "unexpected error: {err}"
+        );
+        // The foreign chain is left untouched, not overwritten with a new anchor.
+        let store = Store::from_db_state(backend, &seeded_genesis)
+            .expect("original DB still loads under its own genesis")
+            .expect("store exists");
+        assert_eq!(store.head_slot(), SEEDED_HEAD_SLOT);
+    }
+
+    /// Same genesis time, different validator registry: the case the previous
+    /// `genesis_time`-only check could not see.
+    #[tokio::test]
+    async fn fails_when_db_validator_set_differs() {
+        let genesis_time = now_secs();
+        let seeded_genesis = test_genesis(genesis_time);
+        let backend = Arc::new(InMemoryBackend::default());
+        seed_db(backend.clone(), &seeded_genesis);
+
+        let mut other_genesis = test_genesis(genesis_time);
+        other_genesis.genesis_validators[0].attestation_pubkey = [9u8; 52];
+        let Err(err) = fetch_initial_state(&[], &other_genesis, backend).await else {
+            panic!("a foreign validator set must not be silently re-anchored");
+        };
+
+        assert!(
+            matches!(err, checkpoint_sync::CheckpointSyncError::DbState(_)),
+            "unexpected error: {err}"
+        );
     }
 }

crates/blockchain/src/lib.rs

@@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet, VecDeque};
 use std::time::{Duration, Instant, SystemTime};
 
 use ethlambda_crypto::signature::{ValidatorPublicKey, ValidatorSignature};
-use ethlambda_network_api::{BlockChainToP2PRef, InitP2P};
+use ethlambda_network_api::{BlockChainToP2PRef, BlockSource, InitP2P};
 use ethlambda_state_transition::is_proposer;
 use ethlambda_storage::{ALL_TABLES, Store};
 use ethlambda_types::{
@@ -109,6 +109,20 @@ impl SlotInterval {
             _ => unreachable!("slots only have 5 intervals"),
         }
     }
+
+    /// Milliseconds from genesis to the start of this interval in `slot`.
+    ///
+    /// Inverse of [`Self::from_ms_since_genesis`].
+    pub(crate) fn to_ms_since_genesis(self, slot: u64) -> u64 {
+        let interval = match self {
+            Self::BlockPublication => 0,
+            Self::AttestationProduction => 1,
+            Self::Aggregation => 2,
+            Self::SafeTargetUpdate => 3,
+            Self::EndOfSlot => 4,
+        };
+        slot * MILLISECONDS_PER_SLOT + interval * MILLISECONDS_PER_INTERVAL
+    }
 }
 
 /// Milliseconds until the next interval boundary, measured relative to genesis.
@@ -1367,16 +1381,37 @@ impl Handler<InitP2P> for BlockChainServer {
 
 impl Handler<NewBlock> for BlockChainServer {
     async fn handle(&mut self, msg: NewBlock, _ctx: &Context<Self>) {
-        self.events.emit(ChainEvent::BlockGossip {
-            slot: msg.block.message.slot,
-            block: msg.block.message.hash_tree_root(),
-        });
+        let arrival_ms = unix_now_ms();
+        // Gate both the event and the arrival metric on BlockSource::Gossip for
+        // two reasons: `ChainEvent::BlockGossip` is documented (events.rs) as "a
+        // block seen on gossip, before import", yet without this gate it also
+        // fired for req/resp sync blocks; and sync backfill delivers blocks many
+        // slots after they were due, which would swamp the arrival histogram
+        // with stale deltas that reflect catch-up speed, not gossip timeliness.
+        // `self.on_block(msg.block)` still runs for every source below: it is
+        // the import path and must not be gated.
+        if msg.source == BlockSource::Gossip {
+            let slot = msg.block.message.slot;
+            self.events.emit(ChainEvent::BlockGossip {
+                slot,
+                block: msg.block.message.hash_tree_root(),
+            });
+            let genesis_ms = self.store.config().expect("config exists").genesis_time * 1000;
+            metrics::observe_gossip_block_arrival(arrival_ms, genesis_ms, slot);
+        }
         self.on_block(msg.block);
     }
 }
 
 impl Handler<NewAttestation> for BlockChainServer {
     async fn handle(&mut self, msg: NewAttestation, ctx: &Context<Self>) {
+        let arrival_ms = unix_now_ms();
+        let genesis_ms = self.store.config().expect("config exists").genesis_time * 1000;
+        metrics::observe_gossip_attestation_arrival(
+            arrival_ms,
+            genesis_ms,
+            msg.attestation.data.slot,
+        );
         self.on_gossip_attestation(&msg.attestation);
         // Early aggregation only advances the current slot's group counts, so a
         // late- or future-slot attestation can never cross the threshold; skip
@@ -1390,6 +1425,9 @@ impl Handler<NewAttestation> for BlockChainServer {
 
 impl Handler<NewAggregatedAttestation> for BlockChainServer {
     async fn handle(&mut self, msg: NewAggregatedAttestation, _ctx: &Context<Self>) {
+        let arrival_ms = unix_now_ms();
+        let genesis_ms = self.store.config().expect("config exists").genesis_time * 1000;
+        metrics::observe_gossip_aggregation_arrival(arrival_ms, genesis_ms);
         self.on_gossip_aggregated_attestation(msg.attestation);
     }
 }
@@ -1400,6 +1438,8 @@ impl Handler<NewAggregatedAttestation> for BlockChainServer {
 
 impl Handler<AggregateProduced> for BlockChainServer {
     async fn handle(&mut self, msg: AggregateProduced, _ctx: &Context<Self>) {
+        let arrival_ms = unix_now_ms();
+
         // Drop results from a prior session (or from an unexpected late worker).
         // Current session may be None if the actor already cleaned it up; accept
         // the message only when ids match.
@@ -1413,6 +1453,17 @@ impl Handler<AggregateProduced> for BlockChainServer {
             return;
         }
 
+        // Count our own aggregate in the same series as gossip-received ones,
+        // so an aggregator does not report an empty aggregate arrival profile.
+        // Delivery of this message is held to the interval-2 boundary upstream,
+        // so a local aggregate lands near zero unless proving overran the
+        // interval. Sharing one series with received aggregates is deliberate
+        // and costs little in practice: a late aggregate is late for every node
+        // at once, so both populations are dominated by production time rather
+        // than propagation and their distributions look alike.
+        let genesis_ms = self.store.config().expect("config exists").genesis_time * 1000;
+        metrics::observe_gossip_aggregation_arrival(arrival_ms, genesis_ms);
+
         // Publish alignment is enforced upstream: the worker delays delivery of
         // this message until the interval-2 boundary, so by the time it lands
         // the aggregate is safe to apply and gossip immediately.