Allow Checkpoint Sync on State Beyond Finalization (#5128)

Squashed commit of the following:

commit 36a5d49
Author: ethDreamer <[email protected]>
Date:   Fri Feb 9 11:27:33 2024 +0800

    Update beacon_node/client/src/builder.rs

    Co-authored-by: realbigsean <[email protected]>

commit 9b8b953
Author: Mark Mackey <[email protected]>
Date:   Mon Feb 5 12:15:55 2024 +0800

    fix up some loose ends

commit 2b82413
Author: Mark Mackey <[email protected]>
Date:   Tue Jan 30 15:00:55 2024 +0800

    Properly Persist AnchorState in ForkChoice

commit 8c071ce
Author: Mark Mackey <[email protected]>
Date:   Mon Jan 29 17:18:34 2024 +0800

    Fix Beacon Chain Tests

commit 433aac5
Author: Mark Mackey <[email protected]>
Date:   Thu Jan 25 16:42:33 2024 +0800

    Enforce Checkpoint State Inside DA Boundary

commit 76a386f
Author: Mark Mackey <[email protected]>
Date:   Thu Jan 25 15:03:27 2024 +0800

    Cleanup some TODOs

commit 117ee3c
Author: Mark Mackey <[email protected]>
Date:   Thu Jan 18 10:43:02 2024 +0100

    Add Peers with Lower Finalization on NonRevertible

commit 3ef3af3
Author: Mark Mackey <[email protected]>
Date:   Wed Jan 17 21:33:38 2024 +0100

    Added AnchorState to ForkChoice

commit aebc66d
Author: Mark Mackey <[email protected]>
Date:   Wed Jan 10 11:22:46 2024 +0100

    Allow Checkpoint Sync from Non-Finalized States

Author: paulhauner

Cargo.lock

@@ -29,6 +29,7 @@ clap = { workspace = true }
 slog = { workspace = true }
 dirs = { workspace = true }
 directory = { workspace = true }
+eth2 = { workspace = true }
 futures = { workspace = true }
 environment = { workspace = true }
 task_executor = { workspace = true }

beacon_node/Cargo.toml

@@ -389,9 +389,9 @@ pub struct PersistedForkChoiceStore {
     pub equivocating_indices: BTreeSet<u64>,
 }
 
-impl Into<PersistedForkChoiceStore> for PersistedForkChoiceStoreV11 {
-    fn into(self) -> PersistedForkChoiceStore {
-        PersistedForkChoiceStore {
+impl Into<PersistedForkChoiceStoreV17> for PersistedForkChoiceStoreV11 {
+    fn into(self) -> PersistedForkChoiceStoreV17 {
+        PersistedForkChoiceStoreV17 {
             balances_cache: self.balances_cache,
             time: self.time,
             finalized_checkpoint: self.finalized_checkpoint,
@@ -405,7 +405,7 @@ impl Into<PersistedForkChoiceStore> for PersistedForkChoiceStoreV11 {
     }
 }
 
-impl Into<PersistedForkChoiceStoreV11> for PersistedForkChoiceStore {
+impl Into<PersistedForkChoiceStoreV11> for PersistedForkChoiceStoreV17 {
     fn into(self) -> PersistedForkChoiceStoreV11 {
         PersistedForkChoiceStoreV11 {
             balances_cache: self.balances_cache,

beacon_node/beacon_chain/src/beacon_fork_choice_store.rs

@@ -23,7 +23,7 @@ use crate::{
 };
 use eth1::Config as Eth1Config;
 use execution_layer::ExecutionLayer;
-use fork_choice::{ForkChoice, ResetPayloadStatuses};
+use fork_choice::{AnchorState, ForkChoice, ResetPayloadStatuses};
 use futures::channel::mpsc::Sender;
 use kzg::{Kzg, TrustedSetup};
 use operation_pool::{OperationPool, PersistedOperationPool};
@@ -417,6 +417,7 @@ where
             genesis.beacon_block_root,
             &genesis.beacon_block,
             &genesis.beacon_state,
+            AnchorState::Finalized,
             current_slot,
             &self.spec,
         )
@@ -433,6 +434,7 @@ where
         mut weak_subj_state: BeaconState<TEthSpec>,
         weak_subj_block: SignedBeaconBlock<TEthSpec>,
         genesis_state: BeaconState<TEthSpec>,
+        anchor_state: AnchorState,
     ) -> Result<Self, String> {
         let store = self
             .store
@@ -557,6 +559,7 @@ where
             snapshot.beacon_block_root,
             &snapshot.beacon_block,
             &snapshot.beacon_state,
+            anchor_state,
             Some(weak_subj_slot),
             &self.spec,
         )
@@ -703,6 +706,7 @@ where
             None
         };
 
+        let anchor_state = fork_choice.anchor_state();
         let initial_head_block_root = fork_choice
             .get_head(current_slot, &self.spec)
             .map_err(|e| format!("Unable to get fork choice head: {:?}", e))?;
@@ -752,6 +756,7 @@ where
                 Some(current_slot),
                 &self.spec,
                 self.chain_config.progressive_balances_mode,
+                anchor_state,
                 &log,
             )?;
         }

beacon_node/beacon_chain/src/builder.rs

@@ -47,8 +47,8 @@ use crate::{
 };
 use eth2::types::{EventKind, SseChainReorg, SseFinalizedCheckpoint, SseHead, SseLateHead};
 use fork_choice::{
-    ExecutionStatus, ForkChoiceStore, ForkChoiceView, ForkchoiceUpdateParameters, ProtoBlock,
-    ResetPayloadStatuses,
+    AnchorState, ExecutionStatus, ForkChoiceStore, ForkChoiceView, ForkchoiceUpdateParameters,
+    ProtoBlock, ResetPayloadStatuses,
 };
 use itertools::process_results;
 use parking_lot::{Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard};
@@ -109,6 +109,8 @@ pub struct CachedHead<E: EthSpec> {
     justified_hash: Option<ExecutionBlockHash>,
     /// The `execution_payload.block_hash` of the finalized block. Set to `None` before Bellatrix.
     finalized_hash: Option<ExecutionBlockHash>,
+    /// The state of the finalized checkpoint
+    anchor_state: AnchorState,
 }
 
 impl<E: EthSpec> CachedHead<E> {
@@ -226,6 +228,11 @@ impl<E: EthSpec> CachedHead<E> {
             finalized_hash: self.finalized_hash,
         }
     }
+
+    /// Returns the anchor state
+    pub fn anchor_state(&self) -> AnchorState {
+        self.anchor_state
+    }
 }
 
 /// Represents the "canonical head" of the beacon chain.
@@ -266,6 +273,7 @@ impl<T: BeaconChainTypes> CanonicalHead<T> {
             head_hash: forkchoice_update_params.head_hash,
             justified_hash: forkchoice_update_params.justified_hash,
             finalized_hash: forkchoice_update_params.finalized_hash,
+            anchor_state: fork_choice_view.anchor_state,
         };
 
         Self {
@@ -318,6 +326,7 @@ impl<T: BeaconChainTypes> CanonicalHead<T> {
             head_hash: forkchoice_update_params.head_hash,
             justified_hash: forkchoice_update_params.justified_hash,
             finalized_hash: forkchoice_update_params.finalized_hash,
+            anchor_state: fork_choice_view.anchor_state,
         };
 
         *fork_choice_write_lock = fork_choice;
@@ -447,6 +456,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         self.canonical_head.cached_head_read_lock().snapshot.clone()
     }
 
+    /// Returns the `AnchorState` of the finalized checkpoint
+    /// TODO: determine if this puts a burdon on the fork choice lock
+    pub fn finalized_checkpoint_anchor_state(&self) -> AnchorState {
+        self.canonical_head.cached_head_read_lock().anchor_state()
+    }
+
     /// Returns the beacon block at the head of the canonical chain.
     ///
     /// See `Self::head` for more information.
@@ -576,6 +591,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             head_block_root: old_cached_head.head_block_root(),
             justified_checkpoint: old_cached_head.justified_checkpoint(),
             finalized_checkpoint: old_cached_head.finalized_checkpoint(),
+            anchor_state: old_cached_head.anchor_state(),
         };
 
         let mut fork_choice_write_lock = self.canonical_head.fork_choice_write_lock();
@@ -702,6 +718,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 head_hash: new_forkchoice_update_parameters.head_hash,
                 justified_hash: new_forkchoice_update_parameters.justified_hash,
                 finalized_hash: new_forkchoice_update_parameters.finalized_hash,
+                anchor_state: new_view.anchor_state,
             };
 
             let new_head = {
@@ -729,6 +746,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 head_hash: new_forkchoice_update_parameters.head_hash,
                 justified_hash: new_forkchoice_update_parameters.justified_hash,
                 finalized_hash: new_forkchoice_update_parameters.finalized_hash,
+                anchor_state: new_view.anchor_state,
             };
 
             let mut cached_head_write_lock = self.canonical_head.cached_head_write_lock();

beacon_node/beacon_chain/src/canonical_head.rs

@@ -1,5 +1,5 @@
 use crate::{BeaconForkChoiceStore, BeaconSnapshot};
-use fork_choice::{ForkChoice, PayloadVerificationStatus};
+use fork_choice::{AnchorState, ForkChoice, PayloadVerificationStatus};
 use itertools::process_results;
 use slog::{info, warn, Logger};
 use state_processing::state_advance::complete_state_advance;
@@ -97,13 +97,15 @@ pub fn revert_to_fork_boundary<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>
 /// WARNING: this function is destructive and causes fork choice to permanently forget all
 /// chains other than the chain leading to `head_block_root`. It should only be used in extreme
 /// circumstances when there is no better alternative.
+#[allow(clippy::too_many_arguments)]
 pub fn reset_fork_choice_to_finalization<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
     head_block_root: Hash256,
     head_state: &BeaconState<E>,
     store: Arc<HotColdDB<E, Hot, Cold>>,
     current_slot: Option<Slot>,
     spec: &ChainSpec,
     progressive_balances_mode: ProgressiveBalancesMode,
+    anchor_state: AnchorState,
     log: &Logger,
 ) -> Result<ForkChoice<BeaconForkChoiceStore<E, Hot, Cold>, E>, String> {
     // Fetch finalized block.
@@ -157,6 +159,7 @@ pub fn reset_fork_choice_to_finalization<E: EthSpec, Hot: ItemStore<E>, Cold: It
         finalized_block_root,
         &finalized_snapshot.beacon_block,
         &finalized_snapshot.beacon_state,
+        anchor_state,
         current_slot,
         spec,
     )

beacon_node/beacon_chain/src/fork_revert.rs

@@ -84,7 +84,7 @@ pub use eth1_chain::{Eth1Chain, Eth1ChainBackend};
 pub use events::ServerSentEventHandler;
 pub use execution_layer::EngineState;
 pub use execution_payload::NotifyExecutionLayer;
-pub use fork_choice::{ExecutionStatus, ForkchoiceUpdateParameters};
+pub use fork_choice::{AnchorState, ExecutionStatus, ForkchoiceUpdateParameters};
 pub use kzg::TrustedSetup;
 pub use metrics::scrape_for_metrics;
 pub use migrate::MigratorConfig;

beacon_node/beacon_chain/src/lib.rs

@@ -5,31 +5,34 @@ use store::{DBColumn, Error, StoreItem};
 use superstruct::superstruct;
 
 // If adding a new version you should update this type alias and fix the breakages.
-pub type PersistedForkChoice = PersistedForkChoiceV17;
+pub type PersistedForkChoice = PersistedForkChoiceV20;
 
 #[superstruct(
-    variants(V11, V17),
+    variants(V11, V17, V20),
     variant_attributes(derive(Encode, Decode)),
     no_enum
 )]
 pub struct PersistedForkChoice {
-    pub fork_choice: fork_choice::PersistedForkChoice,
+    #[superstruct(only(V11, V17))]
+    pub fork_choice: fork_choice::PersistedForkChoiceV19,
+    #[superstruct(only(V20))]
+    pub fork_choice: fork_choice::PersistedForkChoiceV20,
     #[superstruct(only(V11))]
     pub fork_choice_store: PersistedForkChoiceStoreV11,
-    #[superstruct(only(V17))]
+    #[superstruct(only(V17, V20))]
     pub fork_choice_store: PersistedForkChoiceStoreV17,
 }
 
