Remove CountUnrealized (#4357)

## Issue Addressed

Closes #4332

## Proposed Changes

Remove the `CountUnrealized` type, defaulting unrealized justification to _on_. This fixes the #4332 issue by ensuring that importing the same block to fork choice always results in the same outcome.

Finalized sync speed may be slightly impacted by this change, but that is deemed an acceptable trade-off until the optimisation from #4118 is implemented.

TODO:

- [x] Also check that the block isn't a duplicate before importing

beacon_node/network/src/beacon_processor/worker/gossip_methods.rs

@@ -8,8 +8,8 @@ use beacon_chain::{
     observed_operations::ObservationOutcome,
     sync_committee_verification::{self, Error as SyncCommitteeError},
     validator_monitor::get_block_delay_ms,
-    BeaconChainError, BeaconChainTypes, BlockError, CountUnrealized, ForkChoiceError,
-    GossipVerifiedBlock, NotifyExecutionLayer,
+    BeaconChainError, BeaconChainTypes, BlockError, ForkChoiceError, GossipVerifiedBlock,
+    NotifyExecutionLayer,
 };
 use lighthouse_network::{Client, MessageAcceptance, MessageId, PeerAction, PeerId, ReportSource};
 use operation_pool::ReceivedPreCapella;
@@ -949,12 +949,7 @@ impl<T: BeaconChainTypes> Worker<T> {
 
         let result = self
             .chain
-            .process_block(
-                block_root,
-                verified_block,
-                CountUnrealized::True,
-                NotifyExecutionLayer::Yes,
-            )
+            .process_block(block_root, verified_block, NotifyExecutionLayer::Yes)
             .await;
 
         match &result {

beacon_node/network/src/beacon_processor/worker/sync_methods.rs

@@ -7,7 +7,6 @@ use crate::beacon_processor::DuplicateCache;
 use crate::metrics;
 use crate::sync::manager::{BlockProcessType, SyncMessage};
 use crate::sync::{BatchProcessResult, ChainId};
-use beacon_chain::CountUnrealized;
 use beacon_chain::{
     observed_block_producers::Error as ObserveError, validator_monitor::get_block_delay_ms,
     BeaconChainError, BeaconChainTypes, BlockError, ChainSegmentResult, HistoricalBlockError,
@@ -25,7 +24,7 @@ use types::{Epoch, Hash256, SignedBeaconBlock};
 #[derive(Clone, Debug, PartialEq)]
 pub enum ChainSegmentProcessId {
     /// Processing Id of a range syncing batch.
-    RangeBatchId(ChainId, Epoch, CountUnrealized),
+    RangeBatchId(ChainId, Epoch),
     /// Processing ID for a backfill syncing batch.
     BackSyncBatchId(Epoch),
     /// Processing Id of the parent lookup of a block.
@@ -166,12 +165,7 @@ impl<T: BeaconChainTypes> Worker<T> {
         let parent_root = block.message().parent_root();
         let result = self
             .chain
-            .process_block(
-                block_root,
-                block,
-                CountUnrealized::True,
-                NotifyExecutionLayer::Yes,
-            )
+            .process_block(block_root, block, NotifyExecutionLayer::Yes)
             .await;
 
         metrics::inc_counter(&metrics::BEACON_PROCESSOR_RPC_BLOCK_IMPORTED_TOTAL);
@@ -220,17 +214,13 @@ impl<T: BeaconChainTypes> Worker<T> {
     ) {
         let result = match sync_type {
             // this a request from the range sync
-            ChainSegmentProcessId::RangeBatchId(chain_id, epoch, count_unrealized) => {
+            ChainSegmentProcessId::RangeBatchId(chain_id, epoch) => {
                 let start_slot = downloaded_blocks.first().map(|b| b.slot().as_u64());
                 let end_slot = downloaded_blocks.last().map(|b| b.slot().as_u64());
                 let sent_blocks = downloaded_blocks.len();
 
                 match self
-                    .process_blocks(
-                        downloaded_blocks.iter(),
-                        count_unrealized,
-                        notify_execution_layer,
-                    )
+                    .process_blocks(downloaded_blocks.iter(), notify_execution_layer)
                     .await
                 {
                     (_, Ok(_)) => {
@@ -309,11 +299,7 @@ impl<T: BeaconChainTypes> Worker<T> {
                 // parent blocks are ordered from highest slot to lowest, so we need to process in
                 // reverse
                 match self
-                    .process_blocks(
-                        downloaded_blocks.iter().rev(),
-                        CountUnrealized::True,
-                        notify_execution_layer,
-                    )
+                    .process_blocks(downloaded_blocks.iter().rev(), notify_execution_layer)
                     .await
                 {
                     (imported_blocks, Err(e)) => {
@@ -343,13 +329,12 @@ impl<T: BeaconChainTypes> Worker<T> {
     async fn process_blocks<'a>(
         &self,
         downloaded_blocks: impl Iterator<Item = &'a Arc<SignedBeaconBlock<T::EthSpec>>>,
-        count_unrealized: CountUnrealized,
         notify_execution_layer: NotifyExecutionLayer,
     ) -> (usize, Result<(), ChainSegmentFailed>) {
         let blocks: Vec<Arc<_>> = downloaded_blocks.cloned().collect();
         match self
             .chain
-            .process_chain_segment(blocks, count_unrealized, notify_execution_layer)
+            .process_chain_segment(blocks, notify_execution_layer)
             .await
         {
             ChainSegmentResult::Successful { imported_blocks } => {

beacon_node/network/src/sync/manager.rs

@@ -556,7 +556,7 @@ impl<T: BeaconChainTypes> SyncManager<T> {
                     .parent_block_processed(chain_hash, result, &mut self.network),
             },
             SyncMessage::BatchProcessed { sync_type, result } => match sync_type {
-                ChainSegmentProcessId::RangeBatchId(chain_id, epoch, _) => {
+                ChainSegmentProcessId::RangeBatchId(chain_id, epoch) => {
                     self.range_sync.handle_block_process_result(
                         &mut self.network,
                         chain_id,

beacon_node/network/src/sync/range_sync/chain.rs

@@ -3,7 +3,7 @@ use crate::beacon_processor::{ChainSegmentProcessId, WorkEvent as BeaconWorkEven
 use crate::sync::{
     manager::Id, network_context::SyncNetworkContext, BatchOperationOutcome, BatchProcessResult,
 };
-use beacon_chain::{BeaconChainTypes, CountUnrealized};
+use beacon_chain::BeaconChainTypes;
 use fnv::FnvHashMap;
 use lighthouse_network::{PeerAction, PeerId};
 use rand::seq::SliceRandom;
@@ -101,8 +101,6 @@ pub struct SyncingChain<T: BeaconChainTypes> {
     /// Batches validated by this chain.
     validated_batches: u64,
 
-    is_finalized_segment: bool,
-
     /// The chain's log.
     log: slog::Logger,
 }
@@ -128,7 +126,6 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
         target_head_slot: Slot,
         target_head_root: Hash256,
         peer_id: PeerId,
-        is_finalized_segment: bool,
         log: &slog::Logger,
     ) -> Self {
         let mut peers = FnvHashMap::default();
@@ -150,7 +147,6 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
             state: ChainSyncingState::Stopped,
             current_processing_batch: None,
             validated_batches: 0,
-            is_finalized_segment,
             log: log.new(o!("chain" => id)),
         }
     }
@@ -318,12 +314,7 @@ impl<T: BeaconChainTypes> SyncingChain<T> {
         // for removing chains and checking completion is in the callback.
 
         let blocks = batch.start_processing()?;
-        let count_unrealized = if self.is_finalized_segment {
-            CountUnrealized::False
-        } else {
-            CountUnrealized::True
-        };
-        let process_id = ChainSegmentProcessId::RangeBatchId(self.id, batch_id, count_unrealized);
+        let process_id = ChainSegmentProcessId::RangeBatchId(self.id, batch_id);
         self.current_processing_batch = Some(batch_id);
 
         if let Err(e) =

beacon_node/network/src/sync/range_sync/chain_collection.rs

@@ -465,10 +465,10 @@ impl<T: BeaconChainTypes, C: BlockStorage> ChainCollection<T, C> {
         network: &mut SyncNetworkContext<T>,
     ) {
         let id = SyncingChain::<T>::id(&target_head_root, &target_head_slot);
-        let (collection, is_finalized) = if let RangeSyncType::Finalized = sync_type {
-            (&mut self.finalized_chains, true)
+        let collection = if let RangeSyncType::Finalized = sync_type {
+            &mut self.finalized_chains
         } else {
-            (&mut self.head_chains, false)
+            &mut self.head_chains
         };
         match collection.entry(id) {
             Entry::Occupied(mut entry) => {
@@ -493,7 +493,6 @@ impl<T: BeaconChainTypes, C: BlockStorage> ChainCollection<T, C> {
                     target_head_slot,
                     target_head_root,
                     peer,
-                    is_finalized,
                     &self.log,
                 );
                 debug_assert_eq!(new_chain.get_id(), id);