Batch attestation slashibility checking

beacon_node/http_api/tests/fork_tests.rs

@@ -152,11 +152,11 @@ async fn attestations_across_fork_with_skip_slots() {
     assert!(!unaggregated_attestations.is_empty());
     let fork_name = harness.spec.fork_name_at_slot::<E>(fork_slot);
     client
-        .post_beacon_pool_attestations_v1(&unaggregated_attestations)
+        .post_beacon_pool_attestations_v1(&unaggregated_attestations.iter().collect::<Vec<_>>())
         .await
         .unwrap();
     client
-        .post_beacon_pool_attestations_v2(&unaggregated_attestations, fork_name)
+        .post_beacon_pool_attestations_v2(&unaggregated_attestations.iter().collect::<Vec<_>>(), fork_name)
         .await
         .unwrap();
 

beacon_node/http_api/tests/interactive_tests.rs

@@ -896,7 +896,7 @@ async fn queue_attestations_from_http() {
     let fork_name = tester.harness.spec.fork_name_at_slot::<E>(attestation_slot);
     let attestation_future = tokio::spawn(async move {
         client
-            .post_beacon_pool_attestations_v2(&attestations, fork_name)
+            .post_beacon_pool_attestations_v2(&attestations.iter().collect::<Vec<_>>(), fork_name)
             .await
             .expect("attestations should be processed successfully")
     });

beacon_node/http_api/tests/tests.rs

@@ -1704,7 +1704,7 @@ impl ApiTester {
 
     pub async fn test_post_beacon_pool_attestations_valid_v1(mut self) -> Self {
         self.client
-            .post_beacon_pool_attestations_v1(self.attestations.as_slice())
+            .post_beacon_pool_attestations_v1(&self.attestations.iter().collect::<Vec<_>>())
             .await
             .unwrap();
 
@@ -1723,7 +1723,7 @@ impl ApiTester {
             .map(|att| self.chain.spec.fork_name_at_slot::<E>(att.data().slot))
             .unwrap();
         self.client
-            .post_beacon_pool_attestations_v2(self.attestations.as_slice(), fork_name)
+            .post_beacon_pool_attestations_v2(&self.attestations.iter().collect::<Vec<_>>(), fork_name)
             .await
             .unwrap();
         assert!(
@@ -1747,7 +1747,7 @@ impl ApiTester {
 
         let err = self
             .client
-            .post_beacon_pool_attestations_v1(attestations.as_slice())
+            .post_beacon_pool_attestations_v1(&attestations.iter().collect::<Vec<_>>())
             .await
             .unwrap_err();
 
@@ -1789,7 +1789,7 @@ impl ApiTester {
 
         let err_v2 = self
             .client
-            .post_beacon_pool_attestations_v2(attestations.as_slice(), fork_name)
+            .post_beacon_pool_attestations_v2(&attestations.iter().collect::<Vec<_>>(), fork_name)
             .await
             .unwrap_err();
 
@@ -3815,7 +3815,7 @@ impl ApiTester {
 
         // Attest to the current slot
         self.client
-            .post_beacon_pool_attestations_v1(self.attestations.as_slice())
+            .post_beacon_pool_attestations_v1(&self.attestations.iter().collect::<Vec<_>>())
             .await
             .unwrap();
 
@@ -5455,7 +5455,7 @@ impl ApiTester {
 
         // Attest to the current slot
         self.client
-            .post_beacon_pool_attestations_v1(self.attestations.as_slice())
+            .post_beacon_pool_attestations_v1(&self.attestations.iter().collect::<Vec<_>>())
             .await
             .unwrap();
 
@@ -5511,7 +5511,7 @@ impl ApiTester {
         let expected_attestation_len = self.attestations.len();
 
         self.client
-            .post_beacon_pool_attestations_v1(self.attestations.as_slice())
+            .post_beacon_pool_attestations_v1(&self.attestations.iter().collect::<Vec<_>>())
             .await
             .unwrap();
 

common/eth2/src/lib.rs

@@ -1264,7 +1264,7 @@ impl BeaconNodeHttpClient {
     /// `POST v1/beacon/pool/attestations`
     pub async fn post_beacon_pool_attestations_v1<E: EthSpec>(
         &self,
-        attestations: &[Attestation<E>],
+        attestations: &[&Attestation<E>],
     ) -> Result<(), Error> {
         let mut path = self.eth_path(V1)?;
 
@@ -1283,7 +1283,7 @@ impl BeaconNodeHttpClient {
     /// `POST v2/beacon/pool/attestations`
     pub async fn post_beacon_pool_attestations_v2<E: EthSpec>(
         &self,
-        attestations: &[Attestation<E>],
+        attestations: &[&Attestation<E>],
         fork_name: ForkName,
     ) -> Result<(), Error> {
         let mut path = self.eth_path(V2)?;

validator_client/slashing_protection/src/slashing_database.rs

@@ -11,7 +11,7 @@ use rusqlite::{params, OptionalExtension, Transaction, TransactionBehavior};
 use std::fs::File;
 use std::path::Path;
 use std::time::Duration;
-use types::{AttestationData, BeaconBlockHeader, Epoch, Hash256, PublicKeyBytes, SignedRoot, Slot};
+use types::{Attestation, AttestationData, BeaconBlockHeader, Epoch, Hash256, PublicKeyBytes, SignedRoot, Slot};
 
 type Pool = r2d2::Pool<SqliteConnectionManager>;
 
@@ -599,6 +599,13 @@ impl SlashingDatabase {
         Ok(safe)
     }
 
+    // pub fn batch_check_and_insert_attestations(
+    //     &self,
+    //     attestations_to_check: Vec<(Attestation<E>, DutyAndProof)>,
+    // ) {
+
+    // }
+
     /// Check an attestation for slash safety, and if it is safe, record it in the database.
     ///
     /// The checking and inserting happen atomically and exclusively. We enforce exclusivity
@@ -641,6 +648,45 @@ impl SlashingDatabase {
         Ok(safe)
     }
 
+    // TODO(attn-slash) previously I had separated checking the signing root
+    // add inserting. But this doesn't seem necessary and seems to add a bit of unneeded complexity
+    // will review this before deleting.
+    // pub fn check_attestation_signing_root(
+    //     &self,
+    //     validator_pubkey: &PublicKeyBytes,
+    //     attestation: &AttestationData,
+    //     domain: Hash256,
+    // ) -> Result<Safe, NotSafe> {
+    //     let attestation_signing_root = attestation.signing_root(domain).into();
+    //     let mut conn = self.conn_pool.get()?;
+    //     let txn = conn.transaction_with_behavior(TransactionBehavior::Deferred)?;
+    //     self.check_attestation(
+    //         &txn,
+    //         validator_pubkey,
+    //         attestation.source.epoch,
+    //         attestation.target.epoch,
+    //         attestation_signing_root,
+    //     )
+    // }
+
+    // pub fn insert_attestation_signing_root(
+    //     &self,
+    //     validator_pubkey: &PublicKeyBytes,
+    //     attestation: &AttestationData,
+    //     domain: Hash256,
+    // ) -> Result<(), NotSafe> {
+    //     let attestation_signing_root = attestation.signing_root(domain).into();
+    //     let mut conn = self.conn_pool.get()?;
+    //     let txn = conn.transaction_with_behavior(TransactionBehavior::Exclusive)?;
+    //     self.insert_attestation(
+    //         &txn,
+    //         validator_pubkey,
+    //         attestation.source.epoch,
+    //         attestation.target.epoch,
+    //         attestation_signing_root,
+    //     )
+    // }
+
     /// Transactional variant of `check_and_insert_attestation_signing_root`.
     fn check_and_insert_attestation_signing_root_txn(
         &self,

validator_client/src/attestation_data_service.rs

@@ -0,0 +1,84 @@
+use std::{collections::HashMap, sync::Arc};
+
+use slot_clock::SlotClock;
+use types::{AttestationData, CommitteeIndex, EthSpec, ForkName, Slot};
+
+use crate::{
+    beacon_node_fallback::{BeaconNodeFallback, OfflineOnFailure, RequireSynced},
+    http_metrics::metrics,
+};
+
+/// The AttestationDataService is responsible for downloading and caching attestation data at a given slot
+/// for a range of committee indexes. It also helps prevent us from re-downloading identical attestation data.
+pub struct AttestationDataService<T: SlotClock, E: EthSpec> {
+    attestation_data_by_committee: HashMap<CommitteeIndex, AttestationData>,
+    beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
+}
+
+impl<T: SlotClock, E: EthSpec> AttestationDataService<T, E> {
+    pub fn new(beacon_nodes: Arc<BeaconNodeFallback<T, E>>) -> Self {
+        Self {
+            attestation_data_by_committee: HashMap::new(),
+            beacon_nodes,
+        }
+    }
+
+    /// Get previously downloaded attestation data by a given committee index. If the Electra fork is enabled
+    /// we don't care about the committee index
+    pub fn get_data_by_committee_index(
+        &self,
+        committee_index: &CommitteeIndex,
+        fork_name: &ForkName,
+    ) -> Option<AttestationData> {
+        if fork_name.electra_enabled() {
+            let data = self.attestation_data_by_committee.iter().next();
+            if let Some((_, data)) = data {
+                return Some(data.clone());
+            }
+            None
+        } else {
+            self.attestation_data_by_committee
+                .get(committee_index)
+                .cloned()
+        }
+    }
+
+    /// Download attestation data for this slot/committee index from the beacon node.
+    pub async fn download_data(
+        &mut self,
+        committee_index: &CommitteeIndex,
+        slot: &Slot,
+        fork_name: &ForkName,
+    ) -> Result<(), String> {
+        // If we've already downloaded data for this committee index OR electra is enabled and
+        // we've already downloaded data for this slot, there's no need to re-download the data.
+        if let Some(_) = self.get_data_by_committee_index(committee_index, fork_name) {
+            return Ok(());
+        }
+
+        let attestation_data = self
+            .beacon_nodes
+            .first_success(
+                RequireSynced::No,
+                OfflineOnFailure::Yes,
+                |beacon_node| async move {
+                    let _timer = metrics::start_timer_vec(
+                        &metrics::ATTESTATION_SERVICE_TIMES,
+                        &[metrics::ATTESTATIONS_HTTP_GET],
+                    );
+                    beacon_node
+                        .get_validator_attestation_data(slot.clone(), *committee_index)
+                        .await
+                        .map_err(|e| format!("Failed to produce attestation data: {:?}", e))
+                        .map(|result| result.data)
+                },
+            )
+            .await
+            .map_err(|e| e.to_string())?;
+
+        self.attestation_data_by_committee
+            .insert(*committee_index, attestation_data);
+
+        Ok(())
+    }
+}