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