SingleAttestation implementation

beacon_node/http_api/src/publish_attestations.rs

@@ -50,7 +50,7 @@ use tokio::sync::{
     mpsc::{Sender, UnboundedSender},
     oneshot,
 };
-use types::Attestation;
+use types::{Attestation, EthSpec};
 
 // Error variants are only used in `Debug` and considered `dead_code` by the compiler.
 #[derive(Debug)]
@@ -82,6 +82,42 @@ fn verify_and_publish_attestation<T: BeaconChainTypes>(
         .verify_unaggregated_attestation_for_gossip(attestation, None)
         .map_err(Error::Validation)?;
 
+    match attestation.attestation() {
+        types::AttestationRef::Base(_) => {
+            // Publish.
+            network_tx
+                .send(NetworkMessage::Publish {
+                    messages: vec![PubsubMessage::Attestation(Box::new((
+                        attestation.subnet_id(),
+                        attestation.attestation().clone_as_attestation(),
+                    )))],
+                })
+                .map_err(|_| Error::Publication)?;
+        }
+        types::AttestationRef::Electra(attn) => chain
+            .with_committee_cache(
+                attn.data.target.root,
+                attn.data.slot.epoch(T::EthSpec::slots_per_epoch()),
+                |committee_cache, _| {
+                    let committees =
+                        committee_cache.get_beacon_committees_at_slot(attn.data.slot)?;
+
+                    let single_attestation = attn.to_single_attestation(&committees)?;
+
+                    network_tx
+                        .send(NetworkMessage::Publish {
+                            messages: vec![PubsubMessage::SingleAttestation(Box::new((
+                                attestation.subnet_id(),
+                                single_attestation,
+                            )))],
+                        })
+                        .map_err(|_| BeaconChainError::UnableToPublish)?;
+                    Ok(())
+                },
+            )
+            .map_err(|_| Error::Publication)?,
+    };
+
     // Publish.
     network_tx
         .send(NetworkMessage::Publish {

beacon_node/lighthouse_network/src/types/pubsub.rs

@@ -6,15 +6,16 @@ use snap::raw::{decompress_len, Decoder, Encoder};
 use ssz::{Decode, Encode};
 use std::io::{Error, ErrorKind};
 use std::sync::Arc;
+use types::attestation::SingleAttestation;
 use types::{
-    Attestation, AttestationBase, AttestationElectra, AttesterSlashing, AttesterSlashingBase,
-    AttesterSlashingElectra, BlobSidecar, DataColumnSidecar, DataColumnSubnetId, EthSpec,
-    ForkContext, ForkName, LightClientFinalityUpdate, LightClientOptimisticUpdate,
-    ProposerSlashing, SignedAggregateAndProof, SignedAggregateAndProofBase,
-    SignedAggregateAndProofElectra, SignedBeaconBlock, SignedBeaconBlockAltair,
-    SignedBeaconBlockBase, SignedBeaconBlockBellatrix, SignedBeaconBlockCapella,
-    SignedBeaconBlockDeneb, SignedBeaconBlockElectra, SignedBlsToExecutionChange,
-    SignedContributionAndProof, SignedVoluntaryExit, SubnetId, SyncCommitteeMessage, SyncSubnetId,
+    Attestation, AttestationBase, AttesterSlashing, AttesterSlashingBase, AttesterSlashingElectra,
+    BlobSidecar, DataColumnSidecar, DataColumnSubnetId, EthSpec, ForkContext, ForkName,
+    LightClientFinalityUpdate, LightClientOptimisticUpdate, ProposerSlashing,
+    SignedAggregateAndProof, SignedAggregateAndProofBase, SignedAggregateAndProofElectra,
+    SignedBeaconBlock, SignedBeaconBlockAltair, SignedBeaconBlockBase, SignedBeaconBlockBellatrix,
+    SignedBeaconBlockCapella, SignedBeaconBlockDeneb, SignedBeaconBlockElectra,
+    SignedBlsToExecutionChange, SignedContributionAndProof, SignedVoluntaryExit, SubnetId,
+    SyncCommitteeMessage, SyncSubnetId,
 };
 
 #[derive(Debug, Clone, PartialEq)]
@@ -29,6 +30,8 @@ pub enum PubsubMessage<E: EthSpec> {
     AggregateAndProofAttestation(Box<SignedAggregateAndProof<E>>),
     /// Gossipsub message providing notification of a raw un-aggregated attestation with its shard id.
     Attestation(Box<(SubnetId, Attestation<E>)>),
+    /// Gossipsub message providining notification of a `SingleAttestation` with its shard id.
+    SingleAttestation(Box<(SubnetId, SingleAttestation)>),
     /// Gossipsub message providing notification of a voluntary exit.
     VoluntaryExit(Box<SignedVoluntaryExit>),
     /// Gossipsub message providing notification of a new proposer slashing.
@@ -128,6 +131,9 @@ impl<E: EthSpec> PubsubMessage<E> {
             PubsubMessage::Attestation(attestation_data) => {
                 GossipKind::Attestation(attestation_data.0)
             }
+            PubsubMessage::SingleAttestation(attestation_data) => {
+                GossipKind::Attestation(attestation_data.0)
+            }
             PubsubMessage::VoluntaryExit(_) => GossipKind::VoluntaryExit,
             PubsubMessage::ProposerSlashing(_) => GossipKind::ProposerSlashing,
             PubsubMessage::AttesterSlashing(_) => GossipKind::AttesterSlashing,
@@ -192,10 +198,10 @@ impl<E: EthSpec> PubsubMessage<E> {
                             match fork_context.from_context_bytes(gossip_topic.fork_digest) {
                                 Some(&fork_name) => {
                                     if fork_name.electra_enabled() {
-                                        Attestation::Electra(
-                                            AttestationElectra::from_ssz_bytes(data)
-                                                .map_err(|e| format!("{:?}", e))?,
-                                        )
+                                        return Err(format!(
+                                            "Unknown gossipsub fork digest: {:?}",
+                                            gossip_topic.fork_digest
+                                        ));
                                     } else {
                                         Attestation::Base(
                                             AttestationBase::from_ssz_bytes(data)
@@ -411,6 +417,7 @@ impl<E: EthSpec> PubsubMessage<E> {
             PubsubMessage::ProposerSlashing(data) => data.as_ssz_bytes(),
             PubsubMessage::AttesterSlashing(data) => data.as_ssz_bytes(),
             PubsubMessage::Attestation(data) => data.1.as_ssz_bytes(),
+            PubsubMessage::SingleAttestation(data) => data.1.as_ssz_bytes(),
             PubsubMessage::SignedContributionAndProof(data) => data.as_ssz_bytes(),
             PubsubMessage::SyncCommitteeMessage(data) => data.1.as_ssz_bytes(),
             PubsubMessage::BlsToExecutionChange(data) => data.as_ssz_bytes(),
@@ -455,6 +462,11 @@ impl<E: EthSpec> std::fmt::Display for PubsubMessage<E> {
                 data.1.data().slot,
                 data.1.committee_index(),
             ),
+            PubsubMessage::SingleAttestation(data) => write!(
+                f,
+                "SingleAttestation: subnet_id: {}, attestation_slot: {}, attestation_index: {:?}",
+                *data.0, data.1.data.slot, data.1.committee_index
+            ),
             PubsubMessage::VoluntaryExit(_data) => write!(f, "Voluntary Exit"),
             PubsubMessage::ProposerSlashing(_data) => write!(f, "Proposer Slashing"),
             PubsubMessage::AttesterSlashing(_data) => write!(f, "Attester Slashing"),

beacon_node/network/src/network_beacon_processor/gossip_methods.rs

@@ -37,9 +37,9 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use store::hot_cold_store::HotColdDBError;
 use tokio::sync::mpsc;
 use types::{
-    beacon_block::BlockImportSource, Attestation, AttestationRef, AttesterSlashing, BlobSidecar,
-    DataColumnSidecar, DataColumnSubnetId, EthSpec, Hash256, IndexedAttestation,
-    LightClientFinalityUpdate, LightClientOptimisticUpdate, ProposerSlashing,
+    attestation::SingleAttestation, beacon_block::BlockImportSource, Attestation, AttestationRef,
+    AttesterSlashing, BlobSidecar, DataColumnSidecar, DataColumnSubnetId, EthSpec, Hash256,
+    IndexedAttestation, LightClientFinalityUpdate, LightClientOptimisticUpdate, ProposerSlashing,
     SignedAggregateAndProof, SignedBeaconBlock, SignedBlsToExecutionChange,
     SignedContributionAndProof, SignedVoluntaryExit, Slot, SubnetId, SyncCommitteeMessage,
     SyncSubnetId,
@@ -210,6 +210,43 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         })
     }
 
+    #[allow(clippy::too_many_arguments)]
+    pub fn process_gossip_single_attestation(
+        self: Arc<Self>,
+        message_id: MessageId,
+        peer_id: PeerId,
+        attestation: Box<SingleAttestation>,
+        subnet_id: SubnetId,
+        should_import: bool,
+        reprocess_tx: Option<mpsc::Sender<ReprocessQueueMessage>>,
+        seen_timestamp: Duration,
+    ) {
+        let _ = self.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_or_else(|_| vec![]);
+
+                let Ok(attestation) = attestation.to_attestation(&committees) else {
+                    todo!()
+                };
+                self.clone().process_gossip_attestation(
+                    message_id.clone(),
+                    peer_id,
+                    Box::new(attestation),
+                    subnet_id,
+                    should_import,
+                    reprocess_tx.clone(),
+                    seen_timestamp,
+                );
+
+                Ok(())
+            },
+        );
+    }
+
     /* Processing functions */
 
     /// Process the unaggregated attestation received from the gossip network and:

beacon_node/network/src/network_beacon_processor/mod.rs

@@ -1,6 +1,7 @@
 use crate::sync::manager::BlockProcessType;
 use crate::sync::SamplingId;
 use crate::{service::NetworkMessage, sync::manager::SyncMessage};
+use attestation::SingleAttestation;
 use beacon_chain::block_verification_types::RpcBlock;
 use beacon_chain::{builder::Witness, eth1_chain::CachingEth1Backend, BeaconChain};
 use beacon_chain::{BeaconChainTypes, NotifyExecutionLayer};
@@ -70,6 +71,47 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
             .map_err(Into::into)
     }
 
+    /// Create a new `Work` event for some unaggregated attestation.
+    pub fn send_single_attestation(
+        self: &Arc<Self>,
+        message_id: MessageId,
+        peer_id: PeerId,
+        single_attestation: SingleAttestation,
+        subnet_id: SubnetId,
+        should_import: bool,
+        seen_timestamp: Duration,
+    ) -> Result<(), Error<T::EthSpec>> {
+        // Define a closure for processing individual attestations.
+        let Ok(result) = self.chain.with_committee_cache(
+            single_attestation.data.target.root,
+            single_attestation
+                .data
+                .slot
+                .epoch(T::EthSpec::slots_per_epoch()),
+            |committee_cache, _| {
+                let committees =
+                    committee_cache.get_beacon_committees_at_slot(single_attestation.data.slot)?;
+
+                let attestation = single_attestation.to_attestation(&committees)?;
+
+                Ok(self.send_unaggregated_attestation(
+                    message_id.clone(),
+                    peer_id,
+                    attestation,
+                    subnet_id,
+                    should_import,
+                    seen_timestamp,
+                ))
+            },
+        ) else {
+            // TODO(single-attestation) raising a try send error here is problematic...
+            // is logging an error sufficient?
+            todo!()
+        };
+
+        result
+    }
+
     /// Create a new `Work` event for some unaggregated attestation.
     pub fn send_unaggregated_attestation(
         self: &Arc<Self>,

beacon_node/network/src/router.rs

@@ -387,6 +387,17 @@ impl<T: BeaconChainTypes> Router<T> {
                         timestamp_now(),
                     ),
                 ),
+            PubsubMessage::SingleAttestation(subnet_single_attestation) => self
+                .handle_beacon_processor_send_result(
+                    self.network_beacon_processor.send_single_attestation(
+                        message_id,
+                        peer_id,
+                        subnet_single_attestation.1,
+                        subnet_single_attestation.0,
+                        should_process,
+                        timestamp_now(),
+                    ),
+                ),
             PubsubMessage::BeaconBlock(block) => self.handle_beacon_processor_send_result(
                 self.network_beacon_processor.send_gossip_beacon_block(
                     message_id,