diff --git a/src/lean_spec/cli/run.py b/src/lean_spec/cli/run.py index 86065815d..b2b485a09 100644 --- a/src/lean_spec/cli/run.py +++ b/src/lean_spec/cli/run.py @@ -22,9 +22,10 @@ from lean_spec.node.networking.client import LiveNetworkEventSource from lean_spec.node.networking.gossipsub import GossipTopic from lean_spec.node.node import Node, NodeConfig -from lean_spec.spec.forks import SubnetId +from lean_spec.spec.forks import SignedBlock, Slot, SubnetId from lean_spec.spec.forks.lstar.config import ATTESTATION_COMMITTEE_COUNT from lean_spec.spec.observability import set_observer +from lean_spec.spec.ssz import Bytes32 logger = logging.getLogger(__name__) @@ -105,10 +106,23 @@ async def run_node(boot: NodeBootstrap) -> None: logger.info("Node initialized, peer_id=%s", event_source.connection_manager.peer_id) + # Inbound block serving reads the sync service's retained signed blocks. + # + # The requesting peer verifies each served block's proof on import, + # so the lookups return the retained signed blocks, never store blocks + # rewrapped with an empty proof. + async def signed_block_for_root(block_root: Bytes32) -> SignedBlock | None: + return node.sync_service.signed_block_for_root(block_root) + + async def signed_block_by_slot(slot: Slot) -> SignedBlock | None: + return node.sync_service.signed_block_by_slot(slot) + # Bring the listener and outbound dialer online in the spec-required order. await event_source.start_serving( status=anchor.initial_status, current_slot_lookup=node.clock.current_slot, + block_lookup=signed_block_for_root, + block_by_slot_lookup=signed_block_by_slot, listen_address=boot.listen_address, bootnode_multiaddrs=boot.bootnode_multiaddrs, ) diff --git a/src/lean_spec/node/networking/client/event_source/live.py b/src/lean_spec/node/networking/client/event_source/live.py index 381685d5d..f379addd2 100644 --- a/src/lean_spec/node/networking/client/event_source/live.py +++ b/src/lean_spec/node/networking/client/event_source/live.py @@ -81,6 +81,7 @@ from lean_spec.node.networking.gossipsub.types import TopicId from lean_spec.node.networking.reqresp.handler import ( REQRESP_PROTOCOL_IDS, + AsyncBlockBySlotLookup, AsyncBlockLookup, CurrentSlotLookup, ReqRespServer, @@ -320,6 +321,18 @@ def set_block_lookup(self, lookup: AsyncBlockLookup) -> None: """ self._reqresp_handler.block_lookup = lookup + def set_block_by_slot_lookup(self, lookup: AsyncBlockBySlotLookup) -> None: + """ + Set the callback for looking up canonical blocks by slot. + + Used by the inbound ReqResp handler to serve BlocksByRange requests. + + Args: + lookup: Async function that takes a Slot and returns the + canonical SignedBlock if available, None otherwise. + """ + self._reqresp_handler.block_by_slot_lookup = lookup + def set_current_slot_lookup(self, lookup: CurrentSlotLookup) -> None: """ Set the callback returning the node's current slot. @@ -367,6 +380,8 @@ async def start_serving( *, status: Status, current_slot_lookup: CurrentSlotLookup, + block_lookup: AsyncBlockLookup, + block_by_slot_lookup: AsyncBlockBySlotLookup, listen_address: str | None, bootnode_multiaddrs: Sequence[str], ) -> None: @@ -376,7 +391,7 @@ async def start_serving( Five steps, each a precondition for the next: 1. Set the Status the responder serves. - 2. Wire the current-slot lookup the range queries depend on. + 2. Wire the block and current-slot lookups the responder depends on. 3. Dial bootnodes best-effort, since a peerless honest node remains valid. 4. Bind the listener with a short bind-error probe window. 5. Start gossipsub last so the heartbeat reaches reachable peers only. @@ -384,16 +399,20 @@ async def start_serving( Args: status: Initial finalized and head checkpoints the responder serves. current_slot_lookup: Wall-clock-to-slot callback for range bounds. + block_lookup: Callback serving signed blocks by root. + block_by_slot_lookup: Callback serving canonical signed blocks by slot. listen_address: Multiaddr to bind for inbound connections, or None for dial-only. bootnode_multiaddrs: Pre-resolved outbound peers. Raises: OSError: If the listener fails to bind within the probe window. """ - # Status and current-slot lookup must be set before the responder serves. - # Without them, range queries return SERVER_ERROR. + # Status and lookups must be set before the responder serves. + # Without them, block and range queries return SERVER_ERROR. self.set_status(status) self.set_current_slot_lookup(current_slot_lookup) + self.set_block_lookup(block_lookup) + self.set_block_by_slot_lookup(block_by_slot_lookup) # Dial and listen each clear the stop event internally. # Clearing it here covers the no-bootnodes, no-listen case. diff --git a/src/lean_spec/node/sync/service.py b/src/lean_spec/node/sync/service.py index b3e798494..b49ed5da5 100644 --- a/src/lean_spec/node/sync/service.py +++ b/src/lean_spec/node/sync/service.py @@ -13,6 +13,7 @@ from lean_spec.node.chain.clock import SlotClock from lean_spec.node.metrics import 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 from lean_spec.node.storage import Database @@ -133,6 +134,21 @@ class SyncService: _pending_block_aggregates: list[SignedAggregatedAttestation] = field(default_factory=list) """Aggregates recovered from processed blocks, queued for the aggregator to publish.""" + _signed_blocks_for_serving: dict[Bytes32, SignedBlock] = field(default_factory=dict) + """ + Signed blocks retained to serve inbound block requests, keyed by block root. + + The forkchoice store keeps unsigned blocks only. + A served block must carry its original proof. + The requesting peer verifies that proof on import. + + Bounded by the serving history window. + Blocks below the window can never be served again, so they are pruned. + + In-memory only: after a restart the node serves nothing + until new blocks arrive. + """ + def __post_init__(self) -> None: """Wire sub-components and apply the genesis-start state hint.""" # Backfill reads the store through self, so it sees each post-block reassignment. @@ -205,6 +221,13 @@ def ancestors(start: Bytes32) -> set[Bytes32]: # We only count blocks that pass validation and update the store. self._blocks_processed += 1 + # Retain the signed block so peers can request it back. + # + # The store keeps only the unsigned block, which cannot be served: + # the requesting peer verifies the proof when importing the block. + self._signed_blocks_for_serving[hash_tree_root(block.block)] = block + self._prune_signed_blocks_below_serving_window() + # Aggregators recover per-attestation proofs from each processed block. # They queue the recovered proofs for re-broadcast. # Non-aggregators rely on the gossip path instead. @@ -284,6 +307,42 @@ def _persist_block(self, store: Store, block: Block) -> None: keep_roots=frozenset({store.latest_finalized.root}), ) + def _prune_signed_blocks_below_serving_window(self) -> None: + """Drop retained signed blocks that fell out of the serving history window.""" + # The responder refuses range requests below the sliding window floor. + # A block below the floor can never be served again, so retaining it is waste. + current_slot = self.clock.current_slot() + if current_slot < Slot(MIN_SLOTS_FOR_BLOCK_REQUESTS): + return + window_floor = current_slot - Slot(MIN_SLOTS_FOR_BLOCK_REQUESTS) + self._signed_blocks_for_serving = { + block_root: signed_block + for block_root, signed_block in self._signed_blocks_for_serving.items() + if signed_block.block.slot >= window_floor + } + + def signed_block_for_root(self, block_root: Bytes32) -> SignedBlock | None: + """Return the retained signed block for a root, or None when not retained.""" + return self._signed_blocks_for_serving.get(block_root) + + def signed_block_by_slot(self, slot: Slot) -> SignedBlock | None: + """ + Return the retained signed block on the canonical chain at a slot. + + Walks parent links from the head until the slot is reached. + + Returns None for an empty slot, a slot above the head, and a canonical + block whose signed form was never retained. + """ + block_root = self.store.head + block = self.store.blocks.get(block_root) + while block is not None and block.slot > slot: + block_root = block.parent_root + block = self.store.blocks.get(block_root) + if block is None or block.slot != slot: + return None + return self._signed_blocks_for_serving.get(block_root) + def has_root(self, root: Bytes32) -> bool: """Return True if the block root is present in the current store.""" return root in self.store.blocks diff --git a/tests/node/networking/client/event_source/test_live.py b/tests/node/networking/client/event_source/test_live.py index 664e932f6..ea9e6c196 100644 --- a/tests/node/networking/client/event_source/test_live.py +++ b/tests/node/networking/client/event_source/test_live.py @@ -146,6 +146,26 @@ async def mock_lookup(root: Bytes32) -> SignedBlock | None: assert es._reqresp_handler.block_lookup is mock_lookup +class TestLiveNetworkEventSourceSetBlockBySlotLookup: + """ + Canonical block-by-slot lookup callback registration. + + The req/resp handler needs a callback to retrieve canonical blocks + by slot when peers send BlocksByRange requests. + """ + + def test_set_block_by_slot_lookup_propagates_to_handler(self) -> None: + """Sets the block-by-slot lookup callback on the reqresp handler.""" + es = _make_event_source() + + async def mock_lookup(slot: Slot) -> SignedBlock | None: + return None + + es.set_block_by_slot_lookup(mock_lookup) + + assert es._reqresp_handler.block_by_slot_lookup is mock_lookup + + class TestLiveNetworkEventSourceSubscribeGossipTopic: """ Gossip topic subscription delegation to the gossipsub behavior. diff --git a/tests/node/sync/test_service.py b/tests/node/sync/test_service.py index b92119b49..5616eaf5a 100644 --- a/tests/node/sync/test_service.py +++ b/tests/node/sync/test_service.py @@ -18,7 +18,9 @@ make_signed_block, ) from consensus_testing.keys import XmssKeyManager +from lean_spec.node.chain.clock import SlotClock from lean_spec.node.networking import PeerId +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.storage.database import Database from lean_spec.node.sync.config import MAX_PENDING_ATTESTATIONS @@ -34,6 +36,7 @@ ValidatorIndex, ) from lean_spec.spec.forks.lstar import Store +from lean_spec.spec.forks.lstar.config import SECONDS_PER_SLOT from lean_spec.spec.forks.lstar.containers import ( AttestationData, MultiMessageAggregate, @@ -43,7 +46,7 @@ SingleMessageAggregate, ) from lean_spec.spec.forks.lstar.spec import LstarSpec -from lean_spec.spec.ssz import Bytes32 +from lean_spec.spec.ssz import Bytes32, Uint64 def make_store_with_attestation_data( @@ -665,6 +668,103 @@ def test_persist_writes_state_and_prunes_when_finalized_advanced( ] +class TestSignedBlockServing: + """Tests for signed-block retention and the inbound serving lookups.""" + + def test_process_block_retains_signed_block_for_root(self, peer_id: PeerId) -> None: + """A processed block is retrievable by its root with the proof intact.""" + service = create_mock_sync_service(peer_id) + genesis_root = service.store.head + block = make_signed_block( + slot=Slot(1), + proposer_index=ValidatorIndex(0), + parent_root=genesis_root, + state_root=Bytes32.zero(), + ) + service.store = service.process_block(service.store, block) + + block_root = hash_tree_root(block.block) + assert service.signed_block_for_root(block_root) == block + + def test_signed_block_for_root_returns_none_for_unknown_root(self, peer_id: PeerId) -> None: + """An unknown root yields no block.""" + service = create_mock_sync_service(peer_id) + + assert service.signed_block_for_root(Bytes32(b"\x2a" * 32)) is None + + def test_signed_block_by_slot_returns_canonical_block(self, peer_id: PeerId) -> None: + """The canonical block at a filled slot is served with the proof intact.""" + service = create_mock_sync_service(peer_id) + genesis_root = service.store.head + block = make_signed_block( + slot=Slot(1), + proposer_index=ValidatorIndex(0), + parent_root=genesis_root, + state_root=Bytes32.zero(), + ) + service.store = service.process_block(service.store, block) + + assert service.signed_block_by_slot(Slot(1)) == block + + def test_signed_block_by_slot_returns_none_for_empty_slot(self, peer_id: PeerId) -> None: + """A slot the canonical chain skipped yields no block.""" + service = create_mock_sync_service(peer_id) + genesis_root = service.store.head + block = make_signed_block( + slot=Slot(2), + proposer_index=ValidatorIndex(0), + parent_root=genesis_root, + state_root=Bytes32.zero(), + ) + service.store = service.process_block(service.store, block) + + assert service.signed_block_by_slot(Slot(1)) is None + + def test_signed_block_by_slot_returns_none_above_head(self, peer_id: PeerId) -> None: + """A slot above the head yields no block.""" + service = create_mock_sync_service(peer_id) + genesis_root = service.store.head + block = make_signed_block( + slot=Slot(1), + proposer_index=ValidatorIndex(0), + parent_root=genesis_root, + state_root=Bytes32.zero(), + ) + service.store = service.process_block(service.store, block) + + assert service.signed_block_by_slot(Slot(5)) is None + + def test_retained_blocks_pruned_below_serving_window(self, peer_id: PeerId) -> None: + """A retained block below the sliding history window is dropped.""" + service = create_mock_sync_service(peer_id) + # Clock far past genesis: current slot 3610 puts the window floor at slot 10. + current_slot_past_window = MIN_SLOTS_FOR_BLOCK_REQUESTS + 10 + service.clock = SlotClock( + genesis_time=Uint64(0), + time_fn=lambda: float(current_slot_past_window * int(SECONDS_PER_SLOT)), + ) + genesis_root = service.store.head + block_below_window = make_signed_block( + slot=Slot(1), + proposer_index=ValidatorIndex(0), + parent_root=genesis_root, + state_root=Bytes32.zero(), + ) + service.store = service.process_block(service.store, block_below_window) + block_inside_window = make_signed_block( + slot=Slot(current_slot_past_window), + proposer_index=ValidatorIndex(0), + parent_root=hash_tree_root(block_below_window.block), + state_root=Bytes32.zero(), + ) + service.store = service.process_block(service.store, block_inside_window) + + below_window_root = hash_tree_root(block_below_window.block) + inside_window_root = hash_tree_root(block_inside_window.block) + assert service.signed_block_for_root(below_window_root) is None + assert service.signed_block_for_root(inside_window_root) == block_inside_window + + class TestPublishAggregatedAttestation: """Tests for aggregated attestation publish wiring."""