fix: emit nothing for a proof-less vote and drop blocks the stage already holds

crates/verity-chain/src/block_production.rs

@@ -15,6 +15,14 @@
 //! body order, for the caller to fold. Merging changes no voter, so the post-state returned
 //! here is the one the folded block produces.
 //!
+//! # A vote with no proof emits nothing
+//!
+//! Fork choice files every vote a block carries into the counted pool with an empty proof
+//! set (`seed_block_votes`): a block's merged proof is never split back per vote. leanSpec's
+//! builder emits one attestation *per selected proof*, so such a vote is marked processed and
+//! contributes nothing to the body. The same holds here — a body attestation with no voters
+//! and no proof to fold would be one the prover refuses.
+//!
 //! Transcribed from leanSpec `src/lean_spec/spec/forks/lstar/block_production.py` and
 //! `aggregation.py`, read at commit `8603fa63`.
 
@@ -155,19 +163,22 @@ fn select_votes(
     let candidates = in_target_slot_order(aggregated_payloads);
 
     // Insertion-ordered accumulation: `order` fixes the body's attestation order, `groups`
-    // holds every proof chosen for a vote across the passes that reached it.
+    // holds every proof chosen for a vote across the passes that reached it. `processed` is
+    // wider than `groups`: a vote that was eligible but had no proof to select is done with,
+    // and must not be re-examined on the next pass.
     let mut order: Vec<AttestationData> = Vec::new();
     let mut groups: HashMap<AttestationData, Vec<SingleMessageAggregate>> = HashMap::new();
+    let mut processed: HashSet<AttestationData> = HashSet::new();
 
     loop {
         let mut found_new_entries = false;
 
         for (data, proofs) in &candidates {
-            if groups.contains_key(data) {
+            if processed.contains(data) {
                 continue;
             }
             // A proposer-side budget on distinct votes, not a consensus rule.
-            if order.len() >= MAX_ATTESTATIONS_DATA as usize {
+            if processed.len() >= MAX_ATTESTATIONS_DATA as usize {
                 break;
             }
             if !is_eligible(
@@ -181,9 +192,18 @@ fn select_votes(
                 continue;
             }
 
+            processed.insert(*data);
             found_new_entries = true;
+
+            // One attestation per selected proof, as leanSpec emits them: a vote the pool
+            // knows only from a block, with no proof behind it, selects nothing and adds
+            // nothing to the body.
+            let (selected, _) = select_proofs_for_coverage(Some(proofs), None);
+            if selected.is_empty() {
+                continue;
+            }
             order.push(*data);
-            groups.insert(*data, select_proofs_for_coverage(Some(proofs), None).0);
+            groups.insert(*data, selected);
         }
 
         if !found_new_entries {
@@ -540,4 +560,64 @@ mod tests {
         assert_eq!(built.block.body.attestations[0].data, data);
         assert_eq!(built.components.len(), 1);
     }
+
+    /// A block-carried vote is filed into the counted pool with no proof behind it
+    /// (`seed_block_votes`). leanSpec's builder emits one attestation per *selected proof*, so
+    /// such a vote contributes nothing to the body; a body attestation with no voters and no
+    /// proof to fold is what leanVM refuses as "aggregated public keys is empty".
+    #[test]
+    fn should_emit_nothing_for_a_vote_that_has_no_proof_behind_it() {
+        let (store, genesis) = anchored_on_genesis(4);
+        let parent_root = store.head;
+        let data = vote(0, 0, parent_root);
+        let payloads = HashMap::from([(data, HashSet::new())]);
+
+        let built = build_block(
+            &genesis,
+            Slot(1),
+            ValidatorIndex(1),
+            parent_root,
+            &HashSet::from([parent_root]),
+            &payloads,
+        )
+        .expect("block building does not fail on a proof-less vote");
+
+        assert!(
+            built.block.body.attestations.is_empty(),
+            "a vote with no proof must not become an empty-bits attestation"
+        );
+        assert!(built.components.is_empty());
+    }
+
+    /// The proof-less vote still counts against the distinct-vote budget and is not looked
+    /// at again; a vote that does have a proof beside it is carried as before.
+    #[test]
+    fn should_still_carry_the_votes_that_have_proofs() {
+        let (store, genesis) = anchored_on_genesis(4);
+        let parent_root = store.head;
+        let with_proof = vote(0, 0, parent_root);
+        let mut without_proof = with_proof;
+        without_proof.slot = Slot(0);
+        let payloads = HashMap::from([
+            (
+                with_proof,
+                HashSet::from([proof(&[true, false, false, false])]),
+            ),
+            (without_proof, HashSet::new()),
+        ]);
+
+        let built = build_block(
+            &genesis,
+            Slot(1),
+            ValidatorIndex(1),
+            parent_root,
+            &HashSet::from([parent_root]),
+            &payloads,
+        )
+        .expect("a block carrying the one provable vote");
+
+        assert_eq!(built.block.body.attestations.len(), 1);
+        assert_eq!(built.block.body.attestations[0].data, with_proof);
+        assert_eq!(built.components.len(), 1);
+    }
 }

crates/verity-node/src/chain.rs

@@ -291,6 +291,14 @@ impl<B: StorageBackend> ChainTask<B> {
     /// thing that can reject, and it leaves the store untouched when it does, so nothing
     /// unacceptable ever reaches a batch.
     fn import(&mut self, root: Bytes32, block: Block, proof: MultiMessageAggregate) {
+        // The verification stage drops what the snapshot already holds, but a block verified
+        // just before this import's own snapshot was published slips past it. `on_block` is
+        // a no-op for a known root; the commit and the log would not be.
+        if self.store.blocks.contains_key(&root) {
+            tracing::debug!(slot = block.slot.0, "block already imported");
+            return;
+        }
+
         let parent_slot = self
             .store
             .blocks

crates/verity-node/src/verification.rs

@@ -158,6 +158,7 @@ impl core::fmt::Display for VerificationFailure {
 pub struct StageCounters {
     rejected: AtomicU64,
     evicted: AtomicU64,
+    duplicates: AtomicU64,
 }
 
 impl StageCounters {
@@ -170,6 +171,15 @@ impl StageCounters {
     pub fn evicted(&self) -> u64 {
         self.evicted.load(Ordering::Relaxed)
     }
+
+    /// Items dropped because the snapshot, or the pending buffer, already held them.
+    ///
+    /// A block reaches the stage more than once whenever gossip and a sync fetch both
+    /// deliver it, or two peers answer the same request; verifying it again costs a proof
+    /// check and a database write for nothing.
+    pub fn duplicates(&self) -> u64 {
+        self.duplicates.load(Ordering::Relaxed)
+    }
 }
 
 /// What woke the stage up.
@@ -192,7 +202,7 @@ pub struct GossipPayload {
 }
 
 /// A decoded item, still unverified, still possibly waiting for the state it needs.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 enum Decoded {
     Block(Box<SignedBlock>),
     Attestation(Box<SignedAttestation>),
@@ -317,7 +327,16 @@ impl VerificationStage {
     }
 
     /// Verifies an item if its state is in view, and parks it otherwise.
+    ///
+    /// A block the snapshot already holds is dropped first: its proof was checked when it
+    /// was imported, and checking it again would only cost the pool a proof verification and
+    /// the chain task a write.
     async fn resolve(&mut self, decoded: Decoded) -> bool {
+        if self.already_imported(&decoded) {
+            self.counters.duplicates.fetch_add(1, Ordering::Relaxed);
+            tracing::debug!("dropping a block the snapshot already holds");
+            return true;
+        }
         let Some(validators) = self.registry_for(decoded.awaited_root()) else {
             self.park(decoded);
             return true;
@@ -374,12 +393,36 @@ impl VerificationStage {
             .map(|state| state.validators.clone())
     }
 
+    /// Whether the snapshot already holds this item's block.
+    ///
+    /// Only blocks are keyed by something the snapshot indexes. A vote seen twice is
+    /// caught in the chain task's pools, where filing it again changes nothing.
+    fn already_imported(&self, decoded: &Decoded) -> bool {
+        match decoded {
+            Decoded::Block(signed) => self
+                .view
+                .borrow()
+                .block(hash_tree_root(&signed.block))
+                .is_some(),
+            Decoded::Attestation(_) | Decoded::Aggregate(_) => false,
+        }
+    }
+
     /// Parks an item, evicting the oldest arrival when the buffer is full.
     ///
-    /// Both the arriving item and any item displaced to make room for it raise a gap signal:
+    /// An item already parked is not parked twice: the copy that is waiting will be verified
+    /// when its state arrives, and a second copy would only be verified — and imported —
+    /// behind it. The gap it reveals was signalled by the first copy.
+    ///
+    /// Both a new arrival and any item displaced to make room for it raise a gap signal:
     /// the arrival because nothing has been asked for yet, and the eviction because what was
     /// asked for is now the only way that item comes back.
     fn park(&mut self, decoded: Decoded) {
+        if self.pending.contains(&decoded) {
+            self.counters.duplicates.fetch_add(1, Ordering::Relaxed);
+            tracing::debug!("dropping an item already parked");
+            return;
+        }
         if self.pending.len() >= self.pending_capacity
             && let Some(evicted) = self.pending.pop_front()
         {
@@ -523,3 +566,112 @@ fn verify_aggregate(
         attestation: signed,
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use libssz::SszEncode;
+    use tokio::sync::{mpsc, watch};
+    use verity_chain::{ChainView, Store, generate_genesis};
+    use verity_db::stored_header;
+    use verity_p2p::GossipKind;
+    use verity_types::{
+        Block, BlockBody, SignedBlock, Slot, Validator, ValidatorIndex, Validators,
+    };
+
+    use super::{GossipPayload, StageCounters, VerificationStage};
+    use crate::store_open::block_from;
+
+    /// A stage over a snapshot of the genesis anchor, with the channels it writes to.
+    struct Harness {
+        stage: VerificationStage,
+        verified: mpsc::Receiver<super::Verified>,
+        gaps: mpsc::Receiver<crate::sync::fetch::Gap>,
+        counters: Arc<StageCounters>,
+        anchor: Block,
+    }
+
+    fn harness() -> Harness {
+        let validators = Validators::try_from(vec![Validator {
+            attestation_public_key: [1; 52],
+            proposal_public_key: [2; 52],
+            index: ValidatorIndex(0),
+        }])
+        .expect("one validator");
+        let genesis = generate_genesis(0, validators);
+        let anchor = block_from(&stored_header(&genesis), BlockBody::default());
+        let store = Store::new(&genesis, &anchor, None).expect("anchored");
+        let (_view_sender, view) = watch::channel(Arc::new(ChainView::of(&store)));
+
+        let (_inbound_sender, inbound) = mpsc::channel(4);
+        let (verified_sender, verified) = mpsc::channel(4);
+        let (gap_sender, gaps) = mpsc::channel(4);
+        let counters = Arc::new(StageCounters::default());
+        let stage = VerificationStage::new(
+            inbound,
+            verified_sender,
+            view,
+            gap_sender,
+            8,
+            Arc::clone(&counters),
+        );
+        Harness {
+            stage,
+            verified,
+            gaps,
+            counters,
+            anchor,
+        }
+    }
+
+    fn block_payload(block: Block) -> GossipPayload {
+        GossipPayload {
+            kind: GossipKind::Block,
+            payload: SignedBlock {
+                block,
+                proof: Default::default(),
+            }
+            .to_ssz(),
+        }
+    }
+
+    #[tokio::test]
+    async fn should_drop_a_block_the_snapshot_already_holds() {
+        let mut harness = harness();
+        let payload = block_payload(harness.anchor.clone());
+
+        assert!(harness.stage.accept(payload).await);
+
+        assert_eq!(harness.counters.duplicates(), 1);
+        assert_eq!(harness.counters.rejected(), 0);
+        assert!(harness.stage.pending.is_empty());
+        assert!(
+            harness.verified.try_recv().is_err(),
+            "nothing was forwarded"
+        );
+    }
+
+    #[tokio::test]
+    async fn should_park_a_block_with_an_unknown_parent_once() {
+        let mut harness = harness();
+        let orphan = Block {
+            slot: Slot(1),
+            proposer_index: ValidatorIndex(0),
+            parent_root: [7; 32],
+            state_root: [8; 32],
+            body: BlockBody::default(),
+        };
+
+        assert!(harness.stage.accept(block_payload(orphan.clone())).await);
+        assert!(harness.stage.accept(block_payload(orphan)).await);
+
+        assert_eq!(harness.stage.pending.len(), 1, "parked once");
+        assert_eq!(harness.counters.duplicates(), 1);
+        assert!(
+            harness.gaps.try_recv().is_ok(),
+            "the first copy raised the gap"
+        );
+        assert!(harness.gaps.try_recv().is_err(), "the second copy did not");
+    }
+}