gossip light client data durnig sync committee contributions

Author: eserilev

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -2187,6 +2187,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         })
     }
 
+    pub fn get_latest_finality_update(&self) -> Option<LightClientFinalityUpdate<T::EthSpec>> {
+        self.light_client_server_cache.get_latest_finality_update()
+    }
+
+    pub fn get_latest_optimistic_update(&self) -> Option<LightClientOptimisticUpdate<T::EthSpec>> {
+        self.light_client_server_cache
+            .get_latest_optimistic_update()
+    }
+
     /// Accepts some 'LightClientFinalityUpdate' from the network and attempts to verify it
     pub fn verify_finality_update_for_gossip(
         self: &Arc<Self>,

beacon_node/beacon_chain/src/light_client_server_cache.rs

@@ -40,6 +40,10 @@ pub struct LightClientServerCache<T: BeaconChainTypes> {
     latest_written_current_sync_committee: RwLock<Option<Arc<SyncCommittee<T::EthSpec>>>>,
     /// Caches state proofs by block root
     prev_block_cache: Mutex<lru::LruCache<Hash256, LightClientCachedData<T::EthSpec>>>,
+    /// Tracks the latest broadcasted finality update
+    latest_broadcasted_finality_update: RwLock<Option<LightClientFinalityUpdate<T::EthSpec>>>,
+    /// Tracks the latest broadcasted optimistic update
+    latest_broadcasted_optimistic_update: RwLock<Option<LightClientOptimisticUpdate<T::EthSpec>>>,
 }
 
 impl<T: BeaconChainTypes> LightClientServerCache<T> {
@@ -49,6 +53,8 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
             latest_optimistic_update: None.into(),
             latest_light_client_update: None.into(),
             latest_written_current_sync_committee: None.into(),
+            latest_broadcasted_finality_update: None.into(),
+            latest_broadcasted_optimistic_update: None.into(),
             prev_block_cache: lru::LruCache::new(PREV_BLOCK_CACHE_SIZE).into(),
         }
     }
@@ -333,10 +339,66 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
         Ok(new_value)
     }
 
+    pub fn should_broadcast_latest_finality_update(
+        &self,
+    ) -> Option<LightClientFinalityUpdate<T::EthSpec>> {
+        if let Some(latest_finality_update) = self.get_latest_finality_update() {
+            let latest_broadcasted_finality_update = self
+                .latest_broadcasted_finality_update
+                .read()
+                .clone()
+                .take();
+            match latest_broadcasted_finality_update {
+                Some(latest_broadcasted_finality_update) => {
+                    if latest_broadcasted_finality_update != latest_finality_update {
+                        *self.latest_broadcasted_finality_update.write() =
+                            Some(latest_finality_update.clone());
+                        return Some(latest_finality_update);
+                    }
+                }
+                None => {
+                    *self.latest_broadcasted_finality_update.write() =
+                        Some(latest_finality_update.clone());
+                    return Some(latest_finality_update);
+                }
+            }
+        }
+
+        return None;
+    }
+
     pub fn get_latest_finality_update(&self) -> Option<LightClientFinalityUpdate<T::EthSpec>> {
         self.latest_finality_update.read().clone()
     }
 
+    /// Checks if we've already broadcasted the latest optimistic update.
+    /// If we haven't, update the `latest_broadcasted_optimistic_update` cache
+    /// and return the latest optimistic update for broadcasting
+    pub fn should_broadcast_latest_optimistic_update(
+        &self,
+    ) -> Option<LightClientOptimisticUpdate<T::EthSpec>> {
+        if let Some(latest_optimistic_update) = self.get_latest_optimistic_update() {
+            let latest_broadcasted_optimistic_update =
+                self.latest_optimistic_update.read().clone().take();
+            match latest_broadcasted_optimistic_update {
+                Some(latest_broadcasted_optimistic_update) => {
+                    if latest_broadcasted_optimistic_update != latest_optimistic_update {
+                        *self.latest_broadcasted_optimistic_update.write() =
+                            Some(latest_optimistic_update.clone());
+                        return Some(latest_optimistic_update);
+                    }
+                }
+                None => {
+                    *self.latest_broadcasted_optimistic_update.write() =
+                        Some(latest_optimistic_update.clone());
+                    return Some(latest_optimistic_update);
+                }
+            }
+        }
+
+        return None;
+    }
+
     pub fn get_latest_optimistic_update(&self) -> Option<LightClientOptimisticUpdate<T::EthSpec>> {
         self.latest_optimistic_update.read().clone()
     }

beacon_node/client/src/builder.rs

@@ -1,5 +1,5 @@
 use crate::compute_light_client_updates::{
-    compute_and_gossip_light_client_updates, LIGHT_CLIENT_SERVER_CHANNEL_CAPACITY,
+    compute_light_client_updates, LIGHT_CLIENT_SERVER_CHANNEL_CAPACITY,
 };
 use crate::config::{ClientGenesis, Config as ClientConfig};
 use crate::notifier::spawn_notifier;
