Filter out redundant attestations in input builder

core/src/verify.rs

@@ -183,7 +183,7 @@ fn validate_link(
 }
 
 /// Verifies a single attestation returning its attesting validator indices.
-fn process_attestation<S: StateReader, E: EthSpec>(
+pub fn process_attestation<S: StateReader, E: EthSpec>(
     state: &S,
     active_validators: &BTreeMap<ValidatorIndex, &ValidatorInfo>,
     committees: &CommitteeCache<E>,
@@ -250,7 +250,7 @@ fn is_valid_indexed_attestation<S: StateReader>(
 }
 
 /// Return all the committees for the given epoch.
-fn compute_committees<S: StateReader>(
+pub fn compute_committees<S: StateReader>(
     state_reader: &S,
     active_validators: &BTreeMap<ValidatorIndex, &ValidatorInfo>,
     epoch: Epoch,

host/src/bin/host.rs

@@ -172,9 +172,9 @@ async fn main() -> anyhow::Result<()> {
                 Checkpoint::new(trusted_epoch, trusted_beacon_block.hash_tree_root()?);
             info!("Trusted checkpoint: {}", trusted_checkpoint);
 
-            let builder = InputBuilder::<Spec, _>::new(beacon_client.clone());
+            let builder = InputBuilder::<Spec, _, _>::new(&beacon_client, &reader);
 
-            let (input, _) = builder.build(trusted_checkpoint).await?;
+            let (input, _) = builder.build(&DEFAULT_CONFIG, trusted_checkpoint).await?;
             info!("Pre-state: {:#?}", input.consensus_state);
             debug!("Input: {:?}", input);
 
@@ -226,11 +226,12 @@ async fn run_sync<E: EthSpec + Serialize>(
 
     let mut consensus_state = beacon_client.get_consensus_state(start_slot).await?;
     info!("Initial Consensus State: {:#?}", consensus_state);
-    let input_builder = InputBuilder::<E, _>::new(beacon_client.clone());
 
     loop {
+        let input_builder = InputBuilder::<E, _, _>::new(beacon_client, &sr);
+
         let (input, expected_state) = input_builder
-            .build(consensus_state.finalized_checkpoint())
+            .build(&DEFAULT_CONFIG, consensus_state.finalized_checkpoint())
             .await?;
         debug!("Input: {:?}", input);
         let msg = match run_verify(network, mode, &DEFAULT_CONFIG, &sr, input.clone()) {

host/src/input_builder.rs

@@ -15,10 +15,13 @@
 use crate::conv_attestation;
 use beacon_types::EthSpec;
 use ethereum_consensus::{electra::mainnet::SignedBeaconBlockHeader, types::mainnet::BeaconBlock};
-use std::collections::HashMap;
+use std::collections::{BTreeMap, BTreeSet, HashMap};
 use std::fmt::Display;
 use tracing::debug;
-use z_core::{Attestation, Checkpoint, ConsensusError, ConsensusState, Epoch, Input, Link, ensure};
+use z_core::{
+    Attestation, Checkpoint, Config, ConsensusError, ConsensusState, Epoch, Input, Link,
+    StateReader, compute_committees, ensure, get_total_balance, process_attestation,
+};
 
 /// A trait to abstract reading data from an instance of a beacon chain
 /// This could be an RPC to a node or something else (e.g. test harness)
@@ -50,17 +53,33 @@ pub enum InputBuilderError {
     ConsensusError(#[from] ConsensusError),
     #[error("Chain reader error")]
     ChainReader(#[from] anyhow::Error),
+    #[error("Error in called verify method")]
+    VerifyError(#[from] z_core::VerifyError),
+    #[error("Arithmetic error: {0:?}")]
+    ArithError(safe_arith::ArithError),
+    #[error("State reader error: {0}")]
+    StateReader(String),
+    #[error("Duplicate attestations from one-or-more validators detected")]
+    DuplicateAttesters,
 }
 
-pub struct InputBuilder<E, CR> {
-    chain_reader: CR,
+impl From<safe_arith::ArithError> for InputBuilderError {
+    fn from(e: safe_arith::ArithError) -> Self {
+        InputBuilderError::ArithError(e)
+    }
+}
+
+pub struct InputBuilder<'a, E, CR, SR> {
+    chain_reader: &'a CR,
+    state_reader: &'a SR,
     _phantom: std::marker::PhantomData<E>,
 }
 
-impl<E: EthSpec, CR: ChainReader> InputBuilder<E, CR> {
-    pub fn new(chain_reader: CR) -> Self {
+impl<'a, E: EthSpec, CR: ChainReader, SR: StateReader<Spec = E>> InputBuilder<'a, E, CR, SR> {
+    pub fn new(chain_reader: &'a CR, state_reader: &'a SR) -> Self {
         Self {
             chain_reader,
+            state_reader,
             _phantom: std::marker::PhantomData,
         }
     }
@@ -69,6 +88,7 @@ impl<E: EthSpec, CR: ChainReader> InputBuilder<E, CR> {
     /// used to evolve the consensus state at this checkpoint to a new consensus state in the "best" way possible
     pub async fn build(
         &self,
+        cfg: &Config,
         trusted_checkpoint: Checkpoint,
     ) -> Result<(Input<E>, ConsensusState), InputBuilderError> {
         // Find the first consensus state that confirms the finality of the trusted_checkpoint
@@ -88,7 +108,9 @@ impl<E: EthSpec, CR: ChainReader> InputBuilder<E, CR> {
         );
 
         // Concurrently fetch attestations for all required links.
-        let attestations = self.collect_attestations_for_links(&links).await?;
+        let attestations = self
+            .collect_attestations_for_links(cfg, &links, trusted_checkpoint)
+            .await?;
 
         Ok((
             Input {
@@ -168,7 +190,9 @@ impl<E: EthSpec, CR: ChainReader> InputBuilder<E, CR> {
     /// Gathers the attestations for links, looking at block in the range [start_slot, end_slot].
     async fn collect_attestations_for_links(
         &self,
+        cfg: &Config,
         links: &[Link],
+        trusted_checkpoint: Checkpoint,
     ) -> Result<Vec<Attestation<E>>, InputBuilderError> {
         if links.is_empty() {
             return Ok(vec![]);
@@ -208,12 +232,97 @@ impl<E: EthSpec, CR: ChainReader> InputBuilder<E, CR> {
             }
         }
 
-        // 3. Assemble the final nested Vec in the correct order
-        let result = links
-            .iter()
-            .flat_map(|link| attestations_by_link.remove(link).unwrap_or_default())
+        // 3. Assemble the final nested Vec in the correct order and take only as many as are required to meet the threshold requirement
+        // compute all committees for the target epoch
+        let mut result = Vec::new();
+        for link in links {
+            let link_attestations = attestations_by_link.remove(link).unwrap_or_default();
+            result.extend(self.get_min_required_attestations(
+                cfg,
+                link.target.epoch(),
+                link_attestations,
+                trusted_checkpoint,
+            )?);
+        }
+
+        Ok(result)
+    }
+
+    /// Given a target epoch and a list of attestations related to a single link, return only those
+    /// that are required to meet the justification threshold for the link as set in the config.
+    fn get_min_required_attestations(
+        &self,
+        cfg: &Config,
+        target_epoch: Epoch,
+        attestations: Vec<Attestation<E>>,
+        trusted_checkpoint: Checkpoint,
+    ) -> Result<Vec<Attestation<E>>, InputBuilderError> {
+        // We want to use active validators at the target epoch to compute which validators are participating
+        // but should use their balance/slashed status at the trusted checkpoint epoch.
+        // This is to ensure compatibility between the different state reader implementations
+        let mut active_validators: BTreeMap<_, _> = self
+            .state_reader
+            .active_validators(target_epoch)
+            .map_err(|e| InputBuilderError::StateReader(e.to_string()))?
+            .map(|(idx, v)| (idx, v.clone()))
             .collect();
 
+        // Update effective_balance and slashed status from the trusted checkpoint epoch
+        for (idx, validator) in self
+            .state_reader
+            .active_validators(trusted_checkpoint.epoch())
+            .map_err(|e| InputBuilderError::StateReader(e.to_string()))?
+        {
+            if let Some(v) = active_validators.get_mut(&idx) {
+                v.effective_balance = validator.effective_balance;
+                v.slashed = validator.slashed;
+            }
+        }
+
+        // Prepare a reference map for committee computation
+        let active_validators: BTreeMap<_, _> =
+            active_validators.iter().map(|(k, v)| (*k, v)).collect();
+
+        let committees = compute_committees(self.state_reader, &active_validators, target_epoch)?;
+
+        let total_active_balance =
+            get_total_balance(self.state_reader.chain_spec(), active_validators.values())?;
+
+        // the target balance must be sufficient for a new justification
+        let rhs = total_active_balance as u128 * cfg.justification_threshold_factor as u128;
+
+        // Use a greedy search to build a set of attestations that can finalize. In future this could be replaced with
+        // a minimum subset sum if we really want to hit the absolute minimum. It isn't expected to make a huge difference though.
+        let mut target_balance = 0u64;
+        let mut result = Vec::new();
+        let mut all_attesting_indices = BTreeSet::new();
+        for att in attestations {
+            let lhs = target_balance as u128 * cfg.justification_threshold_quotient as u128;
+            if lhs < rhs {
+                let participating_indices =
+                    process_attestation(self.state_reader, &active_validators, &committees, &att)?;
+
+                if !all_attesting_indices.is_disjoint(&participating_indices) {
+                    return Err(InputBuilderError::DuplicateAttesters);
+                }
+
+                all_attesting_indices.extend(participating_indices.iter().cloned());
+
+                let unslashed_participating_validators = participating_indices
+                    .iter()
+                    .map(|i| active_validators.get(i).unwrap())
+                    .filter(|v| !v.slashed);
+
+                target_balance += get_total_balance(
+                    self.state_reader.chain_spec(),
+                    unslashed_participating_validators,
+                )?;
+
+                result.push(att);
+            } else {
+                break; // attesting balance is sufficient, no need to add more attestations
+            }
+        }
         Ok(result)
     }
 }

host/tests/sync.rs

@@ -114,8 +114,11 @@ async fn test_zkasper_sync(
         );
 
         // Build the input and verify it
-        let builder = InputBuilder::new(harness);
-        match builder.build(consensus_state.finalized_checkpoint()).await {
+        let builder = InputBuilder::new(&harness, &state_reader);
+        match builder
+            .build(cfg, consensus_state.finalized_checkpoint())
+            .await
+        {
             Ok((input, _)) => {
                 dbg!(&input);