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.