diff --git a/.github/workflows/docker-antithesis.yml b/.github/workflows/docker-antithesis.yml deleted file mode 100644 index a96431faf..000000000 --- a/.github/workflows/docker-antithesis.yml +++ /dev/null @@ -1,35 +0,0 @@ -name: docker antithesis - -on: - push: - branches: - - unstable - -concurrency: - group: ${{ github.workflow }}-${{ github.ref }} - cancel-in-progress: true - -env: - ANTITHESIS_PASSWORD: ${{ secrets.ANTITHESIS_PASSWORD }} - ANTITHESIS_USERNAME: ${{ secrets.ANTITHESIS_USERNAME }} - ANTITHESIS_SERVER: ${{ secrets.ANTITHESIS_SERVER }} - REPOSITORY: ${{ secrets.ANTITHESIS_REPOSITORY }} - IMAGE_NAME: lighthouse - TAG: libvoidstar - -jobs: - build-docker: - runs-on: ubuntu-22.04 - steps: - - uses: actions/checkout@v3 - - name: Update Rust - run: rustup update stable - - name: Dockerhub login - run: | - echo "${ANTITHESIS_PASSWORD}" | docker login --username ${ANTITHESIS_USERNAME} https://${ANTITHESIS_SERVER} --password-stdin - - name: Build AMD64 dockerfile (with push) - run: | - docker build \ - --tag ${ANTITHESIS_SERVER}/${REPOSITORY}/${IMAGE_NAME}:${TAG} \ - --file ./testing/antithesis/Dockerfile.libvoidstar . - docker push ${ANTITHESIS_SERVER}/${REPOSITORY}/${IMAGE_NAME}:${TAG} diff --git a/Cargo.lock b/Cargo.lock index 5cfa03a9d..ac9cdc043 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -808,14 +808,13 @@ dependencies = [ [[package]] name = "blst" -version = "0.3.10" +version = "0.3.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a30d0edd9dd1c60ddb42b80341c7852f6f985279a5c1a83659dcb65899dec99" +checksum = "c94087b935a822949d3291a9989ad2b2051ea141eda0fd4e478a75f6aa3e604b" dependencies = [ "cc", "glob", "threadpool", - "which", "zeroize", ] @@ -1197,7 +1196,6 @@ dependencies = [ "logging", "monitoring_api", "network", - "num_cpus", "operation_pool", "parking_lot 0.12.1", "sensitive_url", @@ -5256,7 +5254,6 @@ dependencies = [ "logging", "lru_cache", "matches", - "num_cpus", "operation_pool", "parking_lot 0.12.1", "rand 0.8.5", diff --git a/Makefile b/Makefile index bd49a9f90..026a38d1e 100644 --- a/Makefile +++ b/Makefile @@ -215,7 +215,7 @@ arbitrary-fuzz: # Runs cargo audit (Audit Cargo.lock files for crates with security vulnerabilities reported to the RustSec Advisory Database) audit: cargo install --force cargo-audit - cargo audit --ignore RUSTSEC-2020-0071 + cargo audit --ignore RUSTSEC-2020-0071 --ignore RUSTSEC-2022-0093 # Runs `cargo vendor` to make sure dependencies can be vendored for packaging, reproducibility and archival purpose. vendor: diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs index 10a9ce2a4..03d816db2 100644 --- a/beacon_node/beacon_chain/src/beacon_chain.rs +++ b/beacon_node/beacon_chain/src/beacon_chain.rs @@ -4943,6 +4943,7 @@ impl BeaconChain { self.log, "Produced block on state"; "block_size" => block_size, + "slot" => block.slot(), ); metrics::observe(&metrics::BLOCK_SIZE, block_size as f64); @@ -5921,14 +5922,16 @@ impl BeaconChain { let (mut state, state_root) = if let Some((state, state_root)) = head_state_opt { (state, state_root) } else { - let state_root = head_block.state_root; - let state = self + let block_state_root = head_block.state_root; + let max_slot = shuffling_epoch.start_slot(T::EthSpec::slots_per_epoch()); + let (state_root, state) = self .store .get_inconsistent_state_for_attestation_verification_only( - &state_root, - Some(head_block.slot), + &head_block_root, + max_slot, + block_state_root, )? - .ok_or(Error::MissingBeaconState(head_block.state_root))?; + .ok_or(Error::MissingBeaconState(block_state_root))?; (state, state_root) }; diff --git a/beacon_node/beacon_chain/src/beacon_fork_choice_store.rs b/beacon_node/beacon_chain/src/beacon_fork_choice_store.rs index 9b2edbd8b..2a42b49b4 100644 --- a/beacon_node/beacon_chain/src/beacon_fork_choice_store.rs +++ b/beacon_node/beacon_chain/src/beacon_fork_choice_store.rs @@ -321,9 +321,17 @@ where .deconstruct() .0; - let state = self + let max_slot = self + .justified_checkpoint + .epoch + .start_slot(E::slots_per_epoch()); + let (_, state) = self .store - .get_state(&justified_block.state_root(), Some(justified_block.slot())) + .get_advanced_hot_state( + self.justified_checkpoint.root, + max_slot, + justified_block.state_root(), + ) .map_err(Error::FailedToReadState)? .ok_or_else(|| Error::MissingState(justified_block.state_root()))?; diff --git a/beacon_node/beacon_chain/src/block_verification.rs b/beacon_node/beacon_chain/src/block_verification.rs index e60f227c9..615e0c472 100644 --- a/beacon_node/beacon_chain/src/block_verification.rs +++ b/beacon_node/beacon_chain/src/block_verification.rs @@ -1358,7 +1358,7 @@ impl ExecutionPendingBlock { // Perform a sanity check on the pre-state. let parent_slot = parent.beacon_block.slot(); - if state.slot() < parent_slot || state.slot() > parent_slot + 1 { + if state.slot() < parent_slot || state.slot() > block.slot() { return Err(BeaconChainError::BadPreState { parent_root: parent.beacon_block_root, parent_slot, @@ -1860,13 +1860,18 @@ fn load_parent>( BlockError::from(BeaconChainError::MissingBeaconBlock(block.parent_root())) })?; - // Load the parent blocks state from the database, returning an error if it is not found. + // Load the parent block's state from the database, returning an error if it is not found. // It is an error because if we know the parent block we should also know the parent state. - let parent_state_root = parent_block.state_root(); - let parent_state = chain - .get_state(&parent_state_root, Some(parent_block.slot()))? + // Retrieve any state that is advanced through to at most `block.slot()`: this is + // particularly important if `block` descends from the finalized/split block, but at a slot + // prior to the finalized slot (which is invalid and inaccessible in our DB schema). + let (parent_state_root, parent_state) = chain + .store + .get_advanced_hot_state(root, block.slot(), parent_block.state_root())? .ok_or_else(|| { - BeaconChainError::DBInconsistent(format!("Missing state {:?}", parent_state_root)) + BeaconChainError::DBInconsistent( + format!("Missing state for parent block {root:?}",), + ) })?; metrics::inc_counter(&metrics::BLOCK_PROCESSING_SNAPSHOT_CACHE_MISSES); diff --git a/beacon_node/beacon_chain/src/builder.rs b/beacon_node/beacon_chain/src/builder.rs index 1bdcc78a3..a021a2866 100644 --- a/beacon_node/beacon_chain/src/builder.rs +++ b/beacon_node/beacon_chain/src/builder.rs @@ -26,8 +26,9 @@ use operation_pool::{OperationPool, PersistedOperationPool}; use parking_lot::RwLock; use proto_array::{DisallowedReOrgOffsets, ReOrgThreshold}; use slasher::Slasher; -use slog::{crit, error, info, Logger}; +use slog::{crit, debug, error, info, Logger}; use slot_clock::{SlotClock, TestingSlotClock}; +use state_processing::per_slot_processing; use std::marker::PhantomData; use std::sync::Arc; use std::time::Duration; @@ -291,7 +292,7 @@ where let genesis_state = store .get_state(&genesis_block.state_root(), Some(genesis_block.slot())) .map_err(|e| descriptive_db_error("genesis state", &e))? - .ok_or("Genesis block not found in store")?; + .ok_or("Genesis state not found in store")?; self.genesis_time = Some(genesis_state.genesis_time()); @@ -386,6 +387,16 @@ where let (genesis, updated_builder) = self.set_genesis_state(beacon_state)?; self = updated_builder; + // Stage the database's metadata fields for atomic storage when `build` is called. + // Since v4.4.0 we will set the anchor with a dummy state upper limit in order to prevent + // historic states from being retained (unless `--reconstruct-historic-states` is set). + let retain_historic_states = self.chain_config.reconstruct_historic_states; + self.pending_io_batch.push( + store + .init_anchor_info(genesis.beacon_block.message(), retain_historic_states) + .map_err(|e| format!("Failed to initialize genesis anchor: {:?}", e))?, + ); + let fc_store = BeaconForkChoiceStore::get_forkchoice_store(store, &genesis) .map_err(|e| format!("Unable to initialize fork choice store: {e:?}"))?; let current_slot = None; @@ -412,21 +423,28 @@ where weak_subj_block: SignedBeaconBlock, genesis_state: BeaconState, ) -> Result { - let store = self.store.clone().ok_or("genesis_state requires a store")?; + let store = self + .store + .clone() + .ok_or("weak_subjectivity_state requires a store")?; + let log = self + .log + .as_ref() + .ok_or("weak_subjectivity_state requires a log")?; - let weak_subj_slot = weak_subj_state.slot(); - let weak_subj_block_root = weak_subj_block.canonical_root(); - let weak_subj_state_root = weak_subj_block.state_root(); - - // Check that the given state lies on an epoch boundary. Due to the database only storing - // full states on epoch boundaries and at restore points it would be difficult to support - // starting from a mid-epoch state. - if weak_subj_slot % TEthSpec::slots_per_epoch() != 0 { - return Err(format!( - "Checkpoint state at slot {} is not aligned to epoch start. \ - Please supply an aligned checkpoint with state.slot % 32 == 0", - weak_subj_slot, - )); + // Ensure the state is advanced to an epoch boundary. + let slots_per_epoch = TEthSpec::slots_per_epoch(); + if weak_subj_state.slot() % slots_per_epoch != 0 { + debug!( + log, + "Advancing checkpoint state to boundary"; + "state_slot" => weak_subj_state.slot(), + "block_slot" => weak_subj_block.slot(), + ); + while weak_subj_state.slot() % slots_per_epoch != 0 { + per_slot_processing(&mut weak_subj_state, None, &self.spec) + .map_err(|e| format!("Error advancing state: {e:?}"))?; + } } // Prime all caches before storing the state in the database and computing the tree hash @@ -434,21 +452,20 @@ where weak_subj_state .build_caches(&self.spec) .map_err(|e| format!("Error building caches on checkpoint state: {e:?}"))?; - weak_subj_state + let weak_subj_state_root = weak_subj_state .update_tree_hash_cache() .map_err(|e| format!("Error computing checkpoint state root: {:?}", e))?; - let latest_block_slot = weak_subj_state.latest_block_header().slot; + let weak_subj_slot = weak_subj_state.slot(); + let weak_subj_block_root = weak_subj_block.canonical_root(); - // We can only validate the block root if it exists in the state. We can't calculated it - // from the `latest_block_header` because the state root might be set to the zero hash. - if let Ok(state_slot_block_root) = weak_subj_state.get_block_root(latest_block_slot) { - if weak_subj_block_root != *state_slot_block_root { - return Err(format!( - "Snapshot state's most recent block root does not match block, expected: {:?}, got: {:?}", - weak_subj_block_root, state_slot_block_root - )); - } + // Validate the state's `latest_block_header` against the checkpoint block. + let state_latest_block_root = weak_subj_state.get_latest_block_root(weak_subj_state_root); + if weak_subj_block_root != state_latest_block_root { + return Err(format!( + "Snapshot state's most recent block root does not match block, expected: {:?}, got: {:?}", + weak_subj_block_root, state_latest_block_root + )); } // Check that the checkpoint state is for the same network as the genesis state. @@ -464,7 +481,7 @@ where // Set the store's split point *before* storing genesis so that genesis is stored // immediately in the freezer DB. - store.set_split(weak_subj_slot, weak_subj_state_root); + store.set_split(weak_subj_slot, weak_subj_state_root, weak_subj_block_root); let (_, updated_builder) = self.set_genesis_state(genesis_state)?; self = updated_builder; @@ -480,10 +497,11 @@ where // Stage the database's metadata fields for atomic storage when `build` is called. // This prevents the database from restarting in an inconsistent state if the anchor // info or split point is written before the `PersistedBeaconChain`. + let retain_historic_states = self.chain_config.reconstruct_historic_states; self.pending_io_batch.push(store.store_split_in_batch()); self.pending_io_batch.push( store - .init_anchor_info(weak_subj_block.message()) + .init_anchor_info(weak_subj_block.message(), retain_historic_states) .map_err(|e| format!("Failed to initialize anchor info: {:?}", e))?, ); @@ -685,9 +703,8 @@ where Err(e) => return Err(descriptive_db_error("head block", &e)), }; - let head_state_root = head_block.state_root(); - let head_state = store - .get_state(&head_state_root, Some(head_block.slot())) + let (_head_state_root, head_state) = store + .get_advanced_hot_state(head_block_root, current_slot, head_block.state_root()) .map_err(|e| descriptive_db_error("head state", &e))? .ok_or("Head state not found in store")?; diff --git a/beacon_node/beacon_chain/src/canonical_head.rs b/beacon_node/beacon_chain/src/canonical_head.rs index 40ac3ad34..a0b4b5a20 100644 --- a/beacon_node/beacon_chain/src/canonical_head.rs +++ b/beacon_node/beacon_chain/src/canonical_head.rs @@ -47,7 +47,8 @@ use crate::{ }; use eth2::types::{EventKind, SseChainReorg, SseFinalizedCheckpoint, SseHead, SseLateHead}; use fork_choice::{ - ExecutionStatus, ForkChoiceView, ForkchoiceUpdateParameters, ProtoBlock, ResetPayloadStatuses, + ExecutionStatus, ForkChoiceStore, ForkChoiceView, ForkchoiceUpdateParameters, ProtoBlock, + ResetPayloadStatuses, }; use itertools::process_results; use parking_lot::{Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard}; @@ -298,10 +299,10 @@ impl CanonicalHead { let beacon_block = store .get_full_block(&beacon_block_root)? .ok_or(Error::MissingBeaconBlock(beacon_block_root))?; - let beacon_state_root = beacon_block.state_root(); - let beacon_state = store - .get_state(&beacon_state_root, Some(beacon_block.slot()))? - .ok_or(Error::MissingBeaconState(beacon_state_root))?; + let current_slot = fork_choice.fc_store().get_current_slot(); + let (_, beacon_state) = store + .get_advanced_hot_state(beacon_block_root, current_slot, beacon_block.state_root())? + .ok_or(Error::MissingBeaconState(beacon_block.state_root()))?; let snapshot = BeaconSnapshot { beacon_block_root, @@ -669,10 +670,14 @@ impl BeaconChain { .get_full_block(&new_view.head_block_root)? .ok_or(Error::MissingBeaconBlock(new_view.head_block_root))?; - let beacon_state_root = beacon_block.state_root(); - let beacon_state: BeaconState = self - .get_state(&beacon_state_root, Some(beacon_block.slot()))? - .ok_or(Error::MissingBeaconState(beacon_state_root))?; + let (_, beacon_state) = self + .store + .get_advanced_hot_state( + new_view.head_block_root, + current_slot, + beacon_block.state_root(), + )? + .ok_or(Error::MissingBeaconState(beacon_block.state_root()))?; Ok(BeaconSnapshot { beacon_block: Arc::new(beacon_block), diff --git a/beacon_node/beacon_chain/src/events.rs b/beacon_node/beacon_chain/src/events.rs index fed050323..b267cc853 100644 --- a/beacon_node/beacon_chain/src/events.rs +++ b/beacon_node/beacon_chain/src/events.rs @@ -21,8 +21,11 @@ pub struct ServerSentEventHandler { } impl ServerSentEventHandler { - pub fn new(log: Logger) -> Self { - Self::new_with_capacity(log, DEFAULT_CHANNEL_CAPACITY) + pub fn new(log: Logger, capacity_multiplier: usize) -> Self { + Self::new_with_capacity( + log, + capacity_multiplier.saturating_mul(DEFAULT_CHANNEL_CAPACITY), + ) } pub fn new_with_capacity(log: Logger, capacity: usize) -> Self { diff --git a/beacon_node/beacon_chain/src/migrate.rs b/beacon_node/beacon_chain/src/migrate.rs index 4251a9bf5..32c13ccb0 100644 --- a/beacon_node/beacon_chain/src/migrate.rs +++ b/beacon_node/beacon_chain/src/migrate.rs @@ -289,6 +289,7 @@ impl, Cold: ItemStore> BackgroundMigrator state, @@ -342,7 +343,12 @@ impl, Cold: ItemStore> BackgroundMigrator {} Err(Error::HotColdDBError(HotColdDBError::FreezeSlotUnaligned(slot))) => { debug!( diff --git a/beacon_node/beacon_chain/tests/attestation_verification.rs b/beacon_node/beacon_chain/tests/attestation_verification.rs index f6a0a5b95..2501768c7 100644 --- a/beacon_node/beacon_chain/tests/attestation_verification.rs +++ b/beacon_node/beacon_chain/tests/attestation_verification.rs @@ -9,7 +9,7 @@ use beacon_chain::{ test_utils::{ test_spec, AttestationStrategy, BeaconChainHarness, BlockStrategy, EphemeralHarnessType, }, - BeaconChain, BeaconChainError, BeaconChainTypes, WhenSlotSkipped, + BeaconChain, BeaconChainError, BeaconChainTypes, ChainConfig, WhenSlotSkipped, }; use genesis::{interop_genesis_state, DEFAULT_ETH1_BLOCK_HASH}; use int_to_bytes::int_to_bytes32; @@ -47,6 +47,10 @@ fn get_harness(validator_count: usize) -> BeaconChainHarness ( fn get_harness(validator_count: usize) -> BeaconChainHarness> { let harness = BeaconChainHarness::builder(MainnetEthSpec) .default_spec() + .chain_config(ChainConfig { + reconstruct_historic_states: true, + ..ChainConfig::default() + }) .keypairs(KEYPAIRS[0..validator_count].to_vec()) .fresh_ephemeral_store() .mock_execution_layer() diff --git a/beacon_node/beacon_chain/tests/payload_invalidation.rs b/beacon_node/beacon_chain/tests/payload_invalidation.rs index 399487928..5076a6c1a 100644 --- a/beacon_node/beacon_chain/tests/payload_invalidation.rs +++ b/beacon_node/beacon_chain/tests/payload_invalidation.rs @@ -7,7 +7,7 @@ use beacon_chain::otb_verification_service::{ use beacon_chain::{ canonical_head::{CachedHead, CanonicalHead}, test_utils::{BeaconChainHarness, EphemeralHarnessType}, - BeaconChainError, BlockError, ExecutionPayloadError, NotifyExecutionLayer, + BeaconChainError, BlockError, ChainConfig, ExecutionPayloadError, NotifyExecutionLayer, OverrideForkchoiceUpdate, StateSkipConfig, WhenSlotSkipped, INVALID_FINALIZED_MERGE_TRANSITION_BLOCK_SHUTDOWN_REASON, INVALID_JUSTIFIED_PAYLOAD_SHUTDOWN_REASON, @@ -59,6 +59,10 @@ impl InvalidPayloadRig { let harness = BeaconChainHarness::builder(MainnetEthSpec) .spec(spec) + .chain_config(ChainConfig { + reconstruct_historic_states: true, + ..ChainConfig::default() + }) .logger(test_logger()) .deterministic_keypairs(VALIDATOR_COUNT) .mock_execution_layer() diff --git a/beacon_node/beacon_chain/tests/store_tests.rs b/beacon_node/beacon_chain/tests/store_tests.rs index f575d64d4..899b31cb8 100644 --- a/beacon_node/beacon_chain/tests/store_tests.rs +++ b/beacon_node/beacon_chain/tests/store_tests.rs @@ -12,7 +12,7 @@ use beacon_chain::validator_monitor::DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD; use beacon_chain::{ data_availability_checker::MaybeAvailableBlock, historical_blocks::HistoricalBlockError, migrate::MigratorConfig, BeaconChain, BeaconChainError, BeaconChainTypes, BeaconSnapshot, - ChainConfig, NotifyExecutionLayer, ServerSentEventHandler, WhenSlotSkipped, + BlockError, ChainConfig, NotifyExecutionLayer, ServerSentEventHandler, WhenSlotSkipped, }; use eth2_network_config::get_trusted_setup; use kzg::TrustedSetup; @@ -20,7 +20,8 @@ use lazy_static::lazy_static; use logging::test_logger; use maplit::hashset; use rand::Rng; -use state_processing::BlockReplayer; +use slot_clock::{SlotClock, TestingSlotClock}; +use state_processing::{state_advance::complete_state_advance, BlockReplayer}; use std::collections::HashMap; use std::collections::HashSet; use std::convert::TryInto; @@ -77,6 +78,19 @@ fn get_store_with_spec( fn get_harness( store: Arc, LevelDB>>, validator_count: usize, +) -> TestHarness { + // Most tests expect to retain historic states, so we use this as the default. + let chain_config = ChainConfig { + reconstruct_historic_states: true, + ..ChainConfig::default() + }; + get_harness_generic(store, validator_count, chain_config) +} + +fn get_harness_generic( + store: Arc, LevelDB>>, + validator_count: usize, + chain_config: ChainConfig, ) -> TestHarness { let harness = TestHarness::builder(MinimalEthSpec) .default_spec() @@ -84,6 +98,7 @@ fn get_harness( .logger(store.logger().clone()) .fresh_disk_store(store) .mock_execution_layer() + .chain_config(chain_config) .build(); harness.advance_slot(); harness @@ -472,13 +487,15 @@ async fn block_replay_with_inaccurate_state_roots() { .await; // Slot must not be 0 mod 32 or else no blocks will be replayed. - let (mut head_state, head_root) = harness.get_current_state_and_root(); + let (mut head_state, head_state_root) = harness.get_current_state_and_root(); + let head_block_root = harness.head_block_root(); assert_ne!(head_state.slot() % 32, 0); - let mut fast_head_state = store + let (_, mut fast_head_state) = store .get_inconsistent_state_for_attestation_verification_only( - &head_root, - Some(head_state.slot()), + &head_block_root, + head_state.slot(), + head_state_root, ) .unwrap() .unwrap(); @@ -577,14 +594,7 @@ async fn block_replayer_hooks() { async fn delete_blocks_and_states() { let db_path = tempdir().unwrap(); let store = get_store(&db_path); - let validators_keypairs = - types::test_utils::generate_deterministic_keypairs(LOW_VALIDATOR_COUNT); - let harness = TestHarness::builder(MinimalEthSpec) - .default_spec() - .keypairs(validators_keypairs) - .fresh_disk_store(store.clone()) - .mock_execution_layer() - .build(); + let harness = get_harness(store.clone(), LOW_VALIDATOR_COUNT); let unforked_blocks: u64 = 4 * E::slots_per_epoch(); @@ -1027,18 +1037,14 @@ fn check_shuffling_compatible( // Ensure blocks from abandoned forks are pruned from the Hot DB #[tokio::test] async fn prunes_abandoned_fork_between_two_finalized_checkpoints() { - const HONEST_VALIDATOR_COUNT: usize = 32 + 0; - const ADVERSARIAL_VALIDATOR_COUNT: usize = 16 - 0; + const HONEST_VALIDATOR_COUNT: usize = 32; + const ADVERSARIAL_VALIDATOR_COUNT: usize = 16; const VALIDATOR_COUNT: usize = HONEST_VALIDATOR_COUNT + ADVERSARIAL_VALIDATOR_COUNT; - let validators_keypairs = types::test_utils::generate_deterministic_keypairs(VALIDATOR_COUNT); let honest_validators: Vec = (0..HONEST_VALIDATOR_COUNT).collect(); let adversarial_validators: Vec = (HONEST_VALIDATOR_COUNT..VALIDATOR_COUNT).collect(); - let rig = BeaconChainHarness::builder(MinimalEthSpec) - .default_spec() - .keypairs(validators_keypairs) - .fresh_ephemeral_store() - .mock_execution_layer() - .build(); + let db_path = tempdir().unwrap(); + let store = get_store(&db_path); + let rig = get_harness(store.clone(), VALIDATOR_COUNT); let slots_per_epoch = rig.slots_per_epoch(); let (mut state, state_root) = rig.get_current_state_and_root(); @@ -1137,18 +1143,14 @@ async fn prunes_abandoned_fork_between_two_finalized_checkpoints() { #[tokio::test] async fn pruning_does_not_touch_abandoned_block_shared_with_canonical_chain() { - const HONEST_VALIDATOR_COUNT: usize = 32 + 0; - const ADVERSARIAL_VALIDATOR_COUNT: usize = 16 - 0; + const HONEST_VALIDATOR_COUNT: usize = 32; + const ADVERSARIAL_VALIDATOR_COUNT: usize = 16; const VALIDATOR_COUNT: usize = HONEST_VALIDATOR_COUNT + ADVERSARIAL_VALIDATOR_COUNT; - let validators_keypairs = types::test_utils::generate_deterministic_keypairs(VALIDATOR_COUNT); let honest_validators: Vec = (0..HONEST_VALIDATOR_COUNT).collect(); let adversarial_validators: Vec = (HONEST_VALIDATOR_COUNT..VALIDATOR_COUNT).collect(); - let rig = BeaconChainHarness::builder(MinimalEthSpec) - .default_spec() - .keypairs(validators_keypairs) - .fresh_ephemeral_store() - .mock_execution_layer() - .build(); + let db_path = tempdir().unwrap(); + let store = get_store(&db_path); + let rig = get_harness(store.clone(), VALIDATOR_COUNT); let slots_per_epoch = rig.slots_per_epoch(); let (state, state_root) = rig.get_current_state_and_root(); @@ -1272,15 +1274,11 @@ async fn pruning_does_not_touch_blocks_prior_to_finalization() { const HONEST_VALIDATOR_COUNT: usize = 32; const ADVERSARIAL_VALIDATOR_COUNT: usize = 16; const VALIDATOR_COUNT: usize = HONEST_VALIDATOR_COUNT + ADVERSARIAL_VALIDATOR_COUNT; - let validators_keypairs = types::test_utils::generate_deterministic_keypairs(VALIDATOR_COUNT); let honest_validators: Vec = (0..HONEST_VALIDATOR_COUNT).collect(); let adversarial_validators: Vec = (HONEST_VALIDATOR_COUNT..VALIDATOR_COUNT).collect(); - let rig = BeaconChainHarness::builder(MinimalEthSpec) - .default_spec() - .keypairs(validators_keypairs) - .fresh_ephemeral_store() - .mock_execution_layer() - .build(); + let db_path = tempdir().unwrap(); + let store = get_store(&db_path); + let rig = get_harness(store.clone(), VALIDATOR_COUNT); let slots_per_epoch = rig.slots_per_epoch(); let (mut state, state_root) = rig.get_current_state_and_root(); @@ -1364,18 +1362,14 @@ async fn pruning_does_not_touch_blocks_prior_to_finalization() { #[tokio::test] async fn prunes_fork_growing_past_youngest_finalized_checkpoint() { - const HONEST_VALIDATOR_COUNT: usize = 32 + 0; - const ADVERSARIAL_VALIDATOR_COUNT: usize = 16 - 0; + const HONEST_VALIDATOR_COUNT: usize = 32; + const ADVERSARIAL_VALIDATOR_COUNT: usize = 16; const VALIDATOR_COUNT: usize = HONEST_VALIDATOR_COUNT + ADVERSARIAL_VALIDATOR_COUNT; - let validators_keypairs = types::test_utils::generate_deterministic_keypairs(VALIDATOR_COUNT); let honest_validators: Vec = (0..HONEST_VALIDATOR_COUNT).collect(); let adversarial_validators: Vec = (HONEST_VALIDATOR_COUNT..VALIDATOR_COUNT).collect(); - let rig = BeaconChainHarness::builder(MinimalEthSpec) - .default_spec() - .keypairs(validators_keypairs) - .fresh_ephemeral_store() - .mock_execution_layer() - .build(); + let db_path = tempdir().unwrap(); + let store = get_store(&db_path); + let rig = get_harness(store.clone(), VALIDATOR_COUNT); let (state, state_root) = rig.get_current_state_and_root(); // Fill up 0th epoch with canonical chain blocks @@ -1509,18 +1503,14 @@ async fn prunes_fork_growing_past_youngest_finalized_checkpoint() { // This is to check if state outside of normal block processing are pruned correctly. #[tokio::test] async fn prunes_skipped_slots_states() { - const HONEST_VALIDATOR_COUNT: usize = 32 + 0; - const ADVERSARIAL_VALIDATOR_COUNT: usize = 16 - 0; + const HONEST_VALIDATOR_COUNT: usize = 32; + const ADVERSARIAL_VALIDATOR_COUNT: usize = 16; const VALIDATOR_COUNT: usize = HONEST_VALIDATOR_COUNT + ADVERSARIAL_VALIDATOR_COUNT; - let validators_keypairs = types::test_utils::generate_deterministic_keypairs(VALIDATOR_COUNT); let honest_validators: Vec = (0..HONEST_VALIDATOR_COUNT).collect(); let adversarial_validators: Vec = (HONEST_VALIDATOR_COUNT..VALIDATOR_COUNT).collect(); - let rig = BeaconChainHarness::builder(MinimalEthSpec) - .default_spec() - .keypairs(validators_keypairs) - .fresh_ephemeral_store() - .mock_execution_layer() - .build(); + let db_path = tempdir().unwrap(); + let store = get_store(&db_path); + let rig = get_harness(store.clone(), VALIDATOR_COUNT); let (state, state_root) = rig.get_current_state_and_root(); let canonical_slots_zeroth_epoch: Vec = @@ -1638,18 +1628,14 @@ async fn prunes_skipped_slots_states() { // This is to check if state outside of normal block processing are pruned correctly. #[tokio::test] async fn finalizes_non_epoch_start_slot() { - const HONEST_VALIDATOR_COUNT: usize = 32 + 0; - const ADVERSARIAL_VALIDATOR_COUNT: usize = 16 - 0; + const HONEST_VALIDATOR_COUNT: usize = 32; + const ADVERSARIAL_VALIDATOR_COUNT: usize = 16; const VALIDATOR_COUNT: usize = HONEST_VALIDATOR_COUNT + ADVERSARIAL_VALIDATOR_COUNT; - let validators_keypairs = types::test_utils::generate_deterministic_keypairs(VALIDATOR_COUNT); let honest_validators: Vec = (0..HONEST_VALIDATOR_COUNT).collect(); let adversarial_validators: Vec = (HONEST_VALIDATOR_COUNT..VALIDATOR_COUNT).collect(); - let rig = BeaconChainHarness::builder(MinimalEthSpec) - .default_spec() - .keypairs(validators_keypairs) - .fresh_ephemeral_store() - .mock_execution_layer() - .build(); + let db_path = tempdir().unwrap(); + let store = get_store(&db_path); + let rig = get_harness(store.clone(), VALIDATOR_COUNT); let (state, state_root) = rig.get_current_state_and_root(); let canonical_slots_zeroth_epoch: Vec = @@ -2068,39 +2054,82 @@ async fn garbage_collect_temp_states_from_failed_block() { } #[tokio::test] -async fn weak_subjectivity_sync() { +async fn weak_subjectivity_sync_easy() { + let num_initial_slots = E::slots_per_epoch() * 11; + let checkpoint_slot = Slot::new(E::slots_per_epoch() * 9); + let slots = (1..num_initial_slots).map(Slot::new).collect(); + weak_subjectivity_sync_test(slots, checkpoint_slot).await +} + +#[tokio::test] +async fn weak_subjectivity_sync_unaligned_advanced_checkpoint() { + let num_initial_slots = E::slots_per_epoch() * 11; + let checkpoint_slot = Slot::new(E::slots_per_epoch() * 9); + let slots = (1..num_initial_slots) + .map(Slot::new) + .filter(|&slot| { + // Skip 3 slots leading up to the checkpoint slot. + slot <= checkpoint_slot - 3 || slot > checkpoint_slot + }) + .collect(); + weak_subjectivity_sync_test(slots, checkpoint_slot).await +} + +#[tokio::test] +async fn weak_subjectivity_sync_unaligned_unadvanced_checkpoint() { + let num_initial_slots = E::slots_per_epoch() * 11; + let checkpoint_slot = Slot::new(E::slots_per_epoch() * 9 - 3); + let slots = (1..num_initial_slots) + .map(Slot::new) + .filter(|&slot| { + // Skip 3 slots after the checkpoint slot. + slot <= checkpoint_slot || slot > checkpoint_slot + 3 + }) + .collect(); + weak_subjectivity_sync_test(slots, checkpoint_slot).await +} + +async fn weak_subjectivity_sync_test(slots: Vec, checkpoint_slot: Slot) { // Build an initial chain on one harness, representing a synced node with full history. - let num_initial_blocks = E::slots_per_epoch() * 11; let num_final_blocks = E::slots_per_epoch() * 2; let temp1 = tempdir().unwrap(); let full_store = get_store(&temp1); let harness = get_harness(full_store.clone(), LOW_VALIDATOR_COUNT); + let all_validators = (0..LOW_VALIDATOR_COUNT).collect::>(); + + let (genesis_state, genesis_state_root) = harness.get_current_state_and_root(); harness - .extend_chain( - num_initial_blocks as usize, - BlockStrategy::OnCanonicalHead, - AttestationStrategy::AllValidators, + .add_attested_blocks_at_slots( + genesis_state.clone(), + genesis_state_root, + &slots, + &all_validators, ) .await; - let genesis_state = full_store - .get_state(&harness.chain.genesis_state_root, Some(Slot::new(0))) + let wss_block_root = harness + .chain + .block_root_at_slot(checkpoint_slot, WhenSlotSkipped::Prev) .unwrap() .unwrap(); - let wss_checkpoint = harness.finalized_checkpoint(); + let wss_state_root = harness + .chain + .state_root_at_slot(checkpoint_slot) + .unwrap() + .unwrap(); + let wss_block = harness .chain .store - .get_full_block(&wss_checkpoint.root) + .get_full_block(&wss_block_root) .unwrap() .unwrap(); let wss_state = full_store - .get_state(&wss_block.state_root(), None) + .get_state(&wss_state_root, Some(checkpoint_slot)) .unwrap() .unwrap(); - let wss_slot = wss_block.slot(); // Add more blocks that advance finalization further. harness.advance_slot(); @@ -2130,19 +2159,25 @@ async fn weak_subjectivity_sync() { None, ); - // Initialise a new beacon chain from the finalized checkpoint + // Initialise a new beacon chain from the finalized checkpoint. + // The slot clock must be set to a time ahead of the checkpoint state. + let slot_clock = TestingSlotClock::new( + Slot::new(0), + Duration::from_secs(harness.chain.genesis_time), + Duration::from_secs(seconds_per_slot), + ); + slot_clock.set_slot(harness.get_current_slot().as_u64()); let beacon_chain = BeaconChainBuilder::>::new(MinimalEthSpec) .store(store.clone()) .custom_spec(test_spec::()) .task_executor(harness.chain.task_executor.clone()) + .logger(log.clone()) .weak_subjectivity_state(wss_state, wss_block.clone(), genesis_state) .unwrap() - .logger(log.clone()) .store_migrator_config(MigratorConfig::default().blocking()) .dummy_eth1_backend() .expect("should build dummy backend") - .testing_slot_clock(Duration::from_secs(seconds_per_slot)) - .expect("should configure testing slot clock") + .slot_clock(slot_clock) .shutdown_sender(shutdown_tx) .chain_config(ChainConfig::default()) .event_handler(Some(ServerSentEventHandler::new_with_capacity( @@ -2159,9 +2194,9 @@ async fn weak_subjectivity_sync() { // Apply blocks forward to reach head. let chain_dump = harness.chain.chain_dump().unwrap(); - let new_blocks = &chain_dump[wss_slot.as_usize() + 1..]; - - assert_eq!(new_blocks[0].beacon_block.slot(), wss_slot + 1); + let new_blocks = chain_dump + .iter() + .filter(|snapshot| snapshot.beacon_block.slot() > checkpoint_slot); for snapshot in new_blocks { let block_root = snapshot.beacon_block_root; @@ -2271,13 +2306,17 @@ async fn weak_subjectivity_sync() { assert_eq!(forwards, expected); // All blocks can be loaded. + let mut prev_block_root = Hash256::zero(); for (block_root, slot) in beacon_chain .forwards_iter_block_roots(Slot::new(0)) .unwrap() .map(Result::unwrap) { let block = store.get_blinded_block(&block_root).unwrap().unwrap(); - assert_eq!(block.slot(), slot); + if block_root != prev_block_root { + assert_eq!(block.slot(), slot); + } + prev_block_root = block_root; } // All states from the oldest state slot can be loaded. @@ -2292,14 +2331,141 @@ async fn weak_subjectivity_sync() { assert_eq!(state.canonical_root(), state_root); } - // Anchor slot is still set to the starting slot. - assert_eq!(store.get_anchor_slot(), Some(wss_slot)); + // Anchor slot is still set to the slot of the checkpoint block. + assert_eq!(store.get_anchor_slot(), Some(wss_block.slot())); // Reconstruct states. store.clone().reconstruct_historic_states().unwrap(); assert_eq!(store.get_anchor_slot(), None); } +/// Test that blocks and attestations that refer to states around an unaligned split state are +/// processed correctly. +#[tokio::test] +async fn process_blocks_and_attestations_for_unaligned_checkpoint() { + let temp = tempdir().unwrap(); + let store = get_store(&temp); + let chain_config = ChainConfig { + reconstruct_historic_states: false, + ..ChainConfig::default() + }; + let harness = get_harness_generic(store.clone(), LOW_VALIDATOR_COUNT, chain_config); + + let all_validators = (0..LOW_VALIDATOR_COUNT).collect::>(); + + let split_slot = Slot::new(E::slots_per_epoch() * 4); + let pre_skips = 1; + let post_skips = 1; + + // Build the chain up to the intended split slot, with 3 skips before the split. + let slots = (1..=split_slot.as_u64() - pre_skips) + .map(Slot::new) + .collect::>(); + + let (genesis_state, genesis_state_root) = harness.get_current_state_and_root(); + harness + .add_attested_blocks_at_slots( + genesis_state.clone(), + genesis_state_root, + &slots, + &all_validators, + ) + .await; + + // Before the split slot becomes finalized, create two forking blocks that build on the split + // block: + // + // - one that is invalid because it conflicts with finalization (slot <= finalized_slot) + // - one that is valid because its slot is not finalized (slot > finalized_slot) + let (unadvanced_split_state, unadvanced_split_state_root) = + harness.get_current_state_and_root(); + + let (invalid_fork_block, _) = harness + .make_block(unadvanced_split_state.clone(), split_slot) + .await; + let (valid_fork_block, _) = harness + .make_block(unadvanced_split_state.clone(), split_slot + 1) + .await; + + // Advance the chain so that the intended split slot is finalized. + // Do not attest in the epoch boundary slot, to make attestation production later easier (no + // equivocations). + let finalizing_slot = split_slot + 2 * E::slots_per_epoch(); + for _ in 0..pre_skips + post_skips { + harness.advance_slot(); + } + harness.extend_to_slot(finalizing_slot - 1).await; + harness + .add_block_at_slot(finalizing_slot, harness.get_current_state()) + .await + .unwrap(); + + // Check that the split slot is as intended. + let split = store.get_split_info(); + assert_eq!(split.slot, split_slot); + assert_eq!(split.block_root, valid_fork_block.parent_root()); + assert_ne!(split.state_root, unadvanced_split_state_root); + + // Applying the invalid block should fail. + let err = harness + .chain + .process_block( + invalid_fork_block.canonical_root(), + Arc::new(invalid_fork_block.clone()), + NotifyExecutionLayer::Yes, + || Ok(()), + ) + .await + .unwrap_err(); + assert!(matches!(err, BlockError::WouldRevertFinalizedSlot { .. })); + + // Applying the valid block should succeed, but it should not become head. + harness + .chain + .process_block( + valid_fork_block.canonical_root(), + Arc::new(valid_fork_block.clone()), + NotifyExecutionLayer::Yes, + || Ok(()), + ) + .await + .unwrap(); + harness.chain.recompute_head_at_current_slot().await; + assert_ne!(harness.head_block_root(), valid_fork_block.canonical_root()); + + // Attestations to the split block in the next 2 epochs should be processed successfully. + let attestation_start_slot = harness.get_current_slot(); + let attestation_end_slot = attestation_start_slot + 2 * E::slots_per_epoch(); + let (split_state_root, mut advanced_split_state) = harness + .chain + .store + .get_advanced_hot_state(split.block_root, split.slot, split.state_root) + .unwrap() + .unwrap(); + complete_state_advance( + &mut advanced_split_state, + Some(split_state_root), + attestation_start_slot, + &harness.chain.spec, + ) + .unwrap(); + advanced_split_state + .build_caches(&harness.chain.spec) + .unwrap(); + let advanced_split_state_root = advanced_split_state.update_tree_hash_cache().unwrap(); + for slot in (attestation_start_slot.as_u64()..attestation_end_slot.as_u64()).map(Slot::new) { + let attestations = harness.make_attestations( + &all_validators, + &advanced_split_state, + advanced_split_state_root, + split.block_root.into(), + slot, + ); + harness.advance_slot(); + harness.process_attestations(attestations); + } +} + #[tokio::test] async fn finalizes_after_resuming_from_db() { let validator_count = 16; @@ -2358,6 +2524,7 @@ async fn finalizes_after_resuming_from_db() { .default_spec() .keypairs(KEYPAIRS[0..validator_count].to_vec()) .resumed_disk_store(store) + .testing_slot_clock(original_chain.slot_clock.clone()) .execution_layer(original_chain.execution_layer.clone()) .build(); @@ -2611,6 +2778,9 @@ async fn schema_downgrade_to_min_version() { SchemaVersion(11) }; + // Save the slot clock so that the new harness doesn't revert in time. + let slot_clock = harness.chain.slot_clock.clone(); + // Close the database to ensure everything is written to disk. drop(store); drop(harness); @@ -2641,11 +2811,21 @@ async fn schema_downgrade_to_min_version() { ) .expect("schema upgrade from minimum version should work"); - // Rescreate the harness. + // Recreate the harness. + /* + let slot_clock = TestingSlotClock::new( + Slot::new(0), + Duration::from_secs(harness.chain.genesis_time), + Duration::from_secs(spec.seconds_per_slot), + ); + slot_clock.set_slot(harness.get_current_slot().as_u64()); + */ + let harness = BeaconChainHarness::builder(MinimalEthSpec) .default_spec() .keypairs(KEYPAIRS[0..LOW_VALIDATOR_COUNT].to_vec()) .logger(store.logger().clone()) + .testing_slot_clock(slot_clock) .resumed_disk_store(store.clone()) .mock_execution_layer() .build(); diff --git a/beacon_node/beacon_chain/tests/tests.rs b/beacon_node/beacon_chain/tests/tests.rs index 660bf41dc..4334f9083 100644 --- a/beacon_node/beacon_chain/tests/tests.rs +++ b/beacon_node/beacon_chain/tests/tests.rs @@ -6,7 +6,7 @@ use beacon_chain::{ AttestationStrategy, BeaconChainHarness, BlockStrategy, EphemeralHarnessType, OP_POOL_DB_KEY, }, - BeaconChain, NotifyExecutionLayer, StateSkipConfig, WhenSlotSkipped, + BeaconChain, ChainConfig, NotifyExecutionLayer, StateSkipConfig, WhenSlotSkipped, }; use lazy_static::lazy_static; use operation_pool::PersistedOperationPool; @@ -28,6 +28,10 @@ lazy_static! { fn get_harness(validator_count: usize) -> BeaconChainHarness> { let harness = BeaconChainHarness::builder(MinimalEthSpec) .default_spec() + .chain_config(ChainConfig { + reconstruct_historic_states: true, + ..ChainConfig::default() + }) .keypairs(KEYPAIRS[0..validator_count].to_vec()) .fresh_ephemeral_store() .mock_execution_layer() diff --git a/beacon_node/beacon_processor/src/lib.rs b/beacon_node/beacon_processor/src/lib.rs index 7b993f3cb..00108f084 100644 --- a/beacon_node/beacon_processor/src/lib.rs +++ b/beacon_node/beacon_processor/src/lib.rs @@ -743,7 +743,6 @@ impl Stream for InboundEvents { pub struct BeaconProcessor { pub network_globals: Arc>, pub executor: TaskExecutor, - pub max_workers: usize, pub current_workers: usize, pub config: BeaconProcessorConfig, pub log: Logger, @@ -756,7 +755,7 @@ impl BeaconProcessor { /// - Performed immediately, if a worker is available. /// - Queued for later processing, if no worker is currently available. /// - /// Only `self.max_workers` will ever be spawned at one time. Each worker is a `tokio` task + /// Only `self.config.max_workers` will ever be spawned at one time. Each worker is a `tokio` task /// started with `spawn_blocking`. /// /// The optional `work_journal_tx` allows for an outside process to receive a log of all work @@ -935,7 +934,7 @@ impl BeaconProcessor { let _ = work_journal_tx.try_send(id); } - let can_spawn = self.current_workers < self.max_workers; + let can_spawn = self.current_workers < self.config.max_workers; let drop_during_sync = work_event .as_ref() .map_or(false, |event| event.drop_during_sync); diff --git a/beacon_node/client/Cargo.toml b/beacon_node/client/Cargo.toml index 0b517930f..87e165090 100644 --- a/beacon_node/client/Cargo.toml +++ b/beacon_node/client/Cargo.toml @@ -44,4 +44,3 @@ slasher_service = { path = "../../slasher/service" } monitoring_api = {path = "../../common/monitoring_api"} execution_layer = { path = "../execution_layer" } beacon_processor = { path = "../beacon_processor" } -num_cpus = "1.13.0" diff --git a/beacon_node/client/src/builder.rs b/beacon_node/client/src/builder.rs index e1dc56204..54aa63935 100644 --- a/beacon_node/client/src/builder.rs +++ b/beacon_node/client/src/builder.rs @@ -30,8 +30,6 @@ use network::{NetworkConfig, NetworkSenders, NetworkService}; use slasher::Slasher; use slasher_service::SlasherService; use slog::{debug, info, warn, Logger}; -use state_processing::per_slot_processing; -use std::cmp; use std::net::TcpListener; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -161,7 +159,10 @@ where let context = runtime_context.service_context("beacon".into()); let spec = chain_spec.ok_or("beacon_chain_start_method requires a chain spec")?; let event_handler = if self.http_api_config.enabled { - Some(ServerSentEventHandler::new(context.log().clone())) + Some(ServerSentEventHandler::new( + context.log().clone(), + self.http_api_config.sse_capacity_multiplier, + )) } else { None }; @@ -316,7 +317,6 @@ where config.chain.checkpoint_sync_url_timeout, )), ); - let slots_per_epoch = TEthSpec::slots_per_epoch(); let deposit_snapshot = if config.sync_eth1_chain { // We want to fetch deposit snapshot before fetching the finalized beacon state to @@ -367,7 +367,7 @@ where context.log(), "Downloading finalized state"; ); - let mut state = remote + let state = remote .get_debug_beacon_states_ssz::(StateId::Finalized, &spec) .await .map_err(|e| format!("Error loading checkpoint state from remote: {:?}", e))? @@ -393,16 +393,6 @@ where debug!(context.log(), "Downloaded finalized block"); - let epoch_boundary_slot = state.slot() % slots_per_epoch; - if epoch_boundary_slot != 0 { - debug!(context.log(), "Advancing state to epoch boundary"; "state_slot" => state.slot(), "epoch_boundary_slot" => epoch_boundary_slot); - } - - while state.slot() % slots_per_epoch != 0 { - per_slot_processing(&mut state, None, &spec) - .map_err(|e| format!("Error advancing state: {:?}", e))?; - } - let genesis_state = BeaconState::from_ssz_bytes(&genesis_state_bytes, &spec) .map_err(|e| format!("Unable to parse genesis state SSZ: {:?}", e))?; @@ -775,7 +765,6 @@ where BeaconProcessor { network_globals: network_globals.clone(), executor: beacon_processor_context.executor.clone(), - max_workers: cmp::max(1, num_cpus::get()), current_workers: 0, config: beacon_processor_config, log: beacon_processor_context.log().clone(), diff --git a/beacon_node/client/src/config.rs b/beacon_node/client/src/config.rs index 72c918a28..4fac9b8e0 100644 --- a/beacon_node/client/src/config.rs +++ b/beacon_node/client/src/config.rs @@ -84,7 +84,6 @@ pub struct Config { pub monitoring_api: Option, pub slasher: Option, pub logger_config: LoggerConfig, - pub always_prefer_builder_payload: bool, pub beacon_processor: BeaconProcessorConfig, } @@ -114,7 +113,6 @@ impl Default for Config { validator_monitor_pubkeys: vec![], validator_monitor_individual_tracking_threshold: DEFAULT_INDIVIDUAL_TRACKING_THRESHOLD, logger_config: LoggerConfig::default(), - always_prefer_builder_payload: false, beacon_processor: <_>::default(), } } diff --git a/beacon_node/http_api/src/attester_duties.rs b/beacon_node/http_api/src/attester_duties.rs index aad405d56..f3242a2b3 100644 --- a/beacon_node/http_api/src/attester_duties.rs +++ b/beacon_node/http_api/src/attester_duties.rs @@ -35,7 +35,6 @@ pub fn attester_duties( .epoch(T::EthSpec::slots_per_epoch()); if request_epoch == current_epoch - || request_epoch == tolerant_current_epoch || request_epoch == current_epoch + 1 || request_epoch == tolerant_current_epoch + 1 { @@ -46,7 +45,7 @@ pub fn attester_duties( request_epoch, current_epoch ))) } else { - // request_epoch < current_epoch + // request_epoch < current_epoch, in fact we only allow `request_epoch == current_epoch-1` in this case compute_historic_attester_duties(request_epoch, request_indices, chain) } } diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs index 4a58a0dea..e45ad312e 100644 --- a/beacon_node/http_api/src/lib.rs +++ b/beacon_node/http_api/src/lib.rs @@ -66,7 +66,10 @@ use tokio::sync::{ mpsc::{Sender, UnboundedSender}, oneshot, }; -use tokio_stream::{wrappers::BroadcastStream, StreamExt}; +use tokio_stream::{ + wrappers::{errors::BroadcastStreamRecvError, BroadcastStream}, + StreamExt, +}; use types::{ Attestation, AttestationData, AttestationShufflingId, AttesterSlashing, BeaconStateError, BlindedPayload, CommitteeCache, ConfigAndPreset, Epoch, EthSpec, ForkName, FullPayload, @@ -132,6 +135,7 @@ pub struct Config { pub allow_sync_stalled: bool, pub spec_fork_name: Option, pub data_dir: PathBuf, + pub sse_capacity_multiplier: usize, pub enable_beacon_processor: bool, } @@ -146,6 +150,7 @@ impl Default for Config { allow_sync_stalled: false, spec_fork_name: None, data_dir: PathBuf::from(DEFAULT_ROOT_DIR), + sse_capacity_multiplier: 1, enable_beacon_processor: true, } } @@ -4373,22 +4378,29 @@ pub fn serve( } }; - receivers.push(BroadcastStream::new(receiver).map(|msg| { - match msg { - Ok(data) => Event::default() - .event(data.topic_name()) - .json_data(data) - .map_err(|e| { - warp_utils::reject::server_sent_event_error(format!( - "{:?}", - e - )) - }), - Err(e) => Err(warp_utils::reject::server_sent_event_error( - format!("{:?}", e), - )), - } - })); + receivers.push( + BroadcastStream::new(receiver) + .map(|msg| { + match msg { + Ok(data) => Event::default() + .event(data.topic_name()) + .json_data(data) + .unwrap_or_else(|e| { + Event::default() + .comment(format!("error - bad json: {e:?}")) + }), + // Do not terminate the stream if the channel fills + // up. Just drop some messages and send a comment to + // the client. + Err(BroadcastStreamRecvError::Lagged(n)) => { + Event::default().comment(format!( + "error - dropped {n} messages" + )) + } + } + }) + .map(Ok::<_, std::convert::Infallible>), + ); } } else { return Err(warp_utils::reject::custom_server_error( @@ -4398,7 +4410,7 @@ pub fn serve( let s = futures::stream::select_all(receivers); - Ok::<_, warp::Rejection>(warp::sse::reply(warp::sse::keep_alive().stream(s))) + Ok(warp::sse::reply(warp::sse::keep_alive().stream(s))) }) }, ); diff --git a/beacon_node/http_api/src/test_utils.rs b/beacon_node/http_api/src/test_utils.rs index 18c2df31d..86a967710 100644 --- a/beacon_node/http_api/src/test_utils.rs +++ b/beacon_node/http_api/src/test_utils.rs @@ -181,7 +181,14 @@ pub async fn create_api_server_on_port( let eth1_service = eth1::Service::new(eth1::Config::default(), log.clone(), chain.spec.clone()).unwrap(); - let beacon_processor_config = BeaconProcessorConfig::default(); + let beacon_processor_config = BeaconProcessorConfig { + // The number of workers must be greater than one. Tests which use the + // builder workflow sometimes require an internal HTTP request in order + // to fulfill an already in-flight HTTP request, therefore having only + // one worker will result in a deadlock. + max_workers: 2, + ..BeaconProcessorConfig::default() + }; let BeaconProcessorChannels { beacon_processor_tx, beacon_processor_rx, @@ -193,11 +200,6 @@ pub async fn create_api_server_on_port( BeaconProcessor { network_globals: network_globals.clone(), executor: test_runtime.task_executor.clone(), - // The number of workers must be greater than one. Tests which use the - // builder workflow sometimes require an internal HTTP request in order - // to fulfill an already in-flight HTTP request, therefore having only - // one worker will result in a deadlock. - max_workers: 2, current_workers: 0, config: beacon_processor_config, log: log.clone(), @@ -222,6 +224,7 @@ pub async fn create_api_server_on_port( allow_sync_stalled: false, data_dir: std::path::PathBuf::from(DEFAULT_ROOT_DIR), spec_fork_name: None, + sse_capacity_multiplier: 1, enable_beacon_processor: true, }, chain: Some(chain), diff --git a/beacon_node/http_api/tests/tests.rs b/beacon_node/http_api/tests/tests.rs index f89a5c244..16873578f 100644 --- a/beacon_node/http_api/tests/tests.rs +++ b/beacon_node/http_api/tests/tests.rs @@ -1,7 +1,7 @@ use beacon_chain::test_utils::RelativeSyncCommittee; use beacon_chain::{ test_utils::{AttestationStrategy, BeaconChainHarness, BlockStrategy, EphemeralHarnessType}, - BeaconChain, StateSkipConfig, WhenSlotSkipped, + BeaconChain, ChainConfig, StateSkipConfig, WhenSlotSkipped, }; use environment::null_logger; use eth2::{ @@ -77,6 +77,7 @@ struct ApiTester { struct ApiTesterConfig { spec: ChainSpec, + retain_historic_states: bool, builder_threshold: Option, } @@ -86,11 +87,19 @@ impl Default for ApiTesterConfig { spec.shard_committee_period = 2; Self { spec, + retain_historic_states: false, builder_threshold: None, } } } +impl ApiTesterConfig { + fn retain_historic_states(mut self) -> Self { + self.retain_historic_states = true; + self + } +} + impl ApiTester { pub async fn new() -> Self { // This allows for testing voluntary exits without building out a massive chain. @@ -118,6 +127,10 @@ impl ApiTester { let harness = Arc::new( BeaconChainHarness::builder(MainnetEthSpec) .spec(spec.clone()) + .chain_config(ChainConfig { + reconstruct_historic_states: config.retain_historic_states, + ..ChainConfig::default() + }) .logger(logging::test_logger()) .deterministic_keypairs(VALIDATOR_COUNT) .fresh_ephemeral_store() @@ -379,6 +392,7 @@ impl ApiTester { pub async fn new_mev_tester_no_builder_threshold() -> Self { let mut config = ApiTesterConfig { builder_threshold: Some(0), + retain_historic_states: false, spec: E::default_spec(), }; config.spec.altair_fork_epoch = Some(Epoch::new(0)); @@ -4807,7 +4821,7 @@ async fn get_validator_duties_attester_with_skip_slots() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn get_validator_duties_proposer() { - ApiTester::new() + ApiTester::new_from_config(ApiTesterConfig::default().retain_historic_states()) .await .test_get_validator_duties_proposer() .await; @@ -4815,7 +4829,7 @@ async fn get_validator_duties_proposer() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn get_validator_duties_proposer_with_skip_slots() { - ApiTester::new() + ApiTester::new_from_config(ApiTesterConfig::default().retain_historic_states()) .await .skip_slots(E::slots_per_epoch() * 2) .test_get_validator_duties_proposer() @@ -5147,6 +5161,7 @@ async fn builder_payload_chosen_by_profit() { async fn builder_works_post_capella() { let mut config = ApiTesterConfig { builder_threshold: Some(0), + retain_historic_states: false, spec: E::default_spec(), }; config.spec.altair_fork_epoch = Some(Epoch::new(0)); diff --git a/beacon_node/network/Cargo.toml b/beacon_node/network/Cargo.toml index 92ebbbea6..0ecd9e773 100644 --- a/beacon_node/network/Cargo.toml +++ b/beacon_node/network/Cargo.toml @@ -37,7 +37,6 @@ logging = { path = "../../common/logging" } task_executor = { path = "../../common/task_executor" } igd = "0.12.1" itertools = "0.10.0" -num_cpus = "1.13.0" lru_cache = { path = "../../common/lru_cache" } if-addrs = "0.6.4" strum = "0.24.0" diff --git a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs index 282149885..ad885b5fe 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -2302,7 +2302,7 @@ impl NetworkBeaconProcessor { ); } AttnError::BeaconChainError(BeaconChainError::DBError(Error::HotColdDBError( - HotColdDBError::AttestationStateIsFinalized { .. }, + HotColdDBError::FinalizedStateNotInHotDatabase { .. }, ))) => { debug!(self.log, "Attestation for finalized state"; "peer_id" => % peer_id); self.propagate_validation_result(message_id, peer_id, MessageAcceptance::Ignore); diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index 577287750..993551441 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -23,7 +23,6 @@ use lighthouse_network::{ Client, MessageId, NetworkGlobals, PeerId, Response, }; use slot_clock::SlotClock; -use std::cmp; use std::iter::Iterator; use std::sync::Arc; use std::time::Duration; @@ -235,7 +234,6 @@ impl TestRig { let beacon_processor = BeaconProcessor { network_globals, executor, - max_workers: cmp::max(1, num_cpus::get()), current_workers: 0, config: beacon_processor_config, log: log.clone(), diff --git a/beacon_node/src/cli.rs b/beacon_node/src/cli.rs index f112bfa73..def427532 100644 --- a/beacon_node/src/cli.rs +++ b/beacon_node/src/cli.rs @@ -389,6 +389,15 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> { stalled. This is useful for very small testnets. TESTING ONLY. DO NOT USE ON \ MAINNET.") ) + .arg( + Arg::with_name("http-sse-capacity-multiplier") + .long("http-sse-capacity-multiplier") + .takes_value(true) + .default_value("1") + .value_name("N") + .help("Multiplier to apply to the length of HTTP server-sent-event (SSE) channels. \ + Increasing this value can prevent messages from being dropped.") + ) .arg( Arg::with_name("http-enable-beacon-processor") .long("http-enable-beacon-processor") diff --git a/beacon_node/src/config.rs b/beacon_node/src/config.rs index 95fa44edc..28e3d2b7e 100644 --- a/beacon_node/src/config.rs +++ b/beacon_node/src/config.rs @@ -150,6 +150,9 @@ pub fn get_config( client_config.http_api.allow_sync_stalled = true; } + client_config.http_api.sse_capacity_multiplier = + parse_required(cli_args, "http-sse-capacity-multiplier")?; + client_config.http_api.enable_beacon_processor = parse_required(cli_args, "http-enable-beacon-processor")?; @@ -349,6 +352,8 @@ pub fn get_config( el_config.default_datadir = client_config.data_dir().clone(); el_config.builder_profit_threshold = clap_utils::parse_required(cli_args, "builder-profit-threshold")?; + el_config.always_prefer_builder_payload = + cli_args.is_present("always-prefer-builder-payload"); el_config.ignore_builder_override_suggestion_threshold = clap_utils::parse_required(cli_args, "ignore-builder-override-suggestion-threshold")?; let execution_timeout_multiplier = @@ -837,10 +842,6 @@ pub fn get_config( if cli_args.is_present("genesis-backfill") { client_config.chain.genesis_backfill = true; } - // Payload selection configs - if cli_args.is_present("always-prefer-builder-payload") { - client_config.always_prefer_builder_payload = true; - } // Backfill sync rate-limiting client_config.beacon_processor.enable_backfill_rate_limiting = diff --git a/beacon_node/store/src/hot_cold_store.rs b/beacon_node/store/src/hot_cold_store.rs index a05196664..e8d5da38b 100644 --- a/beacon_node/store/src/hot_cold_store.rs +++ b/beacon_node/store/src/hot_cold_store.rs @@ -14,7 +14,7 @@ use crate::memory_store::MemoryStore; use crate::metadata::{ AnchorInfo, BlobInfo, CompactionTimestamp, PruningCheckpoint, SchemaVersion, ANCHOR_INFO_KEY, BLOB_INFO_KEY, COMPACTION_TIMESTAMP_KEY, CONFIG_KEY, CURRENT_SCHEMA_VERSION, - PRUNING_CHECKPOINT_KEY, SCHEMA_VERSION_KEY, SPLIT_KEY, + PRUNING_CHECKPOINT_KEY, SCHEMA_VERSION_KEY, SPLIT_KEY, STATE_UPPER_LIMIT_NO_RETAIN, }; use crate::metrics; use crate::{ @@ -159,10 +159,10 @@ pub enum HotColdDBError { IterationError { unexpected_key: BytesKey, }, - AttestationStateIsFinalized { + FinalizedStateNotInHotDatabase { split_slot: Slot, - request_slot: Option, - state_root: Hash256, + request_slot: Slot, + block_root: Hash256, }, Rollback, } @@ -688,7 +688,7 @@ impl, Cold: ItemStore> HotColdDB /// upon that state (e.g., state roots). Additionally, only states from the hot store are /// returned. /// - /// See `Self::get_state` for information about `slot`. + /// See `Self::get_advanced_hot_state` for information about `max_slot`. /// /// ## Warning /// @@ -700,23 +700,78 @@ impl, Cold: ItemStore> HotColdDB /// - `state.block_roots` pub fn get_inconsistent_state_for_attestation_verification_only( &self, - state_root: &Hash256, - slot: Option, - ) -> Result>, Error> { + block_root: &Hash256, + max_slot: Slot, + state_root: Hash256, + ) -> Result)>, Error> { metrics::inc_counter(&metrics::BEACON_STATE_GET_COUNT); + self.get_advanced_hot_state_with_strategy( + *block_root, + max_slot, + state_root, + StateProcessingStrategy::Inconsistent, + ) + } - let split_slot = self.get_split_slot(); + /// Get a state with `latest_block_root == block_root` advanced through to at most `max_slot`. + /// + /// The `state_root` argument is used to look up the block's un-advanced state in case an + /// advanced state is not found. + /// + /// Return the `(result_state_root, state)` satisfying: + /// + /// - `result_state_root == state.canonical_root()` + /// - `state.slot() <= max_slot` + /// - `state.get_latest_block_root(result_state_root) == block_root` + /// + /// Presently this is only used to avoid loading the un-advanced split state, but in future will + /// be expanded to return states from an in-memory cache. + pub fn get_advanced_hot_state( + &self, + block_root: Hash256, + max_slot: Slot, + state_root: Hash256, + ) -> Result)>, Error> { + self.get_advanced_hot_state_with_strategy( + block_root, + max_slot, + state_root, + StateProcessingStrategy::Accurate, + ) + } - if slot.map_or(false, |slot| slot < split_slot) { - Err(HotColdDBError::AttestationStateIsFinalized { - split_slot, - request_slot: slot, - state_root: *state_root, + /// Same as `get_advanced_hot_state` but taking a `StateProcessingStrategy`. + pub fn get_advanced_hot_state_with_strategy( + &self, + block_root: Hash256, + max_slot: Slot, + state_root: Hash256, + state_processing_strategy: StateProcessingStrategy, + ) -> Result)>, Error> { + // Hold a read lock on the split point so it can't move while we're trying to load the + // state. + let split = self.split.read_recursive(); + + // Sanity check max-slot against the split slot. + if max_slot < split.slot { + return Err(HotColdDBError::FinalizedStateNotInHotDatabase { + split_slot: split.slot, + request_slot: max_slot, + block_root, } - .into()) - } else { - self.load_hot_state(state_root, StateProcessingStrategy::Inconsistent) + .into()); } + + let state_root = if block_root == split.block_root && split.slot <= max_slot { + split.state_root + } else { + state_root + }; + let state = self + .load_hot_state(&state_root, state_processing_strategy)? + .map(|state| (state_root, state)); + drop(split); + Ok(state) } /// Delete a state, ensuring it is removed from the LRU cache, as well as from on-disk. @@ -1434,8 +1489,12 @@ impl, Cold: ItemStore> HotColdDB *self.split.read_recursive() } - pub fn set_split(&self, slot: Slot, state_root: Hash256) { - *self.split.write() = Split { slot, state_root }; + pub fn set_split(&self, slot: Slot, state_root: Hash256, block_root: Hash256) { + *self.split.write() = Split { + slot, + state_root, + block_root, + }; } /// Fetch the slot of the most recently stored restore point. @@ -1470,25 +1529,36 @@ impl, Cold: ItemStore> HotColdDB } /// Initialise the anchor info for checkpoint sync starting from `block`. - pub fn init_anchor_info(&self, block: BeaconBlockRef<'_, E>) -> Result { + pub fn init_anchor_info( + &self, + block: BeaconBlockRef<'_, E>, + retain_historic_states: bool, + ) -> Result { let anchor_slot = block.slot(); let slots_per_restore_point = self.config.slots_per_restore_point; - // Set the `state_upper_limit` to the slot of the *next* restore point. - // See `get_state_upper_limit` for rationale. - let next_restore_point_slot = if anchor_slot % slots_per_restore_point == 0 { + let state_upper_limit = if !retain_historic_states { + STATE_UPPER_LIMIT_NO_RETAIN + } else if anchor_slot % slots_per_restore_point == 0 { anchor_slot } else { + // Set the `state_upper_limit` to the slot of the *next* restore point. + // See `get_state_upper_limit` for rationale. (anchor_slot / slots_per_restore_point + 1) * slots_per_restore_point }; - let anchor_info = AnchorInfo { - anchor_slot, - oldest_block_slot: anchor_slot, - oldest_block_parent: block.parent_root(), - state_upper_limit: next_restore_point_slot, - state_lower_limit: self.spec.genesis_slot, + let anchor_info = if state_upper_limit == 0 && anchor_slot == 0 { + // Genesis archive node: no anchor because we *will* store all states. + None + } else { + Some(AnchorInfo { + anchor_slot, + oldest_block_slot: anchor_slot, + oldest_block_parent: block.parent_root(), + state_upper_limit, + state_lower_limit: self.spec.genesis_slot, + }) }; - self.compare_and_set_anchor_info(None, Some(anchor_info)) + self.compare_and_set_anchor_info(None, anchor_info) } /// Get a clone of the store's anchor info. @@ -1667,11 +1737,26 @@ impl, Cold: ItemStore> HotColdDB self.hot_db.put(&CONFIG_KEY, &self.config.as_disk_config()) } - /// Load the split point from disk. - fn load_split(&self) -> Result, Error> { + /// Load the split point from disk, sans block root. + fn load_split_partial(&self) -> Result, Error> { self.hot_db.get(&SPLIT_KEY) } + /// Load the split point from disk, including block root. + fn load_split(&self) -> Result, Error> { + match self.load_split_partial()? { + Some(mut split) => { + // Load the hot state summary to get the block root. + let summary = self.load_hot_state_summary(&split.state_root)?.ok_or( + HotColdDBError::MissingSplitState(split.state_root, split.slot), + )?; + split.block_root = summary.latest_block_root; + Ok(Some(split)) + } + None => Ok(None), + } + } + /// Stage the split for storage to disk. pub fn store_split_in_batch(&self) -> KeyValueStoreOp { self.split.read_recursive().as_kv_store_op(SPLIT_KEY) @@ -2089,43 +2174,40 @@ impl, Cold: ItemStore> HotColdDB /// Advance the split point of the store, moving new finalized states to the freezer. pub fn migrate_database, Cold: ItemStore>( store: Arc>, - frozen_head_root: Hash256, - frozen_head: &BeaconState, + finalized_state_root: Hash256, + finalized_block_root: Hash256, + finalized_state: &BeaconState, ) -> Result<(), Error> { debug!( store.log, "Freezer migration started"; - "slot" => frozen_head.slot() + "slot" => finalized_state.slot() ); // 0. Check that the migration is sensible. - // The new frozen head must increase the current split slot, and lie on an epoch + // The new finalized state must increase the current split slot, and lie on an epoch // boundary (in order for the hot state summary scheme to work). let current_split_slot = store.split.read_recursive().slot; - let anchor_slot = store - .anchor_info - .read_recursive() - .as_ref() - .map(|a| a.anchor_slot); + let anchor_info = store.anchor_info.read_recursive().clone(); + let anchor_slot = anchor_info.as_ref().map(|a| a.anchor_slot); - if frozen_head.slot() < current_split_slot { + if finalized_state.slot() < current_split_slot { return Err(HotColdDBError::FreezeSlotError { current_split_slot, - proposed_split_slot: frozen_head.slot(), + proposed_split_slot: finalized_state.slot(), } .into()); } - if frozen_head.slot() % E::slots_per_epoch() != 0 { - return Err(HotColdDBError::FreezeSlotUnaligned(frozen_head.slot()).into()); + if finalized_state.slot() % E::slots_per_epoch() != 0 { + return Err(HotColdDBError::FreezeSlotUnaligned(finalized_state.slot()).into()); } let mut hot_db_ops: Vec> = Vec::new(); - // 1. Copy all of the states between the head and the split slot, from the hot DB + // 1. Copy all of the states between the new finalized state and the split slot, from the hot DB // to the cold DB. Delete the execution payloads of these now-finalized blocks. - let state_root_iter = RootsIterator::new(&store, frozen_head); - + let state_root_iter = RootsIterator::new(&store, finalized_state); for maybe_tuple in state_root_iter.take_while(|result| match result { Ok((_, _, slot)) => { slot >= ¤t_split_slot @@ -2135,6 +2217,29 @@ pub fn migrate_database, Cold: ItemStore>( }) { let (block_root, state_root, slot) = maybe_tuple?; + // Delete the execution payload if payload pruning is enabled. At a skipped slot we may + // delete the payload for the finalized block itself, but that's OK as we only guarantee + // that payloads are present for slots >= the split slot. The payload fetching code is also + // forgiving of missing payloads. + if store.config.prune_payloads { + hot_db_ops.push(StoreOp::DeleteExecutionPayload(block_root)); + } + + // Delete the old summary, and the full state if we lie on an epoch boundary. + hot_db_ops.push(StoreOp::DeleteState(state_root, Some(slot))); + + // Do not try to store states if a restore point is yet to be stored, or will never be + // stored (see `STATE_UPPER_LIMIT_NO_RETAIN`). Make an exception for the genesis state + // which always needs to be copied from the hot DB to the freezer and should not be deleted. + if slot != 0 + && anchor_info + .as_ref() + .map_or(false, |anchor| slot < anchor.state_upper_limit) + { + debug!(store.log, "Pruning finalized state"; "slot" => slot); + continue; + } + let mut cold_db_ops: Vec = Vec::new(); if slot % store.config.slots_per_restore_point == 0 { @@ -2153,17 +2258,6 @@ pub fn migrate_database, Cold: ItemStore>( // There are data dependencies between calls to `store_cold_state()` that prevent us from // doing one big call to `store.cold_db.do_atomically()` at end of the loop. store.cold_db.do_atomically(cold_db_ops)?; - - // Delete the old summary, and the full state if we lie on an epoch boundary. - hot_db_ops.push(StoreOp::DeleteState(state_root, Some(slot))); - - // Delete the execution payload if payload pruning is enabled. At a skipped slot we may - // delete the payload for the finalized block itself, but that's OK as we only guarantee - // that payloads are present for slots >= the split slot. The payload fetching code is also - // forgiving of missing payloads. - if store.config.prune_payloads { - hot_db_ops.push(StoreOp::DeleteExecutionPayload(block_root)); - } } // Warning: Critical section. We have to take care not to put any of the two databases in an @@ -2203,8 +2297,9 @@ pub fn migrate_database, Cold: ItemStore>( // Before updating the in-memory split value, we flush it to disk first, so that should the // OS process die at this point, we pick up from the right place after a restart. let split = Split { - slot: frozen_head.slot(), - state_root: frozen_head_root, + slot: finalized_state.slot(), + state_root: finalized_state_root, + block_root: finalized_block_root, }; store.hot_db.put_sync(&SPLIT_KEY, &split)?; @@ -2220,7 +2315,7 @@ pub fn migrate_database, Cold: ItemStore>( debug!( store.log, "Freezer migration complete"; - "slot" => frozen_head.slot() + "slot" => finalized_state.slot() ); Ok(()) @@ -2229,8 +2324,16 @@ pub fn migrate_database, Cold: ItemStore>( /// Struct for storing the split slot and state root in the database. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Encode, Decode, Deserialize, Serialize)] pub struct Split { - pub(crate) slot: Slot, - pub(crate) state_root: Hash256, + pub slot: Slot, + pub state_root: Hash256, + /// The block root of the split state. + /// + /// This is used to provide special handling for the split state in the case where there are + /// skipped slots. The split state will *always* be the advanced state, so callers + /// who only have the finalized block root should use `get_advanced_hot_state` to get this state, + /// rather than fetching `block.state_root()` (the unaligned state) which will have been pruned. + #[ssz(skip_serializing, skip_deserializing)] + pub block_root: Hash256, } impl StoreItem for Split { diff --git a/beacon_node/store/src/metadata.rs b/beacon_node/store/src/metadata.rs index 95e646ff5..6405aff1b 100644 --- a/beacon_node/store/src/metadata.rs +++ b/beacon_node/store/src/metadata.rs @@ -17,6 +17,9 @@ pub const COMPACTION_TIMESTAMP_KEY: Hash256 = Hash256::repeat_byte(4); pub const ANCHOR_INFO_KEY: Hash256 = Hash256::repeat_byte(5); pub const BLOB_INFO_KEY: Hash256 = Hash256::repeat_byte(6); +/// State upper limit value used to indicate that a node is not storing historic states. +pub const STATE_UPPER_LIMIT_NO_RETAIN: Slot = Slot::new(u64::MAX); + #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub struct SchemaVersion(pub u64); diff --git a/book/src/checkpoint-sync.md b/book/src/checkpoint-sync.md index 578838289..0c375a5f0 100644 --- a/book/src/checkpoint-sync.md +++ b/book/src/checkpoint-sync.md @@ -75,7 +75,7 @@ Once backfill is complete, a `INFO Historical block download complete` log will > Note: Since [v4.1.0](https://github.com/sigp/lighthouse/releases/tag/v4.1.0), Lighthouse implements rate-limited backfilling to mitigate validator performance issues after a recent checkpoint sync. This means that the speed at which historical blocks are downloaded is limited, typically to less than 20 slots/sec. This will not affect validator performance. However, if you would still prefer to sync the chain as fast as possible, you can add the flag `--disable-backfill-rate-limiting` to the beacon node. -> Note: Since [v4.2.0](https://github.com/sigp/lighthouse/releases/tag/v4.2.0), Lighthouse limits the backfill sync to only sync backwards to the weak subjectivity point (approximately 5 months). This will help to save disk space. However, if you would like to sync back to the genesis, you can add the flag `--genesis-backfill` to the beacon node. +> Note: Since [v4.2.0](https://github.com/sigp/lighthouse/releases/tag/v4.2.0), Lighthouse limits the backfill sync to only sync backwards to the weak subjectivity point (approximately 5 months). This will help to save disk space. However, if you would like to sync back to the genesis, you can add the flag `--genesis-backfill` to the beacon node. ## FAQ @@ -116,8 +116,9 @@ states: database. Additionally, the genesis block is always available. * `state_lower_limit`: All states with slots _less than or equal to_ this value are available in the database. The minimum value is 0, indicating that the genesis state is always available. -* `state_upper_limit`: All states with slots _greater than or equal to_ this value are available - in the database. +* `state_upper_limit`: All states with slots _greater than or equal to_ `min(split.slot, + state_upper_limit)` are available in the database. In the case where the `state_upper_limit` is + higher than the `split.slot`, this means states are not being written to the freezer database. Reconstruction runs from the state lower limit to the upper limit, narrowing the window of unavailable states as it goes. It will log messages like the following to show its progress: @@ -153,18 +154,8 @@ To manually specify a checkpoint use the following two flags: * `--checkpoint-state`: accepts an SSZ-encoded `BeaconState` blob * `--checkpoint-block`: accepts an SSZ-encoded `SignedBeaconBlock` blob -_Both_ the state and block must be provided and **must** adhere to the [Alignment -Requirements](#alignment-requirements) described below. - -### Alignment Requirements - -* The block must be a finalized block from an epoch boundary, i.e. `block.slot() % 32 == 0`. -* The state must be the state corresponding to `block` with `state.slot() == block.slot()` - and `state.hash_tree_root() == block.state_root()`. - -These requirements are imposed to align with Lighthouse's database schema, and notably exclude -finalized blocks from skipped slots. You can avoid alignment issues by using -[Automatic Checkpoint Sync](#automatic-checkpoint-sync), which will search for a suitable block -and state pair. +_Both_ the state and block must be provided and the state **must** match the block. The +state may be from the same slot as the block (unadvanced), or advanced to an epoch boundary, +in which case it will be assumed to be finalized at that epoch. [weak-subj]: https://blog.ethereum.org/2014/11/25/proof-stake-learned-love-weak-subjectivity/ diff --git a/book/src/setup.md b/book/src/setup.md index 533e1d463..1ae6e6354 100644 --- a/book/src/setup.md +++ b/book/src/setup.md @@ -9,7 +9,7 @@ particularly useful for development but still a good way to ensure you have the base dependencies. The additional requirements for developers are: -- [`anvil`](https://github.com/foundry-rs/foundry/tree/master/anvil). This is used to +- [`anvil`](https://github.com/foundry-rs/foundry/tree/master/crates/anvil). This is used to simulate the execution chain during tests. You'll get failures during tests if you don't have `anvil` available on your `PATH`. - [`cmake`](https://cmake.org/cmake/help/latest/command/install.html). Used by diff --git a/consensus/fork_choice/src/fork_choice.rs b/consensus/fork_choice/src/fork_choice.rs index d20c7eadb..5e9972884 100644 --- a/consensus/fork_choice/src/fork_choice.rs +++ b/consensus/fork_choice/src/fork_choice.rs @@ -751,7 +751,7 @@ where .unrealized_justified_checkpoint .zip(parent_block.unrealized_finalized_checkpoint) .filter(|(parent_justified, parent_finalized)| { - parent_justified.epoch == block_epoch && parent_finalized.epoch + 1 >= block_epoch + parent_justified.epoch == block_epoch && parent_finalized.epoch + 1 == block_epoch }); let (unrealized_justified_checkpoint, unrealized_finalized_checkpoint) = if let Some(( diff --git a/lighthouse/tests/beacon_node.rs b/lighthouse/tests/beacon_node.rs index fe181a68a..9b4069eca 100644 --- a/lighthouse/tests/beacon_node.rs +++ b/lighthouse/tests/beacon_node.rs @@ -366,21 +366,6 @@ fn genesis_backfill_with_historic_flag() { .with_config(|config| assert_eq!(config.chain.genesis_backfill, true)); } -#[test] -fn always_prefer_builder_payload_flag() { - CommandLineTest::new() - .flag("always-prefer-builder-payload", None) - .run_with_zero_port() - .with_config(|config| assert!(config.always_prefer_builder_payload)); -} - -#[test] -fn no_flag_sets_always_prefer_builder_payload_to_false() { - CommandLineTest::new() - .run_with_zero_port() - .with_config(|config| assert!(!config.always_prefer_builder_payload)); -} - // Tests for Eth1 flags. #[test] fn dummy_eth1_flag() { @@ -735,6 +720,38 @@ fn builder_fallback_flags() { ); }, ); + run_payload_builder_flag_test_with_config( + "builder", + "http://meow.cats", + Some("always-prefer-builder-payload"), + None, + |config| { + assert_eq!( + config + .execution_layer + .as_ref() + .unwrap() + .always_prefer_builder_payload, + true + ); + }, + ); + run_payload_builder_flag_test_with_config( + "builder", + "http://meow.cats", + None, + None, + |config| { + assert_eq!( + config + .execution_layer + .as_ref() + .unwrap() + .always_prefer_builder_payload, + false + ); + }, + ); run_payload_builder_flag_test_with_config( "builder", "http://meow.cats", @@ -2420,3 +2437,18 @@ fn beacon_processor_zero_workers() { .flag("beacon-processor-max-workers", Some("0")) .run_with_zero_port(); } + +#[test] +fn http_sse_capacity_multiplier_default() { + CommandLineTest::new() + .run_with_zero_port() + .with_config(|config| assert_eq!(config.http_api.sse_capacity_multiplier, 1)); +} + +#[test] +fn http_sse_capacity_multiplier_override() { + CommandLineTest::new() + .flag("http-sse-capacity-multiplier", Some("10")) + .run_with_zero_port() + .with_config(|config| assert_eq!(config.http_api.sse_capacity_multiplier, 10)); +} diff --git a/testing/antithesis/Dockerfile.libvoidstar b/testing/antithesis/Dockerfile.libvoidstar deleted file mode 100644 index c790e248d..000000000 --- a/testing/antithesis/Dockerfile.libvoidstar +++ /dev/null @@ -1,25 +0,0 @@ -FROM rust:1.68.2-bullseye AS builder -RUN apt-get update && apt-get -y upgrade && apt-get install -y cmake libclang-dev -COPY . lighthouse - -# Build lighthouse directly with a cargo build command, bypassing the Makefile. -RUN cd lighthouse && LD_LIBRARY_PATH=/lighthouse/testing/antithesis/libvoidstar/ RUSTFLAGS="-Cpasses=sancov-module -Cllvm-args=-sanitizer-coverage-level=3 -Cllvm-args=-sanitizer-coverage-trace-pc-guard -Ccodegen-units=1 -Cdebuginfo=2 -L/lighthouse/testing/antithesis/libvoidstar/ -lvoidstar" cargo build --release --manifest-path lighthouse/Cargo.toml --target x86_64-unknown-linux-gnu --features modern --verbose --bin lighthouse -# build lcli binary directly with cargo install command, bypassing the makefile -RUN cargo install --path /lighthouse/lcli --force --locked - -FROM ubuntu:latest -RUN apt-get update && apt-get -y upgrade && apt-get install -y --no-install-recommends \ - libssl-dev \ - ca-certificates \ - && apt-get clean \ - && rm -rf /var/lib/apt/lists/* - -# create and move the libvoidstar file -RUN mkdir libvoidstar -COPY --from=builder /lighthouse/testing/antithesis/libvoidstar/libvoidstar.so /usr/lib/libvoidstar.so - -# set the env variable to avoid having to always set it -ENV LD_LIBRARY_PATH=/usr/lib -# move the lighthouse binary and lcli binary -COPY --from=builder /lighthouse/target/x86_64-unknown-linux-gnu/release/lighthouse /usr/local/bin/lighthouse -COPY --from=builder /lighthouse/target/release/lcli /usr/local/bin/lcli \ No newline at end of file diff --git a/testing/antithesis/libvoidstar/libvoidstar.so b/testing/antithesis/libvoidstar/libvoidstar.so deleted file mode 100644 index 0f8a0f23c..000000000 Binary files a/testing/antithesis/libvoidstar/libvoidstar.so and /dev/null differ diff --git a/testing/ef_tests/src/cases/fork_choice.rs b/testing/ef_tests/src/cases/fork_choice.rs index 8c4a90044..f3afb5f7c 100644 --- a/testing/ef_tests/src/cases/fork_choice.rs +++ b/testing/ef_tests/src/cases/fork_choice.rs @@ -7,7 +7,7 @@ use beacon_chain::{ obtain_indexed_attestation_and_committees_per_slot, VerifiedAttestation, }, test_utils::{BeaconChainHarness, EphemeralHarnessType}, - BeaconChainTypes, CachedHead, NotifyExecutionLayer, + BeaconChainTypes, CachedHead, ChainConfig, NotifyExecutionLayer, }; use execution_layer::{json_structures::JsonPayloadStatusV1Status, PayloadStatusV1}; use serde::Deserialize; @@ -303,6 +303,10 @@ impl Tester { let harness = BeaconChainHarness::>::builder(E::default()) .spec(spec.clone()) .keypairs(vec![]) + .chain_config(ChainConfig { + reconstruct_historic_states: true, + ..ChainConfig::default() + }) .genesis_state_ephemeral_store(case.anchor_state.clone()) .mock_execution_layer() .recalculate_fork_times_with_genesis(0) diff --git a/testing/node_test_rig/src/lib.rs b/testing/node_test_rig/src/lib.rs index cad23800e..7d533ec7d 100644 --- a/testing/node_test_rig/src/lib.rs +++ b/testing/node_test_rig/src/lib.rs @@ -115,6 +115,9 @@ pub fn testing_client_config() -> ClientConfig { genesis_time: now, }; + // Simulator tests expect historic states to be available for post-run checks. + client_config.chain.reconstruct_historic_states = true; + // Specify a constant count of beacon processor workers. Having this number // too low can cause annoying HTTP timeouts, especially on Github runners // with 2 logical CPUs. diff --git a/validator_client/src/duties_service/sync.rs b/validator_client/src/duties_service/sync.rs index 1e66d947a..cf63d8ac6 100644 --- a/validator_client/src/duties_service/sync.rs +++ b/validator_client/src/duties_service/sync.rs @@ -2,8 +2,10 @@ use crate::beacon_node_fallback::{OfflineOnFailure, RequireSynced}; use crate::{ doppelganger_service::DoppelgangerStatus, duties_service::{DutiesService, Error}, + http_metrics::metrics, validator_store::Error as ValidatorStoreError, }; + use futures::future::join_all; use itertools::Itertools; use parking_lot::{MappedRwLockReadGuard, RwLock, RwLockReadGuard, RwLockWriteGuard}; @@ -426,6 +428,10 @@ pub async fn poll_sync_committee_duties_for_period TesterBuilder { let harness = BeaconChainHarness::builder(E::default()) .default_spec() + .chain_config(ChainConfig { + reconstruct_historic_states: true, + ..ChainConfig::default() + }) .deterministic_keypairs(VALIDATOR_COUNT) .fresh_ephemeral_store() .build();