Compute recent lightclient updates (sigp#4969)

* Compute recent lightclient updates

* Review PR

* Merge remote-tracking branch 'upstream/unstable' into lc-prod-recent-updates

* Review PR

* consistent naming

* add metrics

* revert dropping reprocessing queue

* Update light client optimistic update re-processing logic. (sigp#7)

* Add light client server simulator tests. Co-authored by @dapplion.

* Merge branch 'unstable' into fork/dapplion/lc-prod-recent-updates

* Fix lint

* Enable light client server in simulator test.

* Fix test for light client optimistic updates and finality updates.

Author: dapplion

Cargo.lock

@@ -38,6 +38,7 @@ use crate::light_client_finality_update_verification::{
 use crate::light_client_optimistic_update_verification::{
     Error as LightClientOptimisticUpdateError, VerifiedLightClientOptimisticUpdate,
 };
+use crate::light_client_server_cache::LightClientServerCache;
 use crate::migrate::BackgroundMigrator;
 use crate::naive_aggregation_pool::{
     AggregatedAttestationMap, Error as NaiveAggregationError, NaiveAggregationPool,
@@ -339,6 +340,8 @@ struct PartialBeaconBlock<E: EthSpec> {
     bls_to_execution_changes: Vec<SignedBlsToExecutionChange>,
 }
 
+pub type LightClientProducerEvent<T> = (Hash256, Slot, SyncAggregate<T>);
+
 pub type BeaconForkChoice<T> = ForkChoice<
     BeaconForkChoiceStore<
         <T as BeaconChainTypes>::EthSpec,
@@ -420,10 +423,6 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     /// Maintains a record of which validators we've seen BLS to execution changes for.
     pub(crate) observed_bls_to_execution_changes:
         Mutex<ObservedOperations<SignedBlsToExecutionChange, T::EthSpec>>,
-    /// The most recently validated light client finality update received on gossip.
-    pub latest_seen_finality_update: Mutex<Option<LightClientFinalityUpdate<T::EthSpec>>>,
-    /// The most recently validated light client optimistic update received on gossip.
-    pub latest_seen_optimistic_update: Mutex<Option<LightClientOptimisticUpdate<T::EthSpec>>>,
     /// Provides information from the Ethereum 1 (PoW) chain.
     pub eth1_chain: Option<Eth1Chain<T::Eth1Chain, T::EthSpec>>,
     /// Interfaces with the execution client.
@@ -466,6 +465,10 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     pub block_times_cache: Arc<RwLock<BlockTimesCache>>,
     /// A cache used to track pre-finalization block roots for quick rejection.
     pub pre_finalization_block_cache: PreFinalizationBlockCache,
+    /// A cache used to produce light_client server messages
+    pub light_client_server_cache: LightClientServerCache<T>,
+    /// Sender to signal the light_client server to produce new updates
+    pub light_client_server_tx: Option<Sender<LightClientProducerEvent<T::EthSpec>>>,
     /// Sender given to tasks, so that if they encounter a state in which execution cannot
     /// continue they can request that everything shuts down.
     pub shutdown_sender: Sender<ShutdownReason>,
@@ -1344,6 +1347,19 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         self.state_at_slot(load_slot, StateSkipConfig::WithoutStateRoots)
     }
 
+    pub fn recompute_and_cache_light_client_updates(
+        &self,
+        (parent_root, slot, sync_aggregate): LightClientProducerEvent<T::EthSpec>,
+    ) -> Result<(), Error> {
+        self.light_client_server_cache.recompute_and_cache_updates(
+            &self.log,
+            self.store.clone(),
+            &parent_root,
+            slot,
+            &sync_aggregate,
+        )
+    }
+
     /// Returns the current heads of the `BeaconChain`. For the canonical head, see `Self::head`.
     ///
     /// Returns `(block_root, block_slot)`.
@@ -3521,6 +3537,20 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         };
         let current_finalized_checkpoint = state.finalized_checkpoint();
 
+        // compute state proofs for light client updates before inserting the state into the
+        // snapshot cache.
+        if self.config.enable_light_client_server {
+            self.light_client_server_cache
+                .cache_state_data(
+                    &self.spec, block, block_root,
+                    // mutable reference on the state is needed to compute merkle proofs
+                    &mut state,
+                )
+                .unwrap_or_else(|e| {
+                    error!(self.log, "error caching light_client data {:?}", e);
+                });
+        }
+
         self.snapshot_cache
             .try_write_for(BLOCK_PROCESSING_CACHE_LOCK_TIMEOUT)
             .ok_or(Error::SnapshotCacheLockTimeout)
@@ -3893,6 +3923,28 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 }));
             }
         }
