add logic to gossip light client updates post computation

eserilev · eserilev · commit 29b5614bcba7 · 2025-02-21T07:07:31.000-08:00
diff --git a/beacon_node/client/src/builder.rs b/beacon_node/client/src/builder.rs
@@ -1,5 +1,5 @@
 use crate::compute_light_client_updates::{
-    compute_light_client_updates, LIGHT_CLIENT_SERVER_CHANNEL_CAPACITY,
+    compute_and_gossip_light_client_updates, LIGHT_CLIENT_SERVER_CHANNEL_CAPACITY,
 };
 use crate::config::{ClientGenesis, Config as ClientConfig};
 use crate::notifier::spawn_notifier;
@@ -957,10 +957,11 @@ where
                 let log = light_client_update_context.log().clone();
                 light_client_update_context.executor.spawn(
                     async move {
-                        compute_light_client_updates(
+                        compute_and_gossip_light_client_updates(
                             &inner_chain,
                             light_client_server_rv,
                             beacon_processor_channels.work_reprocessing_tx,
+                            self.network_senders.clone(),
                             &log,
                         )
                         .await
diff --git a/beacon_node/client/src/compute_light_client_updates.rs b/beacon_node/client/src/compute_light_client_updates.rs
@@ -2,6 +2,8 @@ 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;
 
@@ -11,10 +13,11 @@ 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_light_client_updates<T: BeaconChainTypes>(
+pub async fn compute_and_gossip_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.
@@ -25,12 +28,53 @@ pub async fn compute_light_client_updates<T: BeaconChainTypes>(
     while let Some(event) = light_client_server_rv.next().await {
         let parent_root = event.0;
 
-        chain
-            .recompute_and_cache_light_client_updates(event)
-            .unwrap_or_else(|e| {
-                error!(log, "error computing light_client updates {:?}", e);
-            });
-
+        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(),
+                    )
+                }
+            }
+        };
         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)
