resolve a few TODOs

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -2261,6 +2261,62 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .map_err(Into::into)
     }
 
+    /// Accepts a `SingleAttestation` and attempts to apply it to the "naive
+    /// aggregation pool".
+    ///
+    /// The naive aggregation pool is used by local validators to produce
+    /// `SignedAggregateAndProof`.
+    ///
+    /// If the attestation is too old (low slot) to be included in the pool it is simply dropped
+    /// and no error is returned.
+    pub fn add_single_attestation_to_naive_aggregation_pool(
+        &self,
+        attestation: SingleAttestation,
+    ) -> Result<(), AttestationError> {
+        let _timer = metrics::start_timer(&metrics::ATTESTATION_PROCESSING_APPLY_TO_AGG_POOL);
+
+        let state = self.state_at_slot(attestation.data.slot, StateSkipConfig::WithoutStateRoots)?;
+
+        // TODO(single-attestation) unwrap
+        let _committees = state.get_beacon_committees_at_slot(attestation.data.slot).unwrap();
+
+
+        let attestation = Attestation::Electra(AttestationElectra::from_single_attestation(attestation).unwrap());
+
+        match self.naive_aggregation_pool.write().insert(attestation.to_ref()) {
+            Ok(outcome) => trace!(
+                self.log,
+                "Stored unaggregated attestation";
+                "outcome" => ?outcome,
+                "index" => attestation.committee_index(),
+                "slot" => attestation.data().slot.as_u64(),
+            ),
+            Err(NaiveAggregationError::SlotTooLow {
+                slot,
+                lowest_permissible_slot,
+            }) => {
+                trace!(
+                    self.log,
+                    "Refused to store unaggregated attestation";
+                    "lowest_permissible_slot" => lowest_permissible_slot.as_u64(),
+                    "slot" => slot.as_u64(),
+                );
+            }
+            Err(e) => {
+                error!(
+                        self.log,
+                        "Failed to store unaggregated attestation";
+                        "error" => ?e,
+                        "index" => attestation.committee_index(),
+                        "slot" => attestation.data().slot.as_u64(),
+                );
+                return Err(Error::from(e).into());
+            }
+        };
+
+        Ok(())
+    }
+
     /// Accepts an `VerifiedUnaggregatedAttestation` and attempts to apply it to the "naive
     /// aggregation pool".
     ///

beacon_node/beacon_chain/src/lib.rs

@@ -53,6 +53,7 @@ mod pre_finalization_cache;
 pub mod proposer_prep_service;
 pub mod schema_change;
 pub mod shuffling_cache;
+mod single_attestation_aggregation_pool;
 pub mod single_attestation_verification;
 pub mod state_advance_timer;
 pub mod sync_committee_rewards;

beacon_node/beacon_chain/src/single_attestation_aggregation_pool.rs

@@ -0,0 +1,24 @@
+// use crate::metrics;
+// use crate::observed_aggregates::AsReference;
+// use itertools::Itertools;
+// use smallvec::SmallVec;
+// use std::collections::HashMap;
+// use tree_hash::{MerkleHasher, TreeHash, TreeHashType};
+// use types::consts::altair::SYNC_COMMITTEE_SUBNET_COUNT;
+// use types::slot_data::SlotData;
+// use types::sync_committee_contribution::SyncContributionData;
+// use types::{
+//     Attestation, AttestationData, AttestationRef, CommitteeIndex, EthSpec, Hash256, Slot,
+//     SyncCommitteeContribution,
+// };
+
+// type AttestationKeyRoot = Hash256;
+// type SyncDataRoot = Hash256;
+
+// /// Post-Electra, we need a new key for Attestations that includes the committee index
+// #[derive(Debug, Clone, PartialEq)]
+// pub struct AttestationKey {
+//     data_root: Hash256,
+//     committee_index: Option<CommitteeIndex>,
+//     slot: Slot,
+// }

beacon_node/beacon_chain/src/single_attestation_verification.rs

@@ -467,22 +467,21 @@ pub fn obtain_indexed_attestation_and_committees_per_slot<T: BeaconChainTypes>(
     chain: &BeaconChain<T>,
     attestation: &SingleAttestation,
 ) -> Result<(IndexedAttestation<T::EthSpec>, CommitteesPerSlot), Error> {
-    // TODO(single-attestation) UNWRAP plus clean up
+    // TODO(single-attestation) ERROR types
     let result = chain
         .with_committee_cache(
             attestation.data.target.root,
             attestation.data.slot.epoch(T::EthSpec::slots_per_epoch()),
             |committee_cache, _| {
                 let committees = committee_cache
                     .get_beacon_committees_at_slot(attestation.data.slot)
-                    .unwrap();
+                    .map_err(|_| Error::InvalidSignature).map_err(|_| BeaconChainError::AttestationCommitteeIndexNotSet)?;
                 let indexed_attestation =
-                    get_indexed_attestation(&committees, attestation).unwrap();
+                    get_indexed_attestation(&committees, attestation).map_err(|_s| BeaconChainError::AttestationCommitteeIndexNotSet)?;
 
                 Ok((indexed_attestation, committees.len() as u64))
             },
-        )
-        .unwrap();
+        )?;
 
     Ok(result)
 }