+
+        // Do not trigger light_client server update producer for old blocks, to extra work
+        // during sync.
+        if self.config.enable_light_client_server
+            && block_delay_total < self.slot_clock.slot_duration() * 32
+        {
+            if let Some(mut light_client_server_tx) = self.light_client_server_tx.clone() {
+                if let Ok(sync_aggregate) = block.body().sync_aggregate() {
+                    if let Err(e) = light_client_server_tx.try_send((
+                        block.parent_root(),
+                        block.slot(),
+                        sync_aggregate.clone(),
+                    )) {
+                        warn!(
+                            self.log,
+                            "Failed to send light_client server event";
+                            "error" => ?e
+                        );
+                    }
+                }
+            }
+        }
     }
 
     // For the current and next epoch of this state, ensure we have the shuffling from this

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -1,11 +1,14 @@
-use crate::beacon_chain::{CanonicalHead, BEACON_CHAIN_DB_KEY, ETH1_CACHE_DB_KEY, OP_POOL_DB_KEY};
+use crate::beacon_chain::{
+    CanonicalHead, LightClientProducerEvent, BEACON_CHAIN_DB_KEY, ETH1_CACHE_DB_KEY, OP_POOL_DB_KEY,
+};
 use crate::beacon_proposer_cache::BeaconProposerCache;
 use crate::data_availability_checker::DataAvailabilityChecker;
 use crate::eth1_chain::{CachingEth1Backend, SszEth1};
 use crate::eth1_finalization_cache::Eth1FinalizationCache;
 use crate::fork_choice_signal::ForkChoiceSignalTx;
 use crate::fork_revert::{reset_fork_choice_to_finalization, revert_to_fork_boundary};
 use crate::head_tracker::HeadTracker;
+use crate::light_client_server_cache::LightClientServerCache;
 use crate::migrate::{BackgroundMigrator, MigratorConfig};
 use crate::persisted_beacon_chain::PersistedBeaconChain;
 use crate::shuffling_cache::{BlockShufflingIds, ShufflingCache};
@@ -87,6 +90,7 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
     event_handler: Option<ServerSentEventHandler<T::EthSpec>>,
     slot_clock: Option<T::SlotClock>,
     shutdown_sender: Option<Sender<ShutdownReason>>,
+    light_client_server_tx: Option<Sender<LightClientProducerEvent<T::EthSpec>>>,
     head_tracker: Option<HeadTracker>,
     validator_pubkey_cache: Option<ValidatorPubkeyCache<T>>,
     spec: ChainSpec,
@@ -129,6 +133,7 @@ where
             event_handler: None,
             slot_clock: None,
             shutdown_sender: None,
+            light_client_server_tx: None,
             head_tracker: None,
             validator_pubkey_cache: None,
             spec: TEthSpec::default_spec(),
@@ -603,6 +608,15 @@ where
         self
     }
 