-impl Into<PersistedForkChoice> for PersistedForkChoiceV11 {
-    fn into(self) -> PersistedForkChoice {
-        PersistedForkChoice {
+impl Into<PersistedForkChoiceV17> for PersistedForkChoiceV11 {
+    fn into(self) -> PersistedForkChoiceV17 {
+        PersistedForkChoiceV17 {
             fork_choice: self.fork_choice,
             fork_choice_store: self.fork_choice_store.into(),
         }
     }
 }
 
-impl Into<PersistedForkChoiceV11> for PersistedForkChoice {
+impl Into<PersistedForkChoiceV11> for PersistedForkChoiceV17 {
     fn into(self) -> PersistedForkChoiceV11 {
         PersistedForkChoiceV11 {
             fork_choice: self.fork_choice,
@@ -38,6 +41,24 @@ impl Into<PersistedForkChoiceV11> for PersistedForkChoice {
     }
 }
 
+impl Into<PersistedForkChoiceV20> for PersistedForkChoiceV17 {
+    fn into(self) -> PersistedForkChoiceV20 {
+        PersistedForkChoiceV20 {
+            fork_choice: self.fork_choice.into(),
+            fork_choice_store: self.fork_choice_store,
+        }
+    }
+}
+
+impl Into<PersistedForkChoiceV17> for PersistedForkChoiceV20 {
+    fn into(self) -> PersistedForkChoiceV17 {
+        PersistedForkChoiceV17 {
+            fork_choice: self.fork_choice.into(),
+            fork_choice_store: self.fork_choice_store,
+        }
+    }
+}
+
 macro_rules! impl_store_item {
     ($type:ty) => {
         impl StoreItem for $type {
@@ -58,3 +79,4 @@ macro_rules! impl_store_item {
 
 impl_store_item!(PersistedForkChoiceV11);
 impl_store_item!(PersistedForkChoiceV17);
+impl_store_item!(PersistedForkChoiceV20);

beacon_node/beacon_chain/src/persisted_fork_choice.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.rs

@@ -0,0 +1,35 @@
+use crate::beacon_chain::{BeaconChainTypes, FORK_CHOICE_DB_KEY};
+use crate::persisted_fork_choice::{PersistedForkChoiceV17, PersistedForkChoiceV20};
+use slog::{debug, Logger};
+use std::sync::Arc;
+use store::{Error, HotColdDB, KeyValueStoreOp, StoreItem};
+
+pub fn upgrade_to_v20<T: BeaconChainTypes>(
+    db: Arc<HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>>,
+    log: Logger,
+) -> Result<Vec<KeyValueStoreOp>, Error> {
+    let v17 = db
+        .get_item::<PersistedForkChoiceV17>(&FORK_CHOICE_DB_KEY)?
+        .ok_or_else(|| Error::SchemaMigrationError("fork choice missing from database".into()))?;
+
+    let v20: PersistedForkChoiceV20 = v17.into();
+
+    debug!(log, "Adding anchor_state to fork choice");
+
+    Ok(vec![v20.as_kv_store_op(FORK_CHOICE_DB_KEY)])
+}
+
+pub fn downgrade_from_v20<T: BeaconChainTypes>(
+    db: Arc<HotColdDB<T::EthSpec, T::HotStore, T::ColdStore>>,
+    log: Logger,
+) -> Result<Vec<KeyValueStoreOp>, Error> {
+    let v20 = db
+        .get_item::<PersistedForkChoiceV20>(&FORK_CHOICE_DB_KEY)?
+        .ok_or_else(|| Error::SchemaMigrationError("fork choice missing from database".into()))?;
+
+    let v17: PersistedForkChoiceV17 = v20.into();
+
+    debug!(log, "Dropping anchor_state from fork choice.");
+
+    Ok(vec![v17.as_kv_store_op(FORK_CHOICE_DB_KEY)])
+}