Smart block deconstruction

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,
     )

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())
 

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 == []