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

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