fix(node): wire inbound ReqResp block serving into the running node (#1193)

The block lookup callbacks on the inbound request handler were never
assigned outside tests, so a running node answered every BlocksByRoot
and BlocksByRange request with SERVER_ERROR while its own client side
kept issuing range requests no leanSpec peer could answer.

The forkchoice store keeps unsigned blocks only, and a requesting peer
verifies each served block's proof on import, so serving store blocks
rewrapped with an empty proof would be rejected at the receiver. The
sync service now retains processed signed blocks inside the sliding
serving-history window and exposes root and canonical-slot lookups.
The boot sequence wires those lookups into the event source before it
starts serving.

Closes #1192

Author: adust09

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,
     )

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,24 +391,28 @@ 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.
 
         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.

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

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.

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."""