merge queue: embarking unstable (669932a) and #6488 together

beacon_node/beacon_chain/src/attestation_verification.rs

@@ -62,7 +62,7 @@ use tree_hash::TreeHash;
 use types::{
     Attestation, AttestationRef, BeaconCommittee, BeaconStateError::NoCommitteeFound, ChainSpec,
     CommitteeIndex, Epoch, EthSpec, Hash256, IndexedAttestation, SelectionProof,
-    SignedAggregateAndProof, Slot, SubnetId,
+    SignedAggregateAndProof, SingleAttestation, Slot, SubnetId,
 };
 
 pub use batch::{batch_verify_aggregated_attestations, batch_verify_unaggregated_attestations};
@@ -317,12 +317,22 @@ pub struct VerifiedUnaggregatedAttestation<'a, T: BeaconChainTypes> {
     attestation: AttestationRef<'a, T::EthSpec>,
     indexed_attestation: IndexedAttestation<T::EthSpec>,
     subnet_id: SubnetId,
+    validator_index: usize,
 }
 
 impl<T: BeaconChainTypes> VerifiedUnaggregatedAttestation<'_, T> {
     pub fn into_indexed_attestation(self) -> IndexedAttestation<T::EthSpec> {
         self.indexed_attestation
     }
+
+    pub fn single_attestation(&self) -> Option<SingleAttestation> {
+        Some(SingleAttestation {
+            committee_index: self.attestation.committee_index()? as usize,
+            attester_index: self.validator_index,
+            data: self.attestation.data().clone(),
+            signature: self.attestation.signature().clone(),
+        })
+    }
 }
 
 /// Custom `Clone` implementation is to avoid the restrictive trait bounds applied by the usual derive
@@ -1035,6 +1045,7 @@ impl<'a, T: BeaconChainTypes> VerifiedUnaggregatedAttestation<'a, T> {
             attestation,
             indexed_attestation,
             subnet_id,
+            validator_index: validator_index as usize,
         })
     }
 

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -2035,10 +2035,30 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             |v| {
                 // This method is called for API and gossip attestations, so this covers all unaggregated attestation events
                 if let Some(event_handler) = self.event_handler.as_ref() {
+                    if event_handler.has_single_attestation_subscribers() {
+                        let current_fork = self
+                            .spec
+                            .fork_name_at_slot::<T::EthSpec>(v.attestation().data().slot);
+                        if current_fork.electra_enabled() {
+                            // I don't see a situation where this could return None. The upstream unaggregated attestation checks
+                            // should have already verified that this is an attestation with a single committee bit set.
+                            if let Some(single_attestation) = v.single_attestation() {
+                                event_handler.register(EventKind::SingleAttestation(Box::new(
+                                    single_attestation,
+                                )));
+                            }
+                        }
+                    }
+
                     if event_handler.has_attestation_subscribers() {
-                        event_handler.register(EventKind::Attestation(Box::new(
-                            v.attestation().clone_as_attestation(),
-                        )));
+                        let current_fork = self
+                            .spec
+                            .fork_name_at_slot::<T::EthSpec>(v.attestation().data().slot);
+                        if !current_fork.electra_enabled() {
+                            event_handler.register(EventKind::Attestation(Box::new(
+                                v.attestation().clone_as_attestation(),
+                            )));
+                        }
                     }
                 }
                 metrics::inc_counter(&metrics::UNAGGREGATED_ATTESTATION_PROCESSING_SUCCESSES);

beacon_node/beacon_chain/src/events.rs

@@ -8,6 +8,7 @@ const DEFAULT_CHANNEL_CAPACITY: usize = 16;
 
 pub struct ServerSentEventHandler<E: EthSpec> {
     attestation_tx: Sender<EventKind<E>>,
+    single_attestation_tx: Sender<EventKind<E>>,
     block_tx: Sender<EventKind<E>>,
     blob_sidecar_tx: Sender<EventKind<E>>,
     finalized_tx: Sender<EventKind<E>>,
@@ -37,6 +38,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
 
     pub fn new_with_capacity(log: Logger, capacity: usize) -> Self {
         let (attestation_tx, _) = broadcast::channel(capacity);
+        let (single_attestation_tx, _) = broadcast::channel(capacity);
         let (block_tx, _) = broadcast::channel(capacity);
         let (blob_sidecar_tx, _) = broadcast::channel(capacity);
         let (finalized_tx, _) = broadcast::channel(capacity);
@@ -56,6 +58,7 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
 
         Self {
             attestation_tx,
+            single_attestation_tx,
             block_tx,
             blob_sidecar_tx,
             finalized_tx,
@@ -90,6 +93,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
                 .attestation_tx
                 .send(kind)
                 .map(|count| log_count("attestation", count)),
+            EventKind::SingleAttestation(_) => self
+                .single_attestation_tx
+                .send(kind)
+                .map(|count| log_count("single_attestation", count)),
             EventKind::Block(_) => self
                 .block_tx
                 .send(kind)
@@ -164,6 +171,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.attestation_tx.subscribe()
     }
 
+    pub fn subscribe_single_attestation(&self) -> Receiver<EventKind<E>> {
+        self.single_attestation_tx.subscribe()
+    }
+
     pub fn subscribe_block(&self) -> Receiver<EventKind<E>> {
         self.block_tx.subscribe()
     }
@@ -232,6 +243,10 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         self.attestation_tx.receiver_count() > 0
     }
 
+    pub fn has_single_attestation_subscribers(&self) -> bool {
+        self.single_attestation_tx.receiver_count() > 0
+    }
+
     pub fn has_block_subscribers(&self) -> bool {
         self.block_tx.receiver_count() > 0
     }