@@ -957,11 +957,10 @@ where
                 let log = light_client_update_context.log().clone();
                 light_client_update_context.executor.spawn(
                     async move {
-                        compute_and_gossip_light_client_updates(
+                        compute_light_client_updates(
                             &inner_chain,
                             light_client_server_rv,
                             beacon_processor_channels.work_reprocessing_tx,
-                            self.network_senders.clone(),
                             &log,
                         )
                         .await

beacon_node/client/src/compute_light_client_updates.rs

@@ -2,8 +2,6 @@ use beacon_chain::{BeaconChain, BeaconChainTypes, LightClientProducerEvent};
 use beacon_processor::work_reprocessing_queue::ReprocessQueueMessage;
 use futures::channel::mpsc::Receiver;
 use futures::StreamExt;
-use lighthouse_network::PubsubMessage;
-use network::{NetworkMessage, NetworkSenders};
 use slog::{error, Logger};
 use tokio::sync::mpsc::Sender;
 
@@ -13,11 +11,10 @@ use tokio::sync::mpsc::Sender;
 // take a few milliseconds. 32 is a small enough arbitrary number.
 pub(crate) const LIGHT_CLIENT_SERVER_CHANNEL_CAPACITY: usize = 32;
 
-pub async fn compute_and_gossip_light_client_updates<T: BeaconChainTypes>(
+pub async fn compute_light_client_updates<T: BeaconChainTypes>(
     chain: &BeaconChain<T>,
     mut light_client_server_rv: Receiver<LightClientProducerEvent<T::EthSpec>>,
     reprocess_tx: Sender<ReprocessQueueMessage>,
-    network_senders: Option<NetworkSenders<T::EthSpec>>,
     log: &Logger,
 ) {
     // Should only receive events for recent blocks, import_block filters by blocks close to clock.
@@ -28,53 +25,12 @@ pub async fn compute_and_gossip_light_client_updates<T: BeaconChainTypes>(
     while let Some(event) = light_client_server_rv.next().await {
         let parent_root = event.0;
 
-        match chain.recompute_and_cache_light_client_updates(event) {
-            Err(e) => error!(log, "error computing light_client updates {:?}", e),
-            Ok(()) => {
-                // gossip the latest computed light client updates
-                if let (
-                    Some(network_tx),
-                    Some(latest_finality_update),
-                    Some(latest_optimistic_update),
-                ) = (
-                    network_senders.as_ref().map(|n| n.network_send()),
-                    chain.light_client_server_cache.get_latest_finality_update(),
-                    chain
-                        .light_client_server_cache
-                        .get_latest_optimistic_update(),
-                ) {
-                    let _ = network_tx
-                        .send(NetworkMessage::Publish {
-                            messages: vec![PubsubMessage::LightClientFinalityUpdate(Box::new(
-                                latest_finality_update,
-                            ))],
-                        })
-                        .inspect_err(|e| {
-                            error!(log, "error publishing light_client finality update {:?}", e)
-                        });
-                    let _ = network_tx
-                        .send(NetworkMessage::Publish {
-                            messages: vec![PubsubMessage::LightClientOptimisticUpdate(Box::new(
-                                latest_optimistic_update,
-                            ))],
-                        })
-                        .inspect_err(|e| {
-                            error!(
-                                log,
-                                "error publishing light_client optimistic update {:?}", e
-                            )
-                        });
-                } else {
-                    error!(
-                        log,
-                        "couldn't publish light_client updates";
-                        "network_senders_avail" => network_senders.is_some(),
-                        "finality_update_avail" => chain.light_client_server_cache.get_latest_finality_update().is_some(),
-                        "optimistic_update_avail" => chain.light_client_server_cache.get_latest_optimistic_update().is_some(),
-                    )
-                }
-            }
-        };
+        chain
+            .recompute_and_cache_light_client_updates(event)
+            .unwrap_or_else(|e| {
+                error!(log, "error computing light_client updates {:?}", e);
+            });
+
         let msg = ReprocessQueueMessage::NewLightClientOptimisticUpdate { parent_root };
         if reprocess_tx.try_send(msg).is_err() {
             error!(log, "Failed to inform light client update"; "parent_root" => %parent_root)

beacon_node/http_api/src/sync_committees.rs

@@ -319,6 +319,40 @@ pub fn process_signed_contribution_and_proofs<T: BeaconChainTypes>(
 
     let seen_timestamp = timestamp_now();
 
+    if let Some(latest_optimistic_update) = chain
+        .light_client_server_cache
+        .should_broadcast_latest_optimistic_update()
+    {
+        let _ = publish_pubsub_message(
+            &network_tx,
+            PubsubMessage::LightClientOptimisticUpdate(Box::new(latest_optimistic_update)),
+        )
+        .inspect_err(|e| {
+            error!(
+                log,
+                "Unable to broadcast latest light client optimistic update";
+                "error" => ?e,
+            );
+        });
+    };
+
+    if let Some(latest_finality_update) = chain
+        .light_client_server_cache
+        .should_broadcast_latest_finality_update()
+    {
+        let _ = publish_pubsub_message(
+            &network_tx,
+            PubsubMessage::LightClientFinalityUpdate(Box::new(latest_finality_update)),
+        )
+        .inspect_err(|e| {
+            error!(
+                log,
+                "Unable to broadcast latest light client finality update";
+                "error" => ?e,
+            );
+        });
+    };
+
     // Verify contributions & broadcast to the network.
     for (index, contribution) in signed_contribution_and_proofs.into_iter().enumerate() {
         let aggregator_index = contribution.message.aggregator_index;