From 35cfcd1a66bd6eabb0e8cbdf36b07d99722915f9 Mon Sep 17 00:00:00 2001 From: adust09 Date: Sat, 15 Aug 2026 15:11:07 +0900 Subject: [PATCH] fix(node): keep slot_index canonical across side-branch imports and reorgs 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 --- .../testing/src/consensus_testing/mocks.py | 21 +++- src/lean_spec/node/storage/database.py | 8 ++ src/lean_spec/node/storage/sqlite.py | 13 +++ src/lean_spec/node/sync/service.py | 66 ++++++++++++- tests/node/storage/test_sqlite.py | 21 ++++ tests/node/sync/test_service.py | 97 +++++++++++++++++++ 6 files changed, 221 insertions(+), 5 deletions(-) diff --git a/packages/testing/src/consensus_testing/mocks.py b/packages/testing/src/consensus_testing/mocks.py index d46ccdf56..b90c24d6b 100644 --- a/packages/testing/src/consensus_testing/mocks.py +++ b/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) diff --git a/src/lean_spec/node/storage/database.py b/src/lean_spec/node/storage/database.py index 6b8ab5f5e..2a2b26913 100644 --- a/src/lean_spec/node/storage/database.py +++ b/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: diff --git a/src/lean_spec/node/storage/sqlite.py b/src/lean_spec/node/storage/sqlite.py index 97ad14b2a..4230c342c 100644 --- a/src/lean_spec/node/storage/sqlite.py +++ b/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: diff --git a/src/lean_spec/node/sync/service.py b/src/lean_spec/node/sync/service.py index b49ed5da5..458cfb2ab 100644 --- a/src/lean_spec/node/sync/service.py +++ b/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. diff --git a/tests/node/storage/test_sqlite.py b/tests/node/storage/test_sqlite.py index 965a8d5c9..8f0408da0 100644 --- a/tests/node/storage/test_sqlite.py +++ b/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.""" diff --git a/tests/node/sync/test_service.py b/tests/node/sync/test_service.py index 5616eaf5a..5777cee54 100644 --- a/tests/node/sync/test_service.py +++ b/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."""