beacon_node/http_api/src/publish_attestations.rs

@@ -94,20 +94,8 @@ fn verify_and_publish_attestation<T: BeaconChainTypes>(
                 })
                 .map_err(|_| Error::Publication)?;
         }
-        types::AttestationRef::Electra(attn) => {
-            let single_attestation = attn
-                .to_single_attestation()
-                .map_err(|_| Error::Publication)?;
-
-            // Publish.
-            network_tx
-                .send(NetworkMessage::Publish {
-                    messages: vec![PubsubMessage::SingleAttestation(Box::new((
-                        attestation.subnet_id(),
-                        single_attestation,
-                    )))],
-                })
-                .map_err(|_| Error::Publication)?;
+        types::AttestationRef::Electra(_) => {
+            return Err(Error::Publication)            
         }
     };
 

beacon_node/lighthouse_network/src/types/pubsub.rs

@@ -8,7 +8,7 @@ use std::io::{Error, ErrorKind};
 use std::sync::Arc;
 use types::attestation::SingleAttestation;
 use types::{
-    Attestation, AttestationBase, AttestationElectra, AttesterSlashing, AttesterSlashingBase,
+    Attestation, AttestationBase, AttesterSlashing, AttesterSlashingBase,
     AttesterSlashingElectra, BlobSidecar, DataColumnSidecar, DataColumnSubnetId, EthSpec,
     ForkContext, ForkName, LightClientFinalityUpdate, LightClientOptimisticUpdate,
     ProposerSlashing, SignedAggregateAndProof, SignedAggregateAndProofBase,
@@ -198,15 +198,11 @@ impl<E: EthSpec> PubsubMessage<E> {
                             match fork_context.from_context_bytes(gossip_topic.fork_digest) {
                                 Some(&fork_name) => {
                                     if fork_name.electra_enabled() {
-                                        let single_attestation =
-                                            SingleAttestation::from_ssz_bytes(data)
-                                                .map_err(|e| format!("{:?}", e))?;
-                                        Attestation::Electra(
-                                            AttestationElectra::from_single_attestation(
-                                                single_attestation,
-                                            )
-                                            .map_err(|e| format!("{:?}", e))?,
-                                        )
+                                        // TODO(single-attestation) raise an error here
+                                        return Err(format!(
+                                            "Unknown gossipsub fork digest: {:?}",
+                                            gossip_topic.fork_digest
+                                        ))
                                     } else {
                                         Attestation::Base(
                                             AttestationBase::from_ssz_bytes(data)

beacon_node/network/src/network_beacon_processor/gossip_methods.rs

@@ -196,7 +196,6 @@ impl<E: EthSpec> FailedAtt<E> {
     pub fn attestation_data(&self) -> AttestationData {
         match self {
             FailedAtt::Unaggregate { attestation, .. } => attestation.data().clone(),
-            // TODO(single-attestation) fix this
             FailedAtt::Single { attestation, .. } => attestation.data.clone(),
             FailedAtt::Aggregate { attestation, .. } => {
                 attestation.message().aggregate().data().clone()

consensus/types/src/attestation.rs

@@ -1,16 +1,16 @@
 use crate::slot_data::SlotData;
 use crate::{test_utils::TestRandom, Hash256, Slot};
-use crate::{Checkpoint, ForkVersionDeserialize};
+use crate::{BeaconCommittee, Checkpoint, ForkVersionDeserialize};
 use derivative::Derivative;
 use safe_arith::ArithError;
 use serde::{Deserialize, Serialize};
 use ssz_derive::{Decode, Encode};
-use ssz_types::typenum::Unsigned;
 use ssz_types::BitVector;
 use std::hash::{Hash, Hasher};
 use superstruct::superstruct;
 use test_random_derive::TestRandom;
 use tree_hash_derive::TreeHash;
+use ssz_types::typenum::Unsigned;
 
 use super::{
     AggregateSignature, AttestationData, BitList, ChainSpec, Domain, EthSpec, Fork, SecretKey,
@@ -361,34 +361,6 @@ impl<E: EthSpec> AttestationElectra<E> {
         }
     }
 
-    pub fn to_single_attestation(&self) -> Result<SingleAttestation, Error> {
-        let committee_indices = self.get_committee_indices();
-        let attester_indices = self.get_aggregation_bits();
-
-        if committee_indices.len() != 1 {
-            return Err(Error::InvalidCommitteeLength);
-        }
-
-        if attester_indices.len() != 1 {
-            return Err(Error::InvalidAggregationBit);
-        }
-
-        let Some(committee_index) = committee_indices.first() else {
-            return Err(Error::InvalidCommitteeLength);
-        };
-
-        let Some(attester_index) = attester_indices.first() else {
-            return Err(Error::InvalidAggregationBit);
-        };
-
-        Ok(SingleAttestation {
-            committee_index: *committee_index as usize,
-            attester_index: *attester_index as usize,
-            data: self.data.clone(),
-            signature: self.signature.clone(),
-        })
-    }
-
     pub fn from_single_attestation(single_attestation: SingleAttestation) -> Result<Self, Error> {
         let mut committee_bits = BitVector::new();
         committee_bits.set(single_attestation.committee_index, true)?;
@@ -636,6 +608,84 @@ impl SingleAttestation {
         self.attester_index = committee_position;
         self.signature = signature.clone();
     }
+
+    //  /// Shortcut for getting the attesting indices while fetching the committee from the state's cache.
+    //  pub fn get_attesting_indices_from_state<E: EthSpec>(
+    //     state: &BeaconState<E>,
+    //     att: &AttestationElectra<E>,
+    // ) -> Result<Vec<u64>, BeaconStateError> {
+    //     let committees = state.get_beacon_committees_at_slot(att.data.slot)?;
+    //     get_attesting_indices::<E>(&committees, &att.aggregation_bits, &att.committee_bits)
+    // }
+
+    // /// Returns validator indices which participated in the attestation, sorted by increasing index.
+    // ///
+    // /// Committees must be sorted by ascending order 0..committees_per_slot
+    // pub fn get_attesting_indices<E: EthSpec>(
+    //     committees: &[BeaconCommittee],
+    //     aggregation_bits: &BitList<E::MaxValidatorsPerSlot>,
+    //     committee_bits: &BitVector<E::MaxCommitteesPerSlot>,
+    // ) -> Result<Vec<u64>, BeaconStateError> {
+    //     let mut attesting_indices = vec![];
+
+    //     let committee_indices = get_committee_indices::<E>(committee_bits);
+
+    //     let mut committee_offset = 0;
+
+    //     let committee_count_per_slot = committees.len() as u64;
+    //     let mut participant_count = 0;
+    //     for index in committee_indices {
+    //         let beacon_committee = committees
+    //             .get(index as usize)
+    //             .ok_or(Error::NoCommitteeFound(index))?;
+
+    //         // This check is new to the spec's `process_attestation` in Electra.
+    //         if index >= committee_count_per_slot {
+    //             return Err(BeaconStateError::InvalidCommitteeIndex(index));
+    //         }
+    //         participant_count.safe_add_assign(beacon_committee.committee.len() as u64)?;
+    //         let committee_attesters = beacon_committee
+    //             .committee
+    //             .iter()
+    //             .enumerate()
+    //             .filter_map(|(i, &index)| {
+    //                 if let Ok(aggregation_bit_index) = committee_offset.safe_add(i) {
+    //                     if aggregation_bits.get(aggregation_bit_index).unwrap_or(false) {
+    //                         return Some(index as u64);
+    //                     }
+    //                 }
+    //                 None
+    //             })
+    //             .collect::<HashSet<u64>>();
+
+    //         attesting_indices.extend(committee_attesters);
+    //         committee_offset.safe_add_assign(beacon_committee.committee.len())?;
+    //     }
+
+    //     // This check is new to the spec's `process_attestation` in Electra.
+    //     if participant_count as usize != aggregation_bits.len() {
+    //         return Err(BeaconStateError::InvalidBitfield);
+    //     }
+
+    //     attesting_indices.sort_unstable();
+
+    //     Ok(attesting_indices)
+    // }
+
+    pub fn to_attestation(&self, _committees: &[BeaconCommittee]) -> Result<(), Error> {
+        // let beacon_committee = committees.get(self.committee_index).unwrap();
+        // let mut participant_count = 0;
+        // beacon_committee
+        //     .committee
+        //     .iter()
+        //     .enumerate()
+        //     .filter_map(|(i, &beacon_committee)| {
+        //         todo!()
+        //     });
+
+        Ok(())
+
+    }
 }
 
 #[cfg(test)]