diff --git a/packages/testing/src/consensus_testing/mocks.py b/packages/testing/src/consensus_testing/mocks.py index e14cf1805..d46ccdf56 100644 --- a/packages/testing/src/consensus_testing/mocks.py +++ b/packages/testing/src/consensus_testing/mocks.py @@ -321,6 +321,7 @@ def create_mock_sync_service( *, database: Database | None = None, genesis_start: bool = False, + is_aggregator: bool = False, ) -> SyncService: """Build a sync service backed by a fake store, spec, and network.""" peer_manager = PeerManager() @@ -338,4 +339,5 @@ def create_mock_sync_service( spec=cast(LstarSpec, forkchoice_double), database=database, genesis_start=genesis_start, + is_aggregator=is_aggregator, ) diff --git a/src/lean_spec/node/sync/service.py b/src/lean_spec/node/sync/service.py index feca86cee..b3e798494 100644 --- a/src/lean_spec/node/sync/service.py +++ b/src/lean_spec/node/sync/service.py @@ -35,6 +35,7 @@ Slot, SpecRejectionError, Store, + ValidatorIndex, ) from lean_spec.spec.forks.lstar.containers import ( AggregationError, @@ -204,10 +205,15 @@ def ancestors(start: Bytes32) -> set[Bytes32]: # We only count blocks that pass validation and update the store. self._blocks_processed += 1 - # Recover per-attestation proofs from every processed block. - # Queue them for publishing only when this node is an aggregator. - new_store, aggregates = self._deconstruct_block_into_store(new_store, block) - if self.is_aggregator: + # Aggregators recover per-attestation proofs from each processed block. + # They queue the recovered proofs for re-broadcast. + # Non-aggregators rely on the gossip path instead. + # The recovery is skipped while syncing. + # Historical blocks flood this path during sync. + # The justified anchor is still moving during sync. + # Recovered votes would then not match a live head. + if self.is_aggregator and self.state == SyncState.SYNCED: + new_store, aggregates = self._deconstruct_block_into_store(new_store, block) self._pending_block_aggregates.extend(aggregates) # Write-through persistence: synchronous and optional. @@ -587,18 +593,31 @@ def _deconstruct_block_into_store( } aggregates: list[SignedAggregatedAttestation] = [] + # A future block built on this head can only pack votes whose source + # is this head's justified checkpoint. + # A vote with an older source can no longer advance justification. + # Such a vote is intentionally dropped. + # A vote with a newer or unrelated source could never have been anchored on this chain. + # Only votes matching this head's justified checkpoint are worth recovering. + # Read this checkpoint from the passed store, + # which holds the post-state of the block just imported. + head_state = store.states.get(store.head) + if head_state is None: + return store, [] + head_state_justified_checkpoint = head_state.latest_justified + for attestation in block_attestations: attestation_data = attestation.data - # Skip targets at or behind justified, which can no longer advance justification. - if attestation_data.target.slot <= store.latest_justified.slot: + # A vote with any other source is never selected into a block. + if attestation_data.source != head_state_justified_checkpoint: continue data_root = hash_tree_root(attestation_data) block_participants = set(attestation.aggregation_bits.to_validator_indices()) local_proofs = local_proofs_by_root.get(data_root, []) - local_union: set = set() + local_union: set[ValidatorIndex] = set() for proof in local_proofs: local_union |= set(proof.participants.to_validator_indices()) diff --git a/tests/node/sync/test_service.py b/tests/node/sync/test_service.py index 6fbde2b6e..b92119b49 100644 --- a/tests/node/sync/test_service.py +++ b/tests/node/sync/test_service.py @@ -850,19 +850,25 @@ def test_replay_plain_mixed_success_and_failure(self, sync_service: SyncService) # Post-block single-message aggregate deconstruction. # -# Exercises the deconstruction core: for every processed block (gossip, -# head-sync, or backfilled), the merged multi-message aggregate proof is split -# into per-attestation single-message aggregate proofs, merged with locally held -# partials, and written into the pending pool, replacing the partials it subsumes. +# The split breaks a block's merged multi-message aggregate proof into +# per-attestation single-message aggregate proofs. +# Each recovered proof is merged with the locally held partials. +# The merged proof is written into the pending pool. +# It replaces the partials it subsumes. +# +# In production the split runs only for an aggregator's blocks. +# It runs only once the node is synced. +# Blocks drained during head-sync catch-up and backfill happen while syncing. +# Those blocks never trigger the split. # # Deconstruction only runs for an attestation when: # -# - its target is ahead of the store's justified checkpoint, so the proof -# can still help move justification, and +# - its source is the head state's justified checkpoint, so a future block +# build could anchor on it and pack the vote, and # - it adds at least one participant the node does not already hold. # -# Only the decision/gate paths are exercised here. -# These tests check when the split runs, not the cryptographic split itself. +# Most of these tests drive the core split directly. +# The gate tests drive the block-import path to check the synced-aggregator gate. # The cryptographic split and merge are covered by the aggregation consensus vectors. # Round-robin proposer is slot % num_validators with four validators. @@ -923,20 +929,53 @@ def _service(peer_id: PeerId): return create_mock_sync_service(peer_id) -def test_skips_when_target_not_ahead_of_justified( +def _expected_recovered_aggregate( + store: Store, + signed_block: SignedBlock, + attestation_data: AttestationData, +) -> SignedAggregatedAttestation: + """The aggregate recovered from a block whose single vote has no local partial.""" + block_attestation = signed_block.block.body.attestations[0] + validators = store.states[signed_block.block.parent_root].validators + public_keys_per_message = [ + [ + PublicKey.decode_bytes(validators[validator_index].attestation_public_key) + for validator_index in block_attestation.aggregation_bits.to_validator_indices() + ], + [PublicKey.decode_bytes(validators[signed_block.block.proposer_index].proposal_public_key)], + ] + combined_proof = signed_block.proof.split_by_message( + message=hash_tree_root(attestation_data), + public_keys_per_message=public_keys_per_message, + participants=block_attestation.aggregation_bits, + ) + return SignedAggregatedAttestation(data=attestation_data, proof=combined_proof) + + +def test_skips_when_source_not_current_justified( peer_id: PeerId, key_manager: XmssKeyManager ) -> None: """ - Target at or behind the justified checkpoint -> no aggregates. + Source other than the head state's justified checkpoint -> no aggregates. - The block's attestation cannot move justification, so the expensive - split is never attempted and the store is returned unchanged. + A future block on this head packs only votes sourced at this head's justified checkpoint. + A vote with a different source could never be selected into such a block. + The expensive split is skipped. + The store is unchanged. """ - chain_store, signed_block, attestation_data = _setup( + chain_store, signed_block, _ = _setup( key_manager, block_participants=[ValidatorIndex(1), ValidatorIndex(2)] ) - # Justified now sits at the attestation's target slot. - store = chain_store.model_copy(update={"latest_justified": attestation_data.target}) + # The attestation's source is the genesis justified checkpoint. + # Shift the head state's justified to the slot-1 block so the source no longer matches. + head_root = chain_store.head + head_state = chain_store.states[head_root] + shifted_state = head_state.model_copy( + update={"latest_justified": Checkpoint(root=head_root, slot=CHAIN_SLOT)} + ) + store = chain_store.model_copy( + update={"states": {**chain_store.states, head_root: shifted_state}} + ) service = _service(peer_id) new_store, aggregates = service._deconstruct_block_into_store(store, signed_block) @@ -945,6 +984,35 @@ def test_skips_when_target_not_ahead_of_justified( assert new_store is store +def test_splits_when_source_is_current_justified( + peer_id: PeerId, key_manager: XmssKeyManager +) -> None: + """ + Source is the head state's justified checkpoint and the block adds a voter -> split runs. + + The attestation sources at the genesis justified checkpoint, which the head state still carries. + With no locally held proof, the block adds new participants. + The proof is split and folded into the pending pool. + """ + block_participants = [ValidatorIndex(1), ValidatorIndex(2)] + chain_store, signed_block, attestation_data = _setup( + key_manager, block_participants=block_participants + ) + service = _service(peer_id) + + new_store, aggregates = service._deconstruct_block_into_store(chain_store, signed_block) + + # Rebuild the exact proof the split yields, the same way processing does. + expected_aggregate = _expected_recovered_aggregate(chain_store, signed_block, attestation_data) + + # One aggregate emerges, carrying the block's vote and exactly its voters. + assert aggregates == [expected_aggregate] + + # The pending pool now holds that one combined proof under the vote. + pending_proofs = new_store.latest_new_aggregated_payloads[attestation_data] + assert pending_proofs == {expected_aggregate.proof} + + def test_skips_when_block_adds_no_new_validators( peer_id: PeerId, key_manager: XmssKeyManager ) -> None: @@ -986,3 +1054,63 @@ def test_noop_when_parent_state_missing(peer_id: PeerId, key_manager: XmssKeyMan assert aggregates == [] assert new_store is store + + +def _gate_setup( + peer_id: PeerId, + key_manager: XmssKeyManager, + *, + is_aggregator: bool, + state: SyncState, +) -> tuple[SyncService, Store, SignedBlock, AttestationData]: + """Wire a real-spec service and a slot-2 block whose single vote is recoverable.""" + spec = LstarSpec() + chain_store, signed_block, attestation_data = _setup( + key_manager, block_participants=[ValidatorIndex(1), ValidatorIndex(2)] + ) + # The block builds at slot 2, so the store must be ticked there to accept it. + ticked_store, _ = spec.on_tick(chain_store, Interval.from_slot(BLOCK_SLOT), has_proposal=True) + service = create_mock_sync_service(peer_id, is_aggregator=is_aggregator) + service.spec = spec + service.state = state + return service, ticked_store, signed_block, attestation_data + + +def test_process_block_recovers_aggregates_for_synced_aggregator( + peer_id: PeerId, key_manager: XmssKeyManager +) -> None: + """Aggregator in SYNCED state runs deconstruction and queues the recovered aggregate.""" + service, ticked_store, signed_block, attestation_data = _gate_setup( + peer_id, key_manager, is_aggregator=True, state=SyncState.SYNCED + ) + + new_store = service.process_block(ticked_store, signed_block) + + expected_aggregate = _expected_recovered_aggregate(new_store, signed_block, attestation_data) + assert service._pending_block_aggregates == [expected_aggregate] + + +def test_process_block_skips_deconstruction_for_syncing_aggregator( + peer_id: PeerId, key_manager: XmssKeyManager +) -> None: + """Aggregator still SYNCING skips deconstruction, leaving no queued aggregates.""" + service, ticked_store, signed_block, _ = _gate_setup( + peer_id, key_manager, is_aggregator=True, state=SyncState.SYNCING + ) + + service.process_block(ticked_store, signed_block) + + assert service._pending_block_aggregates == [] + + +def test_process_block_skips_deconstruction_for_synced_non_aggregator( + peer_id: PeerId, key_manager: XmssKeyManager +) -> None: + """A non-aggregator in SYNCED state skips deconstruction, leaving no queued aggregates.""" + service, ticked_store, signed_block, _ = _gate_setup( + peer_id, key_manager, is_aggregator=False, state=SyncState.SYNCED + ) + + service.process_block(ticked_store, signed_block) + + assert service._pending_block_aggregates == []