Electra attestation changes rm decode impl

beacon_node/beacon_chain/src/schema_change.rs

@@ -2,6 +2,7 @@
 mod migration_schema_v17;
 mod migration_schema_v18;
 mod migration_schema_v19;
+mod migration_schema_v20;
 
 use crate::beacon_chain::BeaconChainTypes;
 use crate::types::ChainSpec;
@@ -78,6 +79,14 @@ pub fn migrate_schema<T: BeaconChainTypes>(
             let ops = migration_schema_v19::downgrade_from_v19::<T>(db.clone(), log)?;
             db.store_schema_version_atomically(to, ops)
         }
+        (SchemaVersion(19), SchemaVersion(20)) => {
+            let ops = migration_schema_v20::upgrade_to_v20::<T>(db.clone(), log)?;
+            db.store_schema_version_atomically(to, ops)
+        }
+        (SchemaVersion(20), SchemaVersion(19)) => {
+            let ops = migration_schema_v20::downgrade_from_v20::<T>(db.clone(), log)?;
+            db.store_schema_version_atomically(to, ops)
+        }
         // Anything else is an error.
         (_, _) => Err(HotColdDBError::UnsupportedSchemaVersion {
             target_version: to,

beacon_node/beacon_chain/src/schema_change/migration_schema_v20.rs

@@ -0,0 +1,103 @@
+use crate::beacon_chain::{BeaconChainTypes, OP_POOL_DB_KEY};
+use operation_pool::{
+    PersistedOperationPool, PersistedOperationPoolV15, PersistedOperationPoolV20,
+};
+use slog::{debug, info, Logger};
+use std::sync::Arc;
+use store::{Error, HotColdDB, KeyValueStoreOp, StoreItem};
+use types::Attestation;
+
+pub fn upgrade_to_v20<T: BeaconChainTypes>(
+    db: Arc<HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>>,
+    log: Logger,
+) -> Result<Vec<KeyValueStoreOp>, Error> {
+    // Load a V14 op pool and transform it to V15.
+    let Some(PersistedOperationPoolV15::<T::EthSpec> {
+        attestations_v15,
+        sync_contributions,
+        attester_slashings_v15,
+        proposer_slashings,
+        voluntary_exits,
+        bls_to_execution_changes,
+        capella_bls_change_broadcast_indices,
+    }) = db.get_item(&OP_POOL_DB_KEY)?
+    else {
+        debug!(log, "Nothing to do, no operation pool stored");
+        return Ok(vec![]);
+    };
+
+    let attestations = attestations_v15
+        .into_iter()
+        .map(|(attestation, indices)| (Attestation::Base(attestation).into(), indices))
+        .collect();
+
+    let attester_slashings = attester_slashings_v15
+        .into_iter()
+        .map(|slashing| slashing.into())
+        .collect();
+
+    let v20 = PersistedOperationPool::V20(PersistedOperationPoolV20 {
+        attestations,
+        sync_contributions,
+        attester_slashings,
+        proposer_slashings,
+        voluntary_exits,
+        bls_to_execution_changes,
+        capella_bls_change_broadcast_indices,
+    });
+    Ok(vec![v20.as_kv_store_op(OP_POOL_DB_KEY)])
+}
+
+pub fn downgrade_from_v20<T: BeaconChainTypes>(
+    db: Arc<HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>>,
+    log: Logger,
+) -> Result<Vec<KeyValueStoreOp>, Error> {
+    // Load a V20 op pool and transform it to V15.
+    let Some(PersistedOperationPoolV20::<T::EthSpec> {
+        attestations,
+        sync_contributions,
+        attester_slashings,
+        proposer_slashings,
+        voluntary_exits,
+        bls_to_execution_changes,
+        capella_bls_change_broadcast_indices,
+    }) = db.get_item(&OP_POOL_DB_KEY)?
+    else {
+        debug!(log, "Nothing to do, no operation pool stored");
+        return Ok(vec![]);
+    };
+
+    let attestations_v15 = attestations
+        .into_iter()
+        .filter_map(|(attestation, indices)| {
+            if let Attestation::Base(attestation) = attestation.into() {
+                Some((attestation, indices))
+            } else {
+                info!(log, "Dropping attestation during downgrade"; "reason" => "not a base attestation");
+                None
+            }
+        })
+        .collect();
+
+    let attester_slashings_v15 = attester_slashings
+        .into_iter()
+        .filter_map(|slashing| match slashing.try_into() {
+            Ok(slashing) => Some(slashing),
+            Err(_) => {
+                info!(log, "Dropping attester slashing during downgrade"; "reason" => "not a base attester slashing");
+                None
+            }
+        })
+        .collect();
+
+    let v15 = PersistedOperationPool::V15(PersistedOperationPoolV15 {
+        attestations_v15,
+        sync_contributions,
+        attester_slashings_v15,
+        proposer_slashings,
+        voluntary_exits,
+        bls_to_execution_changes,
+        capella_bls_change_broadcast_indices,
+    });
+    Ok(vec![v15.as_kv_store_op(OP_POOL_DB_KEY)])
+}

beacon_node/lighthouse_network/src/types/pubsub.rs

@@ -7,13 +7,14 @@ use ssz::{Decode, Encode};
 use std::io::{Error, ErrorKind};
 use std::sync::Arc;
 use types::{
-    Attestation, AttesterSlashing, AttesterSlashingBase, AttesterSlashingElectra, BlobSidecar,
-    EthSpec, ForkContext, ForkName, LightClientFinalityUpdate, LightClientOptimisticUpdate,
-    ProposerSlashing, SignedAggregateAndProof, SignedAggregateAndProofBase,
-    SignedAggregateAndProofElectra, SignedBeaconBlock, SignedBeaconBlockAltair,
-    SignedBeaconBlockBase, SignedBeaconBlockBellatrix, SignedBeaconBlockCapella,
-    SignedBeaconBlockDeneb, SignedBeaconBlockElectra, SignedBlsToExecutionChange,
-    SignedContributionAndProof, SignedVoluntaryExit, SubnetId, SyncCommitteeMessage, SyncSubnetId,
+    Attestation, AttestationBase, AttestationElectra, AttesterSlashing, AttesterSlashingBase,
+    AttesterSlashingElectra, BlobSidecar, 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)]
@@ -182,7 +183,26 @@ impl<E: EthSpec> PubsubMessage<E> {
                     }
                     GossipKind::Attestation(subnet_id) => {
                         let attestation =
-                            Attestation::from_ssz_bytes(data).map_err(|e| format!("{:?}", e))?;
+                            match fork_context.from_context_bytes(gossip_topic.fork_digest) {
+                                Some(ForkName::Base)
+                                | Some(ForkName::Altair)
+                                | Some(ForkName::Bellatrix)
+                                | Some(ForkName::Capella)
+                                | Some(ForkName::Deneb) => Attestation::Base(
+                                    AttestationBase::from_ssz_bytes(data)
+                                        .map_err(|e| format!("{:?}", e))?,
+                                ),
+                                Some(ForkName::Electra) => Attestation::Electra(
+                                    AttestationElectra::from_ssz_bytes(data)
+                                        .map_err(|e| format!("{:?}", e))?,
+                                ),
+                                None => {
+                                    return Err(format!(
+                                        "Unknown gossipsub fork digest: {:?}",
+                                        gossip_topic.fork_digest
+                                    ))
+                                }
+                            };
                         Ok(PubsubMessage::Attestation(Box::new((
                             *subnet_id,
                             attestation,

beacon_node/operation_pool/src/lib.rs

@@ -14,8 +14,7 @@ pub use attestation::{earliest_attestation_validators, AttMaxCover};
 pub use attestation_storage::{AttestationRef, SplitAttestation};
 pub use max_cover::MaxCover;
 pub use persistence::{
-    PersistedOperationPool, PersistedOperationPoolV12, PersistedOperationPoolV14,
-    PersistedOperationPoolV15, PersistedOperationPoolV5,
+    PersistedOperationPool, PersistedOperationPoolV15, PersistedOperationPoolV20,
 };
 pub use reward_cache::RewardCache;
 use state_processing::epoch_cache::is_epoch_cache_initialized;