feat(metrics): add gossip arrival timing metrics

src/lean_spec/node/chain/clock.py

@@ -25,7 +25,7 @@ class SlotClock:
     time_fn: Callable[[], float] = wall_time
     """Time source function (injectable for testing)."""
 
-    def _milliseconds_since_genesis(self) -> Uint64:
+    def milliseconds_since_genesis(self) -> Uint64:
         """Milliseconds elapsed since genesis (0 if before genesis)."""
         now_ms = int(self.time_fn() * 1000)
         genesis_ms = int(self.genesis_time) * 1000
@@ -35,16 +35,16 @@ def _milliseconds_since_genesis(self) -> Uint64:
 
     def current_slot(self) -> Slot:
         """Get the current slot number (0 if before genesis)."""
-        return Slot(self._milliseconds_since_genesis() // MILLISECONDS_PER_SLOT)
+        return Slot(self.milliseconds_since_genesis() // MILLISECONDS_PER_SLOT)
 
     def current_interval(self) -> Interval:
         """Get the current interval within the slot (0-4)."""
-        milliseconds_into_slot = self._milliseconds_since_genesis() % MILLISECONDS_PER_SLOT
+        milliseconds_into_slot = self.milliseconds_since_genesis() % MILLISECONDS_PER_SLOT
         return Interval(milliseconds_into_slot // MILLISECONDS_PER_INTERVAL)
 
     def total_intervals(self) -> Interval:
         """Get total intervals elapsed since genesis."""
-        return Interval(self._milliseconds_since_genesis() // MILLISECONDS_PER_INTERVAL)
+        return Interval(self.milliseconds_since_genesis() // MILLISECONDS_PER_INTERVAL)
 
     def current_time(self) -> Uint64:
         """Get current wall-clock time as Uint64 (Unix timestamp in seconds)."""
@@ -66,7 +66,7 @@ def seconds_until_next_interval(self) -> float:
 
         # Position within the current interval, off the shared millisecond
         # time-base that every other accessor on this clock uses.
-        milliseconds_into_interval = int(self._milliseconds_since_genesis()) % int(
+        milliseconds_into_interval = int(self.milliseconds_since_genesis()) % int(
             MILLISECONDS_PER_INTERVAL
         )
 

src/lean_spec/node/metrics/__init__.py

@@ -5,11 +5,19 @@
 https://github.com/leanEthereum/leanMetrics/blob/main/metrics.md
 """
 
+from lean_spec.node.metrics.arrival import (
+    observe_aggregate_arrival,
+    observe_attestation_arrival,
+    observe_block_arrival,
+)
 from lean_spec.node.metrics.registry import get_metrics_output, registry
 from lean_spec.node.metrics.spec_observer import PrometheusObserver
 
 __all__ = [
     "PrometheusObserver",
     "get_metrics_output",
+    "observe_aggregate_arrival",
+    "observe_attestation_arrival",
+    "observe_block_arrival",
     "registry",
 ]

src/lean_spec/node/metrics/arrival.py

@@ -0,0 +1,171 @@
+"""
+Gossip arrival timing.
+
+Every consensus message has an interval it was due in. A block belongs at
+the start of its own slot, an attestation one interval later, an aggregate at
+the aggregation interval. Measuring each arrival against that boundary turns
+"the network feels slow" into a number.
+
+The histograms record the absolute distance, so an arrival early by 200ms and
+one late by 200ms share a bucket. The counters' `position` label separates
+them: `before`, `inside`, or `after` the interval the message was due in.
+`inside` means that interval specifically. An attestation for slot 10 landing
+during slot 10's aggregation interval counts as `after`, since it missed the
+attestation-production interval it was actually due in.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from lean_spec.node.metrics.registry import registry as metrics
+from lean_spec.spec.forks.lstar.config import (
+    MILLISECONDS_PER_INTERVAL,
+    MILLISECONDS_PER_SLOT,
+)
+
+if TYPE_CHECKING:
+    from lean_spec.node.chain.clock import SlotClock
+    from lean_spec.spec.forks import Slot
+
+BLOCK_PUBLICATION_INTERVAL = 0
+"""Interval within a slot at which that slot's block is published."""
+
+ATTESTATION_PRODUCTION_INTERVAL = 1
+"""Interval within a slot at which validators attest to it."""
+
+AGGREGATION_INTERVAL = 2
+"""Interval within a slot at which aggregators publish proofs (see timeline.py)."""
+
+MAX_MEASURABLE_SLOT_DISTANCE = 256
+"""
+Slots away from the arrival beyond which a message stops being a timing sample.
+
+A gossip message carries its own slot, and that field is attacker-controlled
+until the store validates it. `Slot` is a `Uint64`, so a peer can claim slot
+2**64-1 and push a single observation of roughly 7e19 seconds into the
+histogram, which pins `_sum` for the lifetime of the process and ruins every
+average derived from it. Python bigints absorb that silently.
+
+The top bucket is 16 seconds, four slots, so this bound keeps every
+distinguishable bucket and 64 times the margin above it. Anything past it is
+malformed or a sync artifact rather than a late arrival, and gets dropped
+instead of clamped so the histogram stays a record of real messages.
+"""
+
+
+def _interval_start_ms(anchor_slot: int, interval_within_slot: int) -> int:
+    """Milliseconds since genesis at which an interval of a given slot begins."""
+    slot_start = anchor_slot * int(MILLISECONDS_PER_SLOT)
+    return slot_start + interval_within_slot * int(MILLISECONDS_PER_INTERVAL)
+
+
+def _interval_delta_ms(
+    arrival_ms: int,
+    anchor_slot: int,
+    interval_within_slot: int,
+) -> int:
+    """
+    Signed milliseconds from an interval boundary to an arrival.
+
+    A negative result means the message beat the interval it was due in.
+    """
+    return arrival_ms - _interval_start_ms(anchor_slot, interval_within_slot)
+
+
+def _latest_interval_delta_ms(arrival_ms: int, interval_within_slot: int) -> int:
+    """
+    Milliseconds since the most recent boundary of an interval, any slot.
+
+    Anchoring to the latest boundary at or before the arrival keeps the result
+    in `[0, MILLISECONDS_PER_SLOT)`, so this never goes negative.
+
+    Known limitation, shared with the ethlambda implementation this mirrors:
+    the modulo wraps rather than clamps, so a message that beat the boundary
+    it was aimed at reports as nearly a full slot late instead of as `before`.
+    A receiver whose clock lags the sender by more than the network latency
+    sees this, and the protocol budgets a whole interval of skew via
+    GOSSIP_DISPARITY_INTERVALS. The same wrap applies to any arrival in slot 0
+    ahead of the first boundary, which anchors to one that never happened.
+    Both clients agree on the number, so the series stay comparable; read a
+    hard mode near one full slot as suspected clock skew rather than as
+    genuinely late aggregation.
+    """
+    offset = interval_within_slot * int(MILLISECONDS_PER_INTERVAL)
+    return (arrival_ms - offset) % int(MILLISECONDS_PER_SLOT)
+
+
+def _is_measurable(arrival_ms: int, message_slot: int) -> bool:
+    """Whether a message's own slot is close enough to be a timing sample."""
+    arrival_slot = arrival_ms // int(MILLISECONDS_PER_SLOT)
+    return abs(message_slot - arrival_slot) <= MAX_MEASURABLE_SLOT_DISTANCE
+
+
+def _position(delta_ms: int) -> str:
+    """Classify a signed delta against the width of one interval."""
+    if delta_ms < 0:
+        return "before"
+    if delta_ms < int(MILLISECONDS_PER_INTERVAL):
+        return "inside"
+    return "after"
+
+
+def observe_block_arrival(clock: SlotClock, block_slot: Slot) -> None:
+    """
+    Record a gossip block's arrival against interval 0 of its own slot.
+
+    Call this for blocks received from a peer only. A block this node produced
+    reaches the store through the same handler, and stamping it would sample
+    this node's own build latency as if it were network timing. Blocks pulled
+    over req/resp during sync land long after they were due, and folding those
+    in would measure catch-up speed rather than gossip health.
+    """
+    arrival_ms = int(clock.milliseconds_since_genesis())
+    if not _is_measurable(arrival_ms, int(block_slot)):
+        return
+    delta_ms = _interval_delta_ms(arrival_ms, int(block_slot), BLOCK_PUBLICATION_INTERVAL)
+    metrics.lean_gossip_block_arrival_delay_seconds.observe(abs(delta_ms) / 1000.0)
+    metrics.lean_gossip_block_arrival_total.labels(position=_position(delta_ms)).inc()
+
+
+def observe_attestation_arrival(clock: SlotClock, data_slot: Slot) -> None:
+    """
+    Record a gossip attestation's arrival against interval 1 of its data slot.
+
+    Peer-received attestations only, for the reason given on
+    [`observe_block_arrival`]: a validator's own votes are signed locally and
+    would pile up in the first bucket.
+    """
+    arrival_ms = int(clock.milliseconds_since_genesis())
+    if not _is_measurable(arrival_ms, int(data_slot)):
+        return
+    delta_ms = _interval_delta_ms(arrival_ms, int(data_slot), ATTESTATION_PRODUCTION_INTERVAL)
+    metrics.lean_gossip_attestation_arrival_delay_seconds.observe(abs(delta_ms) / 1000.0)
+    metrics.lean_gossip_attestation_arrival_total.labels(position=_position(delta_ms)).inc()
+
+
+def observe_aggregate_arrival(clock: SlotClock) -> None:
+    """
+    Record a gossip aggregate's arrival, against the latest aggregation boundary.
+
+    Deliberately takes no slot. An aggregate published at the aggregation
+    interval of slot N can carry a data slot well below N when it is catching
+    up on an earlier group, and anchoring to that data slot would fill the
+    histogram with multi-slot values that are not a health problem. Taking no
+    slot also means this helper needs no guard against an implausible one,
+    unlike its block and attestation counterparts.
+
+    Peer-received aggregates only. An aggregate this node produced never
+    crosses the network, so its timing reports local proving cost rather than
+    anything about gossip, and
+    `lean_pq_sig_aggregated_signatures_building_time_seconds` already measures
+    that directly. A node that aggregates but receives nothing therefore
+    reports an empty profile here, which is the honest reading: it has no
+    gossip arrivals to describe.
+    """
+    delta_ms = _latest_interval_delta_ms(
+        int(clock.milliseconds_since_genesis()),
+        AGGREGATION_INTERVAL,
+    )
+    metrics.lean_gossip_aggregation_arrival_delay_seconds.observe(abs(delta_ms) / 1000.0)
+    metrics.lean_gossip_aggregation_arrival_total.labels(position=_position(delta_ms)).inc()

src/lean_spec/node/metrics/registry.py

@@ -45,6 +45,26 @@
 REORG_DEPTH_BUCKETS = (1, 2, 3, 5, 7, 10, 20, 30, 50, 100)
 """Block count. Reorg depths above 10 are rare and signal network issues."""
 
+GOSSIP_ARRIVAL_DELAY_BUCKETS = (0.05, 0.1, 0.2, 0.4, 0.8, 1.2, 1.6, 2.4, 4, 8, 16)
+"""
+Seconds. The upper half sits on interval and slot multiples, so 0.8, 1.6, 2.4
+and 4 answer a timing question directly: did the message make its own
+interval, the next one, its own slot? Below 0.8 the edges halve instead, to
+resolve a healthy population that all lands inside one interval.
+"""
+
+GOSSIP_ARRIVAL_POSITIONS = ("before", "inside", "after")
+"""Every arrival position reachable from a signed delta."""
+
+GOSSIP_ARRIVAL_POSITIONS_NON_NEGATIVE = ("inside", "after")
+"""
+Positions reachable when the anchor cannot follow the arrival.
+
+Aggregates anchor to the latest aggregation boundary at or before the
+arrival, so their delta never goes negative. Exporting `before` there would
+create a series that can never move.
+"""
+
 # Section labels for attestation aggregate coverage gauges. These match the
 # names printed in slot/report logs: timely, late, block, combined,
 # aggregate_start_new, proposal_payloads, proposal_gossip, and proposal_combined.
@@ -170,6 +190,20 @@ class MetricsRegistry:
     lean_connected_peers: Gauge | _NoOpMetric = _NOOP
     """Current number of active peer connections."""
 
+    # Gossip arrival timing
+    lean_gossip_block_arrival_delay_seconds: Histogram | _NoOpMetric = _NOOP
+    """Absolute distance from a gossip block's arrival to its due interval."""
+    lean_gossip_attestation_arrival_delay_seconds: Histogram | _NoOpMetric = _NOOP
+    """Absolute distance from a gossip attestation's arrival to its due interval."""
+    lean_gossip_aggregation_arrival_delay_seconds: Histogram | _NoOpMetric = _NOOP
+    """Distance from a gossip aggregate's arrival to the latest aggregation boundary."""
+    lean_gossip_block_arrival_total: Counter | _NoOpMetric = _NOOP
+    """Gossip blocks counted by arrival position."""
+    lean_gossip_attestation_arrival_total: Counter | _NoOpMetric = _NOOP
+    """Gossip attestations counted by arrival position."""
+    lean_gossip_aggregation_arrival_total: Counter | _NoOpMetric = _NOOP
+    """Gossip aggregates counted by arrival position."""
+
     def init(
         self,
         name: str = "leanspec-node",
@@ -350,6 +384,67 @@ def init(
         )
         self.lean_connected_peers.set(0)
 
+        # Gossip arrival timing (leanMetrics: Gossip Arrival Metrics)
+        #
+        # The histograms carry the absolute distance, so an early arrival and a
+        # late one of the same size share a bucket. The counters' `position`
+        # label is what separates them.
+        self.lean_gossip_block_arrival_delay_seconds = Histogram(
+            "lean_gossip_block_arrival_delay_seconds",
+            (
+                "Absolute delay between a gossip block's arrival and the start "
+                "of the interval it was due in."
+            ),
+            buckets=GOSSIP_ARRIVAL_DELAY_BUCKETS,
+            registry=reg,
+        )
+        self.lean_gossip_attestation_arrival_delay_seconds = Histogram(
+            "lean_gossip_attestation_arrival_delay_seconds",
+            (
+                "Absolute delay between a gossip attestation's arrival and the "
+                "start of the interval it was due in."
+            ),
+            buckets=GOSSIP_ARRIVAL_DELAY_BUCKETS,
+            registry=reg,
+        )
+        self.lean_gossip_aggregation_arrival_delay_seconds = Histogram(
+            "lean_gossip_aggregation_arrival_delay_seconds",
+            (
+                "Delay between a gossip aggregate's arrival and the most recent "
+                "aggregation-interval boundary at or before it."
+            ),
+            buckets=GOSSIP_ARRIVAL_DELAY_BUCKETS,
+            registry=reg,
+        )
+        self.lean_gossip_block_arrival_total = Counter(
+            "lean_gossip_block_arrival_total",
+            "Gossip blocks by arrival position relative to the interval they were due in.",
+            ["position"],
+            registry=reg,
+        )
+        self.lean_gossip_attestation_arrival_total = Counter(
+            "lean_gossip_attestation_arrival_total",
+            "Gossip attestations by arrival position relative to the interval they were due in.",
+            ["position"],
+            registry=reg,
+        )
+        self.lean_gossip_aggregation_arrival_total = Counter(
+            "lean_gossip_aggregation_arrival_total",
+            (
+                "Gossip aggregates by arrival position relative to the most "
+                "recent aggregation-interval boundary."
+            ),
+            ["position"],
+            registry=reg,
+        )
+        # Create every reachable series up front so a dashboard renders a flat
+        # zero instead of a gap before the first message of that kind lands.
+        for position in GOSSIP_ARRIVAL_POSITIONS:
+            self.lean_gossip_block_arrival_total.labels(position=position).inc(0)
+            self.lean_gossip_attestation_arrival_total.labels(position=position).inc(0)
+        for position in GOSSIP_ARRIVAL_POSITIONS_NON_NEGATIVE:
+            self.lean_gossip_aggregation_arrival_total.labels(position=position).inc(0)
+
         self._initialized = True
 
     def reset(self) -> None: