fix(node): keep slot_index canonical across side-branch imports and reorgs (#1202)

The storage interface documents get_block_root_by_slot as returning the
canonical block root at a slot, but _persist_block wrote the index for
every imported block: a side-branch import silently overwrote the
canonical entry at its slot, and nothing repaired the index when the
head moved.

Track the head the database last committed and diff it against the
store's head inside the persistence batch: slots on the new canonical
branch are (re)written, slots only the old branch filled are deleted,
and a block that did not move the head no longer touches the index.

Closes #1201

Author: adust09

packages/testing/src/consensus_testing/mocks.py

@@ -181,6 +181,9 @@ class MockForkchoiceStore:
     advance_finalized_on_block: bool = False
     """Whether processing a block advances the finalized checkpoint to it."""
 
+    advance_head_on_block: bool = True
+    """Whether processing a block moves the head to it; False imports a side branch."""
+
     received_attestations: list[SignedAttestation] = field(default_factory=list)
     """Attestations accepted so far, in arrival order."""
 
@@ -197,10 +200,11 @@ def on_block(self, _store: Store, signed_block: SignedBlock) -> MockForkchoiceSt
         """Record a block as the new head and apply the configured side effects."""
         root = hash_tree_root(signed_block.block)
         self.blocks[root] = signed_block.block
-        self.head = root
-        # No real safe-target rule here, so the head doubles as it.
-        self.safe_target = root
-        self.head_slot = signed_block.block.slot
+        if self.advance_head_on_block:
+            self.head = root
+            # No real safe-target rule here, so the head doubles as it.
+            self.safe_target = root
+            self.head_slot = signed_block.block.slot
         if self.on_block_post_state is not None:
             self.states[root] = self.on_block_post_state
         if self.advance_justified_on_block:
@@ -255,6 +259,7 @@ class RecordingSyncDatabase:
     def __init__(self) -> None:
         """Start with an empty call log."""
         self.calls: list[RecordedCall] = []
+        self.head_root: Bytes32 | None = None
 
     def _record(self, name: str, *args: object, **kwargs: object) -> None:
         self.calls.append(RecordedCall(name=name, args=args, kwargs=MappingProxyType(dict(kwargs))))
@@ -298,6 +303,14 @@ def put_block_root_by_slot(self, slot: object, root: object) -> None:
         """Record a slot to block-root index write."""
         self._record("put_block_root_by_slot", slot, root)
 
+    def delete_block_root_by_slot(self, slot: object) -> None:
+        """Record a slot-index deletion."""
+        self._record("delete_block_root_by_slot", slot)
+
+    def get_head_root(self) -> Bytes32 | None:
+        """Return the seeded head root; reads are not part of the recorded write contract."""
+        return self.head_root
+
     def put_head_root(self, root: object) -> None:
         """Record a head-root write."""
         self._record("put_head_root", root)

src/lean_spec/node/storage/database.py

@@ -83,6 +83,14 @@ def put_block_root_by_slot(self, slot: Slot, root: Bytes32) -> None:
         """Index a block root by its slot."""
         ...
 
+    def delete_block_root_by_slot(self, slot: Slot) -> None:
+        """
+        Remove the slot-index entry at a slot, if present.
+
+        Needed when a reorg leaves a formerly canonical slot empty.
+        """
+        ...
+
     # State Root Index Operations
 
     def get_block_root_by_state_root(self, state_root: Bytes32) -> Bytes32 | None:

src/lean_spec/node/storage/sqlite.py

@@ -327,6 +327,19 @@ def put_block_root_by_slot(self, slot: Slot, root: Bytes32) -> None:
                 f"Failed to write slot index for slot {slot}: {exception}"
             ) from exception
 
+    def delete_block_root_by_slot(self, slot: Slot) -> None:
+        """Remove the slot-index entry at a slot, if present."""
+        try:
+            cursor = self._connection.cursor()
+            cursor.execute(
+                f"DELETE FROM {SLOT_INDEX_TABLE_NAME} WHERE slot = ?",
+                (int(slot),),
+            )
+        except sqlite3.Error as exception:
+            raise StorageWriteError(
+                f"Failed to delete slot index for slot {slot}: {exception}"
+            ) from exception
+
     # State Root Index Operations
 
     def get_block_root_by_state_root(self, state_root: Bytes32) -> Bytes32 | None:

src/lean_spec/node/sync/service.py

@@ -149,8 +149,20 @@ class SyncService:
     until new blocks arrive.
     """
 
+    _persisted_head_root: Bytes32 | None = field(default=None)
+    """
+    Head root the database last committed, or None when untracked.
+
+    The slot-index update diffs the store's head against this to find the
+    slots whose canonical block changed.
+    """
+
     def __post_init__(self) -> None:
         """Wire sub-components and apply the genesis-start state hint."""
+        # Seed the persisted-head tracker from disk so a restarted node
+        # diffs its slot index against the chain the database last saw.
+        if self.database is not None:
+            self._persisted_head_root = self.database.get_head_root()
         # Backfill reads the store through self, so it sees each post-block reassignment.
         self._backfill = BackfillSync(
             peer_manager=self.peer_manager,
@@ -293,7 +305,12 @@ def _persist_block(self, store: Store, block: Block) -> None:
             # On restart these tell us where the chain ended last session.
             #
             # The node can resume forkchoice without re-deriving from scratch.
-            self.database.put_block_root_by_slot(block.slot, block_root)
+            #
+            # The slot index tracks the canonical chain, not the import stream:
+            # a block that did not move the head sits on a side branch (its
+            # parent was already known, so it cannot be a head ancestor) and
+            # must not displace the canonical entry at its slot.
+            self._reindex_canonical_slots(store)
             self.database.put_head_root(store.head)
             self.database.put_justified_checkpoint(store.latest_justified)
             self.database.put_finalized_checkpoint(store.latest_finalized)
@@ -307,6 +324,53 @@ def _persist_block(self, store: Store, block: Block) -> None:
                     keep_roots=frozenset({store.latest_finalized.root}),
                 )
 
+        # The tracker mirrors the database, so it moves only after a commit.
+        self._persisted_head_root = store.head
+
+    def _reindex_canonical_slots(self, store: Store) -> None:
+        """
+        Align the persisted slot index with the store's canonical chain.
+
+        Walks the old and new head branches down to their fork point: slots
+        on the new branch are (re)written, slots only the old branch filled
+        are deleted. A no-op when the head did not move.
+
+        Runs inside the caller's batch, so the index and the head pointer
+        commit together.
+        """
+        # Bytes32 rejects comparison against None, so the None case is explicit.
+        if self.database is None or (
+            self._persisted_head_root is not None and store.head == self._persisted_head_root
+        ):
+            return
+
+        new_entries: dict[Slot, Bytes32] = {}
+        stale_slots: set[Slot] = set()
+        new_root = store.head
+        old_root = self._persisted_head_root
+
+        # Lower the higher tip one parent link at a time until the branches
+        # meet (or leave the store, e.g. an untracked previous head).
+        while old_root is None or new_root != old_root:
+            new_block = store.blocks.get(new_root)
+            old_block = None if old_root is None else store.blocks.get(old_root)
+            if new_block is not None and (old_block is None or new_block.slot >= old_block.slot):
+                new_entries[new_block.slot] = new_root
+                new_root = new_block.parent_root
+            elif old_block is not None:
+                stale_slots.add(old_block.slot)
+                old_root = old_block.parent_root
+            else:
+                break
+
+        for slot in sorted(new_entries):
+            self.database.put_block_root_by_slot(slot, new_entries[slot])
+
+        # A slot the new branch refilled keeps its fresh entry; only slots
+        # left empty by the reorg lose theirs.
+        for slot in sorted(stale_slots - new_entries.keys()):
+            self.database.delete_block_root_by_slot(slot)
+
     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.

tests/node/storage/test_sqlite.py

@@ -169,6 +169,27 @@ def test_slot_index_reorg(self, db: SQLiteDatabase) -> None:
             db.put_block_root_by_slot(slot, root_b)
         assert db.get_block_root_by_slot(slot) == root_b
 
+    def test_delete_block_root_by_slot(self, db: SQLiteDatabase) -> None:
+        """Deleting a slot entry removes it while other slots survive."""
+        root_a = Bytes32(b"\x0b" * 32)
+        root_b = Bytes32(b"\x0c" * 32)
+        with db.batch_write():
+            db.put_block_root_by_slot(Slot(1), root_a)
+            db.put_block_root_by_slot(Slot(2), root_b)
+
+        with db.batch_write():
+            db.delete_block_root_by_slot(Slot(1))
+
+        assert db.get_block_root_by_slot(Slot(1)) is None
+        assert db.get_block_root_by_slot(Slot(2)) == root_b
+
+    def test_delete_nonexistent_slot_is_noop(self, db: SQLiteDatabase) -> None:
+        """Deleting an absent slot entry succeeds without effect."""
+        with db.batch_write():
+            db.delete_block_root_by_slot(Slot(999))
+
+        assert db.get_block_root_by_slot(Slot(999)) is None
+
 
 class TestStateRootIndex:
     """Tests for state root to block root index."""

tests/node/sync/test_service.py

@@ -568,6 +568,8 @@ def test_persist_skips_state_when_post_state_missing(
     ) -> None:
         """No put_state when the store has no post-state for the block root."""
         db = RecordingSyncDatabase()
+        # The persisted head matches the mock genesis, as after a real genesis write.
+        db.head_root = Bytes32.zero()
         service = create_mock_sync_service(
             peer_id,
             database=cast(Database, db),
@@ -612,6 +614,8 @@ def test_persist_writes_state_and_prunes_when_finalized_advanced(
     ) -> None:
         """Post-state indexing and pruning run when finalization is past genesis."""
         db = RecordingSyncDatabase()
+        # The persisted head matches the mock genesis, as after a real genesis write.
+        db.head_root = Bytes32.zero()
         service = create_mock_sync_service(
             peer_id,
             database=cast(Database, db),
@@ -667,6 +671,99 @@ def test_persist_writes_state_and_prunes_when_finalized_advanced(
             ),
         ]
 
+    def test_persist_skips_slot_index_for_side_branch_block(
+        self,
+        peer_id: PeerId,
+    ) -> None:
+        """A block that does not move the head leaves the slot index untouched."""
+        db = RecordingSyncDatabase()
+        db.head_root = Bytes32.zero()
+        service = create_mock_sync_service(
+            peer_id,
+            database=cast(Database, db),
+        )
+        mock_store = cast(MockForkchoiceStore, service.store)
+        mock_store.advance_head_on_block = False
+        service.state = SyncState.SYNCING
+        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)
+
+        inner = db.calls_inside_batch()
+        call_names = [call.name for call in inner]
+        assert "put_block_root_by_slot" not in call_names
+        assert "delete_block_root_by_slot" not in call_names
+        # The unchanged head pointer is still persisted with the block.
+        empty: MappingProxyType[str, object] = MappingProxyType({})
+        assert RecordedCall(name="put_head_root", args=(Bytes32.zero(),), kwargs=empty) in inner
+
+    def test_persist_reindexes_slot_index_on_reorg(
+        self,
+        peer_id: PeerId,
+    ) -> None:
+        """A head switch rewrites differing slots and deletes vacated ones."""
+        db = RecordingSyncDatabase()
+        genesis_root = Bytes32.zero()
+
+        # Old branch genesis <- B1 (slot 1) <- B2 (slot 2) is the persisted canonical chain.
+        b1 = make_signed_block(
+            slot=Slot(1),
+            proposer_index=ValidatorIndex(0),
+            parent_root=genesis_root,
+            state_root=Bytes32.zero(),
+        )
+        b1_root = hash_tree_root(b1.block)
+        b2 = make_signed_block(
+            slot=Slot(2),
+            proposer_index=ValidatorIndex(0),
+            parent_root=b1_root,
+            state_root=Bytes32.zero(),
+        )
+        b2_root = hash_tree_root(b2.block)
+        # Seed before service creation: the tracker reads the head at wiring time.
+        db.head_root = b2_root
+
+        service = create_mock_sync_service(
+            peer_id,
+            database=cast(Database, db),
+        )
+        mock_store = cast(MockForkchoiceStore, service.store)
+        service.state = SyncState.SYNCING
+        mock_store.blocks[b1_root] = b1.block
+        mock_store.blocks[b2_root] = b2.block
+        mock_store.head = b2_root
+
+        # Importing C2 (slot 2, child of genesis) reorgs the head onto the new branch.
+        c2 = make_signed_block(
+            slot=Slot(2),
+            proposer_index=ValidatorIndex(1),
+            parent_root=genesis_root,
+            state_root=Bytes32.zero(),
+        )
+        service.store = service.process_block(service.store, c2)
+        c2_root = hash_tree_root(c2.block)
+
+        inner = db.calls_inside_batch()
+        empty: MappingProxyType[str, object] = MappingProxyType({})
+        # Slot 2 is rewritten to the new branch; slot 1 has no canonical block anymore.
+        assert (
+            RecordedCall(name="put_block_root_by_slot", args=(Slot(2), c2_root), kwargs=empty)
+            in inner
+        )
+        assert (
+            RecordedCall(name="delete_block_root_by_slot", args=(Slot(1),), kwargs=empty) in inner
+        )
+        # The refilled slot is never deleted.
+        assert (
+            RecordedCall(name="delete_block_root_by_slot", args=(Slot(2),), kwargs=empty)
+            not in inner
+        )
+
 
 class TestSignedBlockServing:
     """Tests for signed-block retention and the inbound serving lookups."""