De-duplicate attestations in the slasher (#2767)

## Issue Addressed

Closes https://github.com/sigp/lighthouse/issues/2112
Closes https://github.com/sigp/lighthouse/issues/1861

## Proposed Changes

Collect attestations by validator index in the slasher, and use the magic of reference counting to automatically discard redundant attestations. This results in us storing only 1-2% of the attestations observed when subscribed to all subnets, which carries over to a 50-100x reduction in data stored 🎉 

## Additional Info

There's some nuance to the configuration of the `slot-offset`. It has a profound effect on the effictiveness of de-duplication, see the docs added to the book for an explanation: 5442e695e5/book/src/slasher.md (slot-offset)

slasher/src/attestation_queue.rs

@@ -1,64 +1,85 @@
-use crate::{AttesterRecord, Config};
+use crate::{AttesterRecord, Config, IndexedAttesterRecord};
 use parking_lot::Mutex;
-use std::collections::BTreeSet;
-use std::sync::Arc;
-use types::{EthSpec, IndexedAttestation};
+use std::collections::BTreeMap;
+use std::sync::{Arc, Weak};
+use types::{EthSpec, Hash256, IndexedAttestation};
 
 /// Staging area for attestations received from the network.
 ///
-/// To be added to the database in batches, for efficiency and to prevent data races.
+/// Attestations are not grouped by validator index at this stage so that they can be easily
+/// filtered for timeliness.
 #[derive(Debug, Default)]
 pub struct AttestationQueue<E: EthSpec> {
-    /// All attestations (unique) for storage on disk.
-    pub queue: Mutex<AttestationBatch<E>>,
+    pub queue: Mutex<SimpleBatch<E>>,
+}
+
+pub type SimpleBatch<E> = Vec<Arc<IndexedAttesterRecord<E>>>;
+
+/// Attestations dequeued from the queue and in preparation for processing.
+///
+/// This struct is responsible for mapping validator indices to attestations and performing
+/// de-duplication to remove redundant attestations.
+#[derive(Debug, Default)]
+pub struct AttestationBatch<E: EthSpec> {
+    /// Map from (`validator_index`, `attestation_data_hash`) to indexed attester record.
+    ///
+    /// This mapping is used for de-duplication, see:
+    ///
+    /// https://github.com/sigp/lighthouse/issues/2112
+    pub attesters: BTreeMap<(u64, Hash256), Arc<IndexedAttesterRecord<E>>>,
+
+    /// Vec of all unique indexed attester records.
+    ///
+    /// The weak references account for the fact that some records might prove useless after
+    /// de-duplication.
+    pub attestations: Vec<Weak<IndexedAttesterRecord<E>>>,
 }
 
 /// Attestations grouped by validator index range.
 #[derive(Debug)]
 pub struct GroupedAttestations<E: EthSpec> {
-    pub subqueues: Vec<AttestationBatch<E>>,
-}
-
-/// A queue of attestations for a range of validator indices.
-#[derive(Debug, Default)]
-pub struct AttestationBatch<E: EthSpec> {
-    pub attestations: Vec<Arc<(IndexedAttestation<E>, AttesterRecord)>>,
+    pub subqueues: Vec<SimpleBatch<E>>,
 }
 
 impl<E: EthSpec> AttestationBatch<E> {
-    pub fn len(&self) -> usize {
-        self.attestations.len()
+    /// Add an attestation to the queue.
+    pub fn queue(&mut self, indexed_record: Arc<IndexedAttesterRecord<E>>) {
+        self.attestations.push(Arc::downgrade(&indexed_record));
+
+        let attestation_data_hash = indexed_record.record.attestation_data_hash;
+        for &validator_index in &indexed_record.indexed.attesting_indices {
+            self.attesters
+                .entry((validator_index, attestation_data_hash))
+                .and_modify(|existing_entry| {
+                    // If the new record is for the same attestation data but with more bits set
+                    // then replace the existing record so that we might avoid storing the
+                    // smaller indexed attestation. Single-bit attestations will usually be removed
+                    // completely by this process, and aggregates will only be retained if they
+                    // are not redundant with respect to a larger aggregate seen in the same batch.
+                    if existing_entry.indexed.attesting_indices.len()
+                        < indexed_record.indexed.attesting_indices.len()
+                    {
+                        *existing_entry = indexed_record.clone();
+                    }
+                })
+                .or_insert_with(|| indexed_record.clone());
+        }
     }
 
-    pub fn is_empty(&self) -> bool {
-        self.attestations.is_empty()
-    }
-
-    /// Group the attestations by validator index.
-    pub fn group_by_validator_index(self, config: &Config) -> GroupedAttestations<E> {
+    /// Group the attestations by validator chunk index.
+    pub fn group_by_validator_chunk_index(self, config: &Config) -> GroupedAttestations<E> {
         let mut grouped_attestations = GroupedAttestations { subqueues: vec![] };
 
-        for attestation in self.attestations {
-            let subqueue_ids = attestation
-                .0
-                .attesting_indices
-                .iter()
-                .map(|validator_index| config.validator_chunk_index(*validator_index))
-                .collect::<BTreeSet<_>>();
+        for ((validator_index, _), indexed_record) in self.attesters {
+            let subqueue_id = config.validator_chunk_index(validator_index);
 
-            if let Some(max_subqueue_id) = subqueue_ids.iter().next_back() {
-                if *max_subqueue_id >= grouped_attestations.subqueues.len() {
-                    grouped_attestations
-                        .subqueues
-                        .resize_with(max_subqueue_id + 1, AttestationBatch::default);
-                }
+            if subqueue_id >= grouped_attestations.subqueues.len() {
+                grouped_attestations
+                    .subqueues
+                    .resize_with(subqueue_id + 1, SimpleBatch::default);
             }
 
-            for subqueue_id in subqueue_ids {
-                grouped_attestations.subqueues[subqueue_id]
-                    .attestations
-                    .push(attestation.clone());
-            }
+            grouped_attestations.subqueues[subqueue_id].push(indexed_record);
         }
 
         grouped_attestations
@@ -66,21 +87,18 @@ impl<E: EthSpec> AttestationBatch<E> {
 }
 
 impl<E: EthSpec> AttestationQueue<E> {
-    /// Add an attestation to the queue.
     pub fn queue(&self, attestation: IndexedAttestation<E>) {
         let attester_record = AttesterRecord::from(attestation.clone());
-        self.queue
-            .lock()
-            .attestations
-            .push(Arc::new((attestation, attester_record)));
+        let indexed_record = IndexedAttesterRecord::new(attestation, attester_record);
+        self.queue.lock().push(indexed_record);
     }
 
-    pub fn dequeue(&self) -> AttestationBatch<E> {
+    pub fn dequeue(&self) -> SimpleBatch<E> {
         std::mem::take(&mut self.queue.lock())
     }
 
-    pub fn requeue(&self, batch: AttestationBatch<E>) {
-        self.queue.lock().attestations.extend(batch.attestations);
+    pub fn requeue(&self, batch: SimpleBatch<E>) {
+        self.queue.lock().extend(batch);
     }
 
     pub fn len(&self) -> usize {