feat(metrics): add gossip arrival-time histograms and position counters

Network health had plenty of block-production timing but nothing about
reception timing, so "are votes arriving late, or not arriving at all?" could
only be answered out-of-band with tooling/event-monitor's collector-side clock.

Adds three histograms recording the absolute distance between a gossip
message's arrival and the start of the interval it was due in, plus three
counters splitting arrivals into before/inside/after that interval. The
interval is what counts as inside, not the slot: an attestation landing in its
own slot's aggregation interval missed the production interval it was due in,
so it reads as after.

Aggregates anchor to the most recent aggregation-interval boundary instead of
their own data.slot, since a stale-group catch-up aggregate legitimately
carries an older slot and would otherwise fill the histogram with large values
that are not a health problem. That bounds their delay to one slot and makes
before unreachable, so the aggregate counter does not export that series.

Blocks are sampled only when received on gossip; req/resp sync backfill
delivers them many slots after they were due and would swamp the histogram.
Threading a BlockSource through new_block to tell the two apart also fixes
ChainEvent::BlockGossip, which events.rs already documents as gossip-only yet
until now also fired for sync-fetched blocks.

Author: MegaRedHand

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);
     }
 }
@@ -1478,3 +1516,23 @@ impl Handler<AggregationDeadline> for BlockChainServer {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn interval_ms_round_trips() {
+        let intervals = [
+            SlotInterval::BlockPublication,
+            SlotInterval::AttestationProduction,
+            SlotInterval::Aggregation,
+            SlotInterval::SafeTargetUpdate,
+            SlotInterval::EndOfSlot,
+        ];
+        for interval in intervals {
+            let ms = interval.to_ms_since_genesis(7);
+            assert_eq!(SlotInterval::from_ms_since_genesis(ms), interval);
+        }
+    }
+}