feat(api): restore /lean/v0/blocks/finalized for checkpoint-sync anchor block

packages/testing/src/consensus_testing/__init__.py

@@ -131,6 +131,8 @@
     make_signed_block,
     make_test_block,
     make_test_status,
+    signed_block_with_empty_proof,
+    store_backed_signed_block_getter,
 )
 
 StateTransitionTestFiller = Callable[..., StateTransitionFixture]
@@ -207,6 +209,8 @@
     "make_signed_block",
     "make_test_block",
     "make_test_status",
+    "signed_block_with_empty_proof",
+    "store_backed_signed_block_getter",
     # Unit-test fakes
     "MockEventSource",
     "MockForkchoiceStore",

packages/testing/src/consensus_testing/values.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
+from typing import Callable
+
 from consensus_testing.keys import create_dummy_signature
 from lean_spec.node.networking.reqresp.message import Status
 from lean_spec.spec.forks import Checkpoint, Slot, ValidatorIndex
+from lean_spec.spec.forks.lstar import Store
 from lean_spec.spec.forks.lstar.containers import (
     AggregatedAttestations,
     AttestationData,
@@ -20,22 +23,52 @@
 """Validator index a node owns by default in unit tests."""
 
 
+def signed_block_with_empty_proof(block: Block) -> SignedBlock:
+    """
+    Wrap an unsigned block in an empty proof.
+
+    The fork-choice store retains only unsigned blocks.
+    A genesis or anchor block that no proposer ever signed carries an empty proof.
+    """
+    return SignedBlock(
+        block=block,
+        proof=MultiMessageAggregate(proof=ByteList512KiB(data=b"")),
+    )
+
+
+def store_backed_signed_block_getter(
+    store: Store,
+) -> Callable[[Bytes32], SignedBlock | None]:
+    """
+    Build a signed-block lookup over a store's unsigned blocks.
+
+    Returns None for an unknown root, mirroring a node that lacks the block.
+    """
+
+    def signed_block_for(root: Bytes32) -> SignedBlock | None:
+        block = store.blocks.get(root)
+        if block is None:
+            return None
+        return signed_block_with_empty_proof(block)
+
+    return signed_block_for
+
+
 def make_signed_block(
     slot: Slot,
     proposer_index: ValidatorIndex,
     parent_root: Bytes32,
     state_root: Bytes32,
 ) -> SignedBlock:
     """Build a signed block with an empty proof for structural tests."""
-    return SignedBlock(
-        block=Block(
+    return signed_block_with_empty_proof(
+        Block(
             slot=slot,
             proposer_index=proposer_index,
             parent_root=parent_root,
             state_root=state_root,
             body=BlockBody(attestations=AggregatedAttestations(data=[])),
-        ),
-        proof=MultiMessageAggregate(proof=ByteList512KiB(data=b"")),
+        )
     )
 
 

src/lean_spec/node/anchor.py

@@ -5,7 +5,7 @@
 Two sources land on the same return shape:
 
 - Genesis: synthesise the store from the genesis validator set.
-- Checkpoint: fetch a finalized state from a peer and build the store from it.
+- Checkpoint: fetch a finalized block and state from a peer and build the store.
 
 Once the store exists the protocol cannot tell the two sources apart.
 """
@@ -19,14 +19,12 @@
 from lean_spec.node.networking.reqresp.message import Status
 from lean_spec.node.sync.checkpoint_sync import (
     CheckpointSyncError,
+    fetch_finalized_block,
     fetch_finalized_state,
     verify_checkpoint_state,
 )
 from lean_spec.spec.crypto.merkleization import hash_tree_root
 from lean_spec.spec.forks import (
-    AggregatedAttestations,
-    Block,
-    BlockBody,
     Checkpoint,
     ForkProtocol,
     Slot,
@@ -81,10 +79,12 @@ async def from_checkpoint(
         validator_index: ValidatorIndex | None,
     ) -> Anchor:
         """
-        Build an anchor by fetching a finalized state from a peer.
+        Build an anchor by fetching a finalized block and state from a peer.
 
         The fetched state replaces the genesis validator set.
         Deposits and exits since genesis are already baked into it.
+        The fetched block anchors the store at the same finalized root the
+        network agrees on; a source that cannot serve it cannot be used.
 
         Args:
             url: HTTP endpoint of the node serving the checkpoint state.
@@ -95,7 +95,14 @@ async def from_checkpoint(
         Raises:
             CheckpointSyncError: For every failure mode covering transport,
                 structural verification, and genesis-time mismatch.
+                Also raised when the fetched block and state do not pair.
+                That case is retryable: the source advanced finalization
+                between the two requests.
         """
+        # The block comes first: it is small, so an incapable source fails
+        # fast before the multi-megabyte state download starts.
+        signed_block = await fetch_finalized_block(url)
+
         state = await fetch_finalized_state(url, fork.state_class)
 
         # Catches a corrupt download before it contaminates the forkchoice store.
@@ -110,24 +117,17 @@ async def from_checkpoint(
                 f"local={genesis.genesis_time}"
             )
 
-        # Reconstruct the anchor block from the header embedded in the state.
-        # A header stored before its post-state root carries a zero placeholder;
-        # in that case we recompute the root from the state itself.
-        # Fork choice only needs identity and lineage, so the body is left empty.
-        header = state.latest_block_header
-        state_root = (
-            header.state_root if header.state_root != Bytes32.zero() else hash_tree_root(state)
-        )
-        anchor_block = Block(
-            slot=header.slot,
-            proposer_index=header.proposer_index,
-            parent_root=header.parent_root,
-            state_root=state_root,
-            body=BlockBody(attestations=AggregatedAttestations(data=[])),
-        )
+        # Both fetches read the snapshot at the finalized root.
+        # A pairing mismatch means finalization advanced between the two
+        # requests; refetching is the fix.
+        if signed_block.block.state_root != hash_tree_root(state):
+            raise CheckpointSyncError(
+                "anchor block / state mismatch; "
+                "source advanced finalization between requests, retry"
+            )
 
         # The protocol return type is structural, but only one concrete store ships.
-        store = cast(Store, fork.create_store(state, anchor_block, validator_index))
+        store = cast(Store, fork.create_store(state, signed_block.block, validator_index))
 
         return cls(
             validators=state.validators,

src/lean_spec/node/api/context.py

@@ -8,7 +8,8 @@
 
 from aiohttp import web
 
-from lean_spec.spec.forks import LstarSpec, Store
+from lean_spec.spec.forks import LstarSpec, SignedBlock, Store
+from lean_spec.spec.ssz import Bytes32
 
 
 class AggregatorRoleControl(Protocol):
@@ -30,6 +31,9 @@ class ApiContext:
     aggregator_role_control: AggregatorRoleControl | None
     """Holder of the aggregator flag, or None when aggregator control is unwired."""
 
+    signed_block_getter: Callable[[Bytes32], SignedBlock | None] | None
+    """Callable returning the signed block for a block root, or None when unwired."""
+
     def require_store(self) -> Store:
         """
         Return the live store, or raise 503 when the node has no store yet.
@@ -46,3 +50,9 @@ def require_aggregator_role_control(self) -> AggregatorRoleControl:
         if self.aggregator_role_control is None:
             raise web.HTTPServiceUnavailable(reason="Aggregator role control not available")
         return self.aggregator_role_control
+
+    def require_signed_block_getter(self) -> Callable[[Bytes32], SignedBlock | None]:
+        """Return the signed-block source, or raise 503 when it is unwired."""
+        if self.signed_block_getter is None:
+            raise web.HTTPServiceUnavailable(reason="Signed block source not configured")
+        return self.signed_block_getter

src/lean_spec/node/api/handlers.py

@@ -129,6 +129,30 @@ async def finalized_state(self, request: web.Request) -> web.Response:
 
         return web.Response(body=ssz_bytes, content_type="application/octet-stream")
 
+    async def finalized_block(self, request: web.Request) -> web.Response:
+        """
+        Return the finalized signed block as SSZ bytes.
+
+        Raises:
+            HTTPNotFound: The source has no block for the finalized root.
+            HTTPInternalServerError: Encoding the signed block failed.
+        """
+        store = self.context.require_store()
+        signed_block_getter = self.context.require_signed_block_getter()
+
+        signed_block = signed_block_getter(store.latest_finalized.root)
+        if signed_block is None:
+            raise web.HTTPNotFound(reason="Finalized signed block not available")
+
+        # Encoding a full block is CPU-heavy, so run it off the event loop.
+        try:
+            ssz_bytes = await asyncio.to_thread(signed_block.encode_bytes)
+        except Exception as exception:
+            logger.error("Failed to encode signed block: %s", exception)
+            raise web.HTTPInternalServerError(reason="Encoding failed") from exception
+
+        return web.Response(body=ssz_bytes, content_type="application/octet-stream")
+
     async def aggregator_status(self, request: web.Request) -> web.Response:
         """Report whether the node is acting as an aggregator."""
         aggregator_role_control = self.context.require_aggregator_role_control()

src/lean_spec/node/api/server.py

@@ -11,7 +11,8 @@
 
 from lean_spec.node.api.context import AggregatorRoleControl, ApiContext
 from lean_spec.node.api.handlers import ApiHandlers
-from lean_spec.spec.forks import LstarSpec, Store
+from lean_spec.spec.forks import LstarSpec, SignedBlock, Store
+from lean_spec.spec.ssz import Bytes32
 
 logger = logging.getLogger(__name__)
 
@@ -40,6 +41,9 @@ class ApiServer:
     store_getter: Callable[[], Store | None] | None = None
     """Callable that returns the current Store instance."""
 
+    signed_block_getter: Callable[[Bytes32], SignedBlock | None] | None = None
+    """Optional callable returning the signed block for a block root."""
+
     aggregator_role_control: AggregatorRoleControl | None = None
     """Optional runtime accessor for the node's aggregator role."""
 
@@ -57,6 +61,19 @@ def store(self) -> Store | None:
         """Get the current Store instance."""
         return self.store_getter() if self.store_getter else None
 
+    @property
+    def bound_port(self) -> int:
+        """
+        TCP port the running server actually listens on.
+
+        Resolves the OS-assigned port when the configuration requested port 0.
+        """
+        if self._runner is None:
+            raise RuntimeError("API server is not running")
+        # The runner exposes one socket address per listening site.
+        # The port is the second element of the first address.
+        return int(self._runner.addresses[0][1])
+
     async def start(self) -> None:
         """Start the API server in the background."""
         app = web.Application()
@@ -67,6 +84,7 @@ async def start(self) -> None:
             spec=self.spec,
             store_getter=self.store_getter,
             aggregator_role_control=self.aggregator_role_control,
+            signed_block_getter=self.signed_block_getter,
         )
         handlers = ApiHandlers(context)
 
@@ -76,6 +94,7 @@ async def start(self) -> None:
             [
                 web.get("/lean/v0/health", handlers.health),
                 web.get("/lean/v0/states/finalized", handlers.finalized_state),
+                web.get("/lean/v0/blocks/finalized", handlers.finalized_block),
                 web.get("/lean/v0/checkpoints/justified", handlers.justified_checkpoint),
                 web.get("/lean/v0/fork_choice", handlers.fork_choice),
                 web.get("/metrics", handlers.metrics),
@@ -90,7 +109,7 @@ async def start(self) -> None:
         self._site = web.TCPSite(self._runner, self.config.host, self.config.port)
         await self._site.start()
 
-        logger.info("API server listening on %s:%d", self.config.host, self.config.port)
+        logger.info("API server listening on %s:%d", self.config.host, self.bound_port)
 
     async def run(self) -> None:
         """Run the API server until it is asked to stop."""

src/lean_spec/node/node.py

@@ -44,7 +44,8 @@
     Validators,
 )
 from lean_spec.spec.forks.lstar.config import ATTESTATION_COMMITTEE_COUNT
-from lean_spec.spec.ssz import Bytes32, Uint64
+from lean_spec.spec.forks.lstar.containers import MultiMessageAggregate
+from lean_spec.spec.ssz import ByteList512KiB, Bytes32, Uint64
 
 logger = logging.getLogger(__name__)
 
@@ -310,13 +311,29 @@ def from_genesis(cls, config: NodeConfig) -> Node:
         # Create API server if configured
         api_server: ApiServer | None = None
         if config.api_config is not None:
+
+            def signed_block_for_root(block_root: Bytes32) -> SignedBlock | None:
+                # The store and database retain only unsigned blocks.
+                # Wrap the looked-up block in an empty proof to serve the
+                # checkpoint-sync anchor pair.
+                # The receiving peer pairs the block with the finalized state
+                # and never verifies this proof, so an empty one suffices.
+                block = sync_service.store.blocks.get(block_root)
+                if block is None:
+                    return None
+                return SignedBlock(
+                    block=block,
+                    proof=MultiMessageAggregate(proof=ByteList512KiB(data=b"")),
+                )
+
             # The admin API reads and mutates the sync service aggregator flag,
             # letting operators rotate the role at runtime without a restart.
             # Store getter captures sync_service to get the live store.
             api_server = ApiServer(
                 config=config.api_config,
                 spec=fork,
                 store_getter=lambda: sync_service.store,
+                signed_block_getter=signed_block_for_root,
                 aggregator_role_control=sync_service,
             )