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

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