From c3094cf7b6745418ed970afcbc09d012ab3e978f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tom=C3=A1s=20Gr=C3=BCner?= <47506558+MegaRedHand@users.noreply.github.com> Date: Thu, 6 Aug 2026 12:00:00 -0300 Subject: [PATCH] feat(metrics): add 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. Nothing measured how far off a message actually landed, so a slow network looked the same as aggregators that were slow to prove. Six metrics cover it. Three histograms carry the absolute distance from the boundary, and three counters label each arrival by position: before, inside, or after the interval it was due in. Absolute distance keeps an early arrival and a late one of the same size in a single bucket, and the position label is what tells them apart. Blocks anchor to interval 0 of their own slot, attestations to interval 1 of their data slot. Aggregates anchor instead to the most recent aggregation boundary at or before the arrival. An aggregate catching up on an earlier group carries a data slot several slots back, and anchoring there would report a multi-slot delay that is not a health problem. The boundary anchor also bounds the value to one slot and rules out `before`, so the aggregate counter exports two series rather than three. It wraps rather than clamps, so a message that beat its boundary reads as nearly a slot late instead of as early, a limitation the helper documents. Only messages that crossed the network are sampled, all three kinds. A proposer's own block and a validator's own votes come back through the same handlers with no peer id (see `publish_and_process_block` in node.py), and stamping those would sample local build latency as if it were network timing. Aggregates this node produced are left out for the same reason: their timing is proving cost, which `lean_pq_sig_aggregated_signatures_building_time_seconds` already measures directly. A message's slot is attacker-controlled until the store validates it, and `Slot` is a `Uint64`. One block claiming slot 2**64-1 would push a single observation of roughly 7e19 seconds into the histogram and pin `_sum` for the life of the process, so arrivals further than MAX_MEASURABLE_SLOT_DISTANCE from the clock are dropped rather than measured. The aggregate metric takes no slot and needs no such guard. `SlotClock._milliseconds_since_genesis` becomes public: the arrival helpers need millisecond resolution, which the slot and interval accessors have already rounded away. Names, types, and buckets match the leanMetrics registry entry. --- src/lean_spec/node/chain/clock.py | 10 +- src/lean_spec/node/metrics/__init__.py | 8 ++ src/lean_spec/node/metrics/arrival.py | 171 +++++++++++++++++++++++++ src/lean_spec/node/metrics/registry.py | 95 ++++++++++++++ src/lean_spec/node/sync/service.py | 31 ++++- 5 files changed, 309 insertions(+), 6 deletions(-) create mode 100644 src/lean_spec/node/metrics/arrival.py diff --git a/src/lean_spec/node/chain/clock.py b/src/lean_spec/node/chain/clock.py index 0095f5fb0..cc2e3cd09 100644 --- a/src/lean_spec/node/chain/clock.py +++ b/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 ) diff --git a/src/lean_spec/node/metrics/__init__.py b/src/lean_spec/node/metrics/__init__.py index b6e8ed978..0f925de7c 100644 --- a/src/lean_spec/node/metrics/__init__.py +++ b/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", ] diff --git a/src/lean_spec/node/metrics/arrival.py b/src/lean_spec/node/metrics/arrival.py new file mode 100644 index 000000000..376d11453 --- /dev/null +++ b/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() diff --git a/src/lean_spec/node/metrics/registry.py b/src/lean_spec/node/metrics/registry.py index 70521019f..76f9707ea 100644 --- a/src/lean_spec/node/metrics/registry.py +++ b/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: diff --git a/src/lean_spec/node/sync/service.py b/src/lean_spec/node/sync/service.py index b49ed5da5..9aecfa4f7 100644 --- a/src/lean_spec/node/sync/service.py +++ b/src/lean_spec/node/sync/service.py @@ -12,7 +12,12 @@ from dataclasses import dataclass, field from lean_spec.node.chain.clock import SlotClock -from lean_spec.node.metrics import registry as metrics +from lean_spec.node.metrics import ( + observe_aggregate_arrival, + observe_attestation_arrival, + observe_block_arrival, + registry as metrics, +) from lean_spec.node.networking.config import MIN_SLOTS_FOR_BLOCK_REQUESTS from lean_spec.node.networking.reqresp.message import Status from lean_spec.node.networking.transport.peer_id import PeerId @@ -392,6 +397,18 @@ async def on_gossip_block( self.state.name, ) + # Stamp arrival before import, so the sample carries network timing + # rather than network timing plus however long this node took to + # validate and apply the block. Invalid blocks therefore count too, + # unlike the valid/invalid counters, which is the intent: a block that + # arrives late and then fails validation still arrived late. + # + # A missing peer means this node produced the block (see + # `publish_and_process_block` in node.py), and its own build latency is + # not gossip timing. + if peer_id is not None: + observe_block_arrival(self.clock, block.block.slot) + if self._head_sync is None: raise RuntimeError("HeadSync not initialized") @@ -446,6 +463,12 @@ async def on_gossip_attestation( validator_index, ) + # Peer-received votes only. This node's own attestations arrive here + # through `publish_and_process_attestation` with no peer, moments after + # signing, and would bunch in the first bucket. + if peer_id is not None: + observe_attestation_arrival(self.clock, slot) + # Aggregator role requires both an active validator and operator opt-in. is_aggregator_role = self.store.validator_index is not None and self.is_aggregator @@ -502,6 +525,12 @@ async def on_gossip_aggregated_attestation( slot, ) + # Peer-received aggregates only, matching the other two handlers. Today + # every caller supplies a peer, but this handler accepts None, and an + # aggregate this node built would report proving cost, not gossip. + if peer_id is not None: + observe_aggregate_arrival(self.clock) + # The store: # - verifies the aggregated signature, # - credits weight to every validator covered by the aggregate.