+    /// Sets a `Sender` to allow the beacon chain to trigger light_client update production.
+    pub fn light_client_server_tx(
+        mut self,
+        sender: Sender<LightClientProducerEvent<TEthSpec>>,
+    ) -> Self {
+        self.light_client_server_tx = Some(sender);
+        self
+    }
+
     /// Creates a new, empty operation pool.
     fn empty_op_pool(mut self) -> Self {
         self.op_pool = Some(OperationPool::new());
@@ -887,8 +901,6 @@ where
             observed_proposer_slashings: <_>::default(),
             observed_attester_slashings: <_>::default(),
             observed_bls_to_execution_changes: <_>::default(),
-            latest_seen_finality_update: <_>::default(),
-            latest_seen_optimistic_update: <_>::default(),
             eth1_chain: self.eth1_chain,
             execution_layer: self.execution_layer,
             genesis_validators_root,
@@ -916,6 +928,8 @@ where
             validator_pubkey_cache: TimeoutRwLock::new(validator_pubkey_cache),
             attester_cache: <_>::default(),
             early_attester_cache: <_>::default(),
+            light_client_server_cache: LightClientServerCache::new(),
+            light_client_server_tx: self.light_client_server_tx,
             shutdown_sender: self
                 .shutdown_sender
                 .ok_or("Cannot build without a shutdown sender.")?,

beacon_node/beacon_chain/src/builder.rs

@@ -83,6 +83,8 @@ pub struct ChainConfig {
     pub progressive_balances_mode: ProgressiveBalancesMode,
     /// Number of epochs between each migration of data from the hot database to the freezer.
     pub epochs_per_migration: u64,
+    /// When set to true Light client server computes and caches state proofs for serving updates
+    pub enable_light_client_server: bool,
 }
 
 impl Default for ChainConfig {
@@ -114,6 +116,7 @@ impl Default for ChainConfig {
             always_prepare_payload: false,
             progressive_balances_mode: ProgressiveBalancesMode::Fast,
             epochs_per_migration: crate::migrate::DEFAULT_EPOCHS_PER_MIGRATION,
+            enable_light_client_server: false,
         }
     }
 }

beacon_node/beacon_chain/src/chain_config.rs

@@ -32,6 +32,7 @@ pub mod historical_blocks;
 pub mod kzg_utils;
 pub mod light_client_finality_update_verification;
 pub mod light_client_optimistic_update_verification;
+mod light_client_server_cache;
 pub mod merge_readiness;
 pub mod metrics;
 pub mod migrate;
@@ -61,8 +62,8 @@ pub mod validator_pubkey_cache;
 pub use self::beacon_chain::{
     AttestationProcessingOutcome, AvailabilityProcessingStatus, BeaconBlockResponse,
     BeaconBlockResponseWrapper, BeaconChain, BeaconChainTypes, BeaconStore, ChainSegmentResult,
-    ForkChoiceError, OverrideForkchoiceUpdate, ProduceBlockVerification, StateSkipConfig,
-    WhenSlotSkipped, INVALID_FINALIZED_MERGE_TRANSITION_BLOCK_SHUTDOWN_REASON,
+    ForkChoiceError, LightClientProducerEvent, OverrideForkchoiceUpdate, ProduceBlockVerification,
+    StateSkipConfig, WhenSlotSkipped, INVALID_FINALIZED_MERGE_TRANSITION_BLOCK_SHUTDOWN_REASON,
     INVALID_JUSTIFIED_PAYLOAD_SHUTDOWN_REASON,
 };
 pub use self::beacon_snapshot::BeaconSnapshot;

beacon_node/beacon_chain/src/lib.rs

@@ -1,11 +1,9 @@
-use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
+use crate::{BeaconChain, BeaconChainTypes};
 use derivative::Derivative;
 use slot_clock::SlotClock;
 use std::time::Duration;
 use strum::AsRefStr;
-use types::{
-    light_client_update::Error as LightClientUpdateError, LightClientFinalityUpdate, Slot,
-};
+use types::LightClientFinalityUpdate;
 
 /// Returned when a light client finality update was not successfully verified. It might not have been verified for
 /// two reasons:
@@ -16,8 +14,6 @@ use types::{
 ///   (the `BeaconChainError` variant)
 #[derive(Debug, AsRefStr)]
 pub enum Error {
-    /// Light client finality update message with a lower or equal finalized_header slot already forwarded.
-    FinalityUpdateAlreadySeen,
     /// The light client finality message was received is prior to one-third of slot duration passage. (with
     /// respect to the gossip clock disparity and slot clock duration).
     ///
@@ -26,29 +22,11 @@ pub enum Error {
     /// Assuming the local clock is correct, the peer has sent an invalid message.
     TooEarly,
     /// Light client finality update message does not match the locally constructed one.
-    ///
-    /// ## Peer Scoring
-    ///
     InvalidLightClientFinalityUpdate,
     /// Signature slot start time is none.
     SigSlotStartIsNone,
     /// Failed to construct a LightClientFinalityUpdate from state.
     FailedConstructingUpdate,
-    /// Beacon chain error occurred.
-    BeaconChainError(BeaconChainError),
-    LightClientUpdateError(LightClientUpdateError),
-}
-
-impl From<BeaconChainError> for Error {
-    fn from(e: BeaconChainError) -> Self {
-        Error::BeaconChainError(e)
-    }
-}
-
-impl From<LightClientUpdateError> for Error {
-    fn from(e: LightClientUpdateError) -> Self {
-        Error::LightClientUpdateError(e)
-    }
 }
 
 /// Wraps a `LightClientFinalityUpdate` that has been verified for propagation on the gossip network.
@@ -63,71 +41,34 @@ impl<T: BeaconChainTypes> VerifiedLightClientFinalityUpdate<T> {
     /// Returns `Ok(Self)` if the `light_client_finality_update` is valid to be (re)published on the gossip
     /// network.
     pub fn verify(
-        light_client_finality_update: LightClientFinalityUpdate<T::EthSpec>,
+        rcv_finality_update: LightClientFinalityUpdate<T::EthSpec>,
         chain: &BeaconChain<T>,
         seen_timestamp: Duration,
     ) -> Result<Self, Error> {
-        let gossiped_finality_slot = light_client_finality_update.finalized_header.beacon.slot;
-        let one_third_slot_duration = Duration::new(chain.spec.seconds_per_slot / 3, 0);
-        let signature_slot = light_client_finality_update.signature_slot;
-        let start_time = chain.slot_clock.start_of(signature_slot);
-        let mut latest_seen_finality_update = chain.latest_seen_finality_update.lock();
-
-        let head = chain.canonical_head.cached_head();
-        let head_block = &head.snapshot.beacon_block;
-        let attested_block_root = head_block.message().parent_root();
-        let attested_block = chain
-            .get_blinded_block(&attested_block_root)?
-            .ok_or(Error::FailedConstructingUpdate)?;
-        let mut attested_state = chain
-            .get_state(&attested_block.state_root(), Some(attested_block.slot()))?
-            .ok_or(Error::FailedConstructingUpdate)?;
-
-        let finalized_block_root = attested_state.finalized_checkpoint().root;
-        let finalized_block = chain
-            .get_blinded_block(&finalized_block_root)?
-            .ok_or(Error::FailedConstructingUpdate)?;
-        let latest_seen_finality_update_slot = match latest_seen_finality_update.as_ref() {
-            Some(update) => update.finalized_header.beacon.slot,
-            None => Slot::new(0),
-        };
-
-        // verify that no other finality_update with a lower or equal
-        // finalized_header.slot was already forwarded on the network
-        if gossiped_finality_slot <= latest_seen_finality_update_slot {
-            return Err(Error::FinalityUpdateAlreadySeen);
-        }
-
         // verify that enough time has passed for the block to have been propagated
-        match start_time {
-            Some(time) => {
-                if seen_timestamp + chain.spec.maximum_gossip_clock_disparity()
-                    < time + one_third_slot_duration
-                {
-                    return Err(Error::TooEarly);
-                }
-            }
-            None => return Err(Error::SigSlotStartIsNone),
+        let start_time = chain
+            .slot_clock
+            .start_of(rcv_finality_update.signature_slot)
+            .ok_or(Error::SigSlotStartIsNone)?;
+        let one_third_slot_duration = Duration::new(chain.spec.seconds_per_slot / 3, 0);
+        if seen_timestamp + chain.spec.maximum_gossip_clock_disparity()
+            < start_time + one_third_slot_duration
+        {
+            return Err(Error::TooEarly);
         }
 
-        let head_state = &head.snapshot.beacon_state;
-        let finality_update = LightClientFinalityUpdate::new(
-            &chain.spec,
-            head_state,
-            head_block,
-            &mut attested_state,
-            &finalized_block,
-        )?;
+        let latest_finality_update = chain
+            .light_client_server_cache
+            .get_latest_finality_update()
+            .ok_or(Error::FailedConstructingUpdate)?;
 
         // verify that the gossiped finality update is the same as the locally constructed one.
-        if finality_update != light_client_finality_update {
+        if latest_finality_update != rcv_finality_update {
             return Err(Error::InvalidLightClientFinalityUpdate);
         }
 
-        *latest_seen_finality_update = Some(light_client_finality_update.clone());
-
         Ok(Self {
-            light_client_finality_update,
+            light_client_finality_update: rcv_finality_update,
             seen_timestamp,
         })
     }