Maintain attestations that reference unknown blocks (#2319)

## Issue Addressed

#635 

## Proposed Changes
- Keep attestations that reference a block we have not seen for 30secs before being re processed
- If we do import the block before that time elapses, it is reprocessed in that moment
- The first time it fails, do nothing wrt to gossipsub propagation or peer downscoring. If after being re processed it fails, downscore with a `LowToleranceError` and ignore the message.

beacon_node/beacon_chain/src/attestation_verification.rs

@@ -375,15 +375,17 @@ impl<T: BeaconChainTypes> VerifiedAggregatedAttestation<T> {
     pub fn verify(
         signed_aggregate: SignedAggregateAndProof<T::EthSpec>,
         chain: &BeaconChain<T>,
-    ) -> Result<Self, Error> {
+    ) -> Result<Self, (Error, SignedAggregateAndProof<T::EthSpec>)> {
         Self::verify_slashable(signed_aggregate, chain)
             .map(|verified_aggregate| {
                 if let Some(slasher) = chain.slasher.as_ref() {
                     slasher.accept_attestation(verified_aggregate.indexed_attestation.clone());
                 }
                 verified_aggregate
             })
-            .map_err(|slash_info| process_slash_info(slash_info, chain))
+            .map_err(|(slash_info, original_aggregate)| {
+                (process_slash_info(slash_info, chain), original_aggregate)
+            })
     }
 
     /// Run the checks that happen before an indexed attestation is constructed.
@@ -509,17 +511,31 @@ impl<T: BeaconChainTypes> VerifiedAggregatedAttestation<T> {
     }
 
     /// Verify the attestation, producing extra information about whether it might be slashable.
+    // NOTE: Clippy considers the return too complex. This tuple is not used elsewhere so it is not
+    // worth creating an alias.
+    #[allow(clippy::type_complexity)]
     pub fn verify_slashable(
         signed_aggregate: SignedAggregateAndProof<T::EthSpec>,
         chain: &BeaconChain<T>,
-    ) -> Result<Self, AttestationSlashInfo<T, Error>> {
+    ) -> Result<
+        Self,
+        (
+            AttestationSlashInfo<T, Error>,
+            SignedAggregateAndProof<T::EthSpec>,
+        ),
+    > {
         use AttestationSlashInfo::*;
 
         let attestation = &signed_aggregate.message.aggregate;
         let aggregator_index = signed_aggregate.message.aggregator_index;
         let attestation_root = match Self::verify_early_checks(&signed_aggregate, chain) {
             Ok(root) => root,
-            Err(e) => return Err(SignatureNotChecked(signed_aggregate.message.aggregate, e)),
+            Err(e) => {
+                return Err((
+                    SignatureNotChecked(signed_aggregate.message.aggregate.clone(), e),
+                    signed_aggregate,
+                ))
+            }
         };
 
         let indexed_attestation =
@@ -546,7 +562,12 @@ impl<T: BeaconChainTypes> VerifiedAggregatedAttestation<T> {
                     .map_err(|e| BeaconChainError::from(e).into())
             }) {
                 Ok(indexed_attestation) => indexed_attestation,
-                Err(e) => return Err(SignatureNotChecked(signed_aggregate.message.aggregate, e)),
+                Err(e) => {
+                    return Err((
+                        SignatureNotChecked(signed_aggregate.message.aggregate.clone(), e),
+                        signed_aggregate,
+                    ))
+                }
             };
 
         // Ensure that all signatures are valid.
@@ -560,11 +581,11 @@ impl<T: BeaconChainTypes> VerifiedAggregatedAttestation<T> {
                     }
                 })
         {
-            return Err(SignatureInvalid(e));
+            return Err((SignatureInvalid(e), signed_aggregate));
         }
 
         if let Err(e) = Self::verify_late_checks(&signed_aggregate, attestation_root, chain) {
-            return Err(SignatureValid(indexed_attestation, e));
+            return Err((SignatureValid(indexed_attestation, e), signed_aggregate));
         }
 
         Ok(VerifiedAggregatedAttestation {
@@ -715,34 +736,39 @@ impl<T: BeaconChainTypes> VerifiedUnaggregatedAttestation<T> {
         attestation: Attestation<T::EthSpec>,
         subnet_id: Option<SubnetId>,
         chain: &BeaconChain<T>,
-    ) -> Result<Self, Error> {
+    ) -> Result<Self, (Error, Attestation<T::EthSpec>)> {
         Self::verify_slashable(attestation, subnet_id, chain)
             .map(|verified_unaggregated| {
                 if let Some(slasher) = chain.slasher.as_ref() {
                     slasher.accept_attestation(verified_unaggregated.indexed_attestation.clone());
                 }
                 verified_unaggregated
             })
-            .map_err(|slash_info| process_slash_info(slash_info, chain))
+            .map_err(|(slash_info, original_attestation)| {
+                (process_slash_info(slash_info, chain), original_attestation)
+            })
     }
 
     /// Verify the attestation, producing extra information about whether it might be slashable.
+    // NOTE: Clippy considers the return too complex. This tuple is not used elsewhere so it is not
+    // worth creating an alias.
+    #[allow(clippy::type_complexity)]
     pub fn verify_slashable(
         attestation: Attestation<T::EthSpec>,
         subnet_id: Option<SubnetId>,
         chain: &BeaconChain<T>,
-    ) -> Result<Self, AttestationSlashInfo<T, Error>> {
+    ) -> Result<Self, (AttestationSlashInfo<T, Error>, Attestation<T::EthSpec>)> {
         use AttestationSlashInfo::*;
 
         if let Err(e) = Self::verify_early_checks(&attestation, chain) {
-            return Err(SignatureNotChecked(attestation, e));
+            return Err((SignatureNotChecked(attestation.clone(), e), attestation));
         }
 
         let (indexed_attestation, committees_per_slot) =
             match obtain_indexed_attestation_and_committees_per_slot(chain, &attestation) {
                 Ok(x) => x,
                 Err(e) => {
-                    return Err(SignatureNotChecked(attestation, e));
+                    return Err((SignatureNotChecked(attestation.clone(), e), attestation));
                 }
             };
 
@@ -754,16 +780,21 @@ impl<T: BeaconChainTypes> VerifiedUnaggregatedAttestation<T> {
             chain,
         ) {
             Ok(t) => t,
-            Err(e) => return Err(SignatureNotCheckedIndexed(indexed_attestation, e)),
+            Err(e) => {
+                return Err((
+                    SignatureNotCheckedIndexed(indexed_attestation, e),
+                    attestation,
+                ))
+            }
         };
 
         // The aggregate signature of the attestation is valid.
         if let Err(e) = verify_attestation_signature(chain, &indexed_attestation) {
-            return Err(SignatureInvalid(e));
+            return Err((SignatureInvalid(e), attestation));
         }
 
         if let Err(e) = Self::verify_late_checks(&attestation, validator_index, chain) {
-            return Err(SignatureValid(indexed_attestation, e));
+            return Err((SignatureValid(indexed_attestation, e), attestation));
         }
 
         Ok(Self {

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -1225,7 +1225,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         &self,
         unaggregated_attestation: Attestation<T::EthSpec>,
         subnet_id: Option<SubnetId>,
-    ) -> Result<VerifiedUnaggregatedAttestation<T>, AttestationError> {
+    ) -> Result<VerifiedUnaggregatedAttestation<T>, (AttestationError, Attestation<T::EthSpec>)>
+    {
         metrics::inc_counter(&metrics::UNAGGREGATED_ATTESTATION_PROCESSING_REQUESTS);
         let _timer =
             metrics::start_timer(&metrics::UNAGGREGATED_ATTESTATION_GOSSIP_VERIFICATION_TIMES);
@@ -1249,7 +1250,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub fn verify_aggregated_attestation_for_gossip(
         &self,
         signed_aggregate: SignedAggregateAndProof<T::EthSpec>,
-    ) -> Result<VerifiedAggregatedAttestation<T>, AttestationError> {
+    ) -> Result<
+        VerifiedAggregatedAttestation<T>,
+        (AttestationError, SignedAggregateAndProof<T::EthSpec>),
+    > {
         metrics::inc_counter(&metrics::AGGREGATED_ATTESTATION_PROCESSING_REQUESTS);
         let _timer =
             metrics::start_timer(&metrics::AGGREGATED_ATTESTATION_GOSSIP_VERIFICATION_TIMES);

beacon_node/beacon_chain/tests/attestation_verification.rs

@@ -223,7 +223,7 @@ fn aggregated_gossip_verification() {
                         .expect(&format!(
                             "{} should error during verify_aggregated_attestation_for_gossip",
                             $desc
-                        )),
+                        )).0,
                     $( $error ) |+ $( if $guard )?
                 ),
                 "case: {}",
@@ -606,7 +606,7 @@ fn unaggregated_gossip_verification() {
                         .expect(&format!(
                             "{} should error during verify_unaggregated_attestation_for_gossip",
                             $desc
-                        )),
+                        )).0,
                     $( $error ) |+ $( if $guard )?
                 ),
                 "case: {}",

beacon_node/beacon_chain/tests/store_tests.rs

@@ -332,7 +332,7 @@ fn epoch_boundary_state_attestation_processing() {
         {
             checked_pre_fin = true;
             assert!(matches!(
-                res.err().unwrap(),
+                res.err().unwrap().0,
                 AttnError::PastSlot {
                     attestation_slot,
                     earliest_permissible_slot,

beacon_node/beacon_chain/tests/tests.rs

@@ -537,7 +537,7 @@ fn attestations_with_increasing_slots() {
 
         if expected_attestation_slot < expected_earliest_permissible_slot {
             assert!(matches!(
-                res.err().unwrap(),
+                res.err().unwrap().0,
                 AttnError::PastSlot {
                     attestation_slot,
                     earliest_permissible_slot,

beacon_node/http_api/src/lib.rs

@@ -1790,8 +1790,8 @@ pub fn serve<T: BeaconChainTypes>(
                     let mut failures = Vec::new();
 
                     // Verify that all messages in the post are valid before processing further
-                    for (index, aggregate) in aggregates.as_slice().iter().enumerate() {
-                        match chain.verify_aggregated_attestation_for_gossip(aggregate.clone()) {
+                    for (index, aggregate) in aggregates.into_iter().enumerate() {
+                        match chain.verify_aggregated_attestation_for_gossip(aggregate) {
                             Ok(verified_aggregate) => {
                                 messages.push(PubsubMessage::AggregateAndProofAttestation(Box::new(
                                     verified_aggregate.aggregate().clone(),
@@ -1816,8 +1816,8 @@ pub fn serve<T: BeaconChainTypes>(
                             // It's reasonably likely that two different validators produce
                             // identical aggregates, especially if they're using the same beacon
                             // node.
-                            Err(AttnError::AttestationAlreadyKnown(_)) => continue,
-                            Err(e) => {
+                            Err((AttnError::AttestationAlreadyKnown(_), _)) => continue,
+                            Err((e, aggregate)) => {
                                 error!(log,
                                     "Failure verifying aggregate and proofs";
                                     "error" => format!("{:?}", e),