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

beacon_node/beacon_processor/src/work_reprocessing_queue.rs

@@ -82,12 +82,15 @@ pub enum ReprocessQueueMessage {
     /// A gossip block for hash `X` is being imported, we should queue the rpc block for the same
     /// hash until the gossip block is imported.
     RpcBlock(QueuedRpcBlock),
-    /// A block that was successfully processed. We use this to handle attestations and light client updates
+    /// A block that was successfully processed. We use this to handle attestations updates
     /// for unknown blocks.
     BlockImported {
         block_root: Hash256,
         parent_root: Hash256,
     },
+    /// A new `LightClientOptimisticUpdate` has been produced. We use this to handle light client
+    /// updates for unknown parent blocks.
+    NewLightClientOptimisticUpdate { parent_root: Hash256 },
     /// An unaggregated attestation that references an unknown block.
     UnknownBlockUnaggregate(QueuedUnaggregate),
     /// An aggregated attestation that references an unknown block.
@@ -688,6 +691,8 @@ impl<S: SlotClock> ReprocessQueue<S> {
                         );
                     }
                 }
+            }
+            InboundEvent::Msg(NewLightClientOptimisticUpdate { parent_root }) => {
                 // Unqueue the light client optimistic updates we have for this root, if any.
                 if let Some(queued_lc_id) = self
                     .awaiting_lc_updates_per_parent_root

beacon_node/client/src/builder.rs

@@ -768,7 +768,7 @@ where
                 }
                 .spawn_manager(
                     beacon_processor_channels.beacon_processor_rx,
-                    beacon_processor_channels.work_reprocessing_tx,
+                    beacon_processor_channels.work_reprocessing_tx.clone(),
                     beacon_processor_channels.work_reprocessing_rx,
                     None,
                     beacon_chain.slot_clock.clone(),
@@ -857,8 +857,13 @@ where
                     let log = broadcast_context.log().clone();
                     broadcast_context.executor.spawn(
                         async move {
-                            compute_light_client_updates(&inner_chain, light_client_server_rv, &log)
-                                .await
+                            compute_light_client_updates(
+                                &inner_chain,
+                                light_client_server_rv,
+                                beacon_processor_channels.work_reprocessing_tx,
+                                &log,
+                            )
+                            .await
                         },
                         "lcserv_broadcast",
                     );

beacon_node/client/src/compute_light_client_updates.rs

@@ -1,7 +1,9 @@
 use beacon_chain::{BeaconChain, BeaconChainTypes, LightClientProducerEvent};
+use beacon_processor::work_reprocessing_queue::ReprocessQueueMessage;
 use futures::channel::mpsc::Receiver;
 use futures::StreamExt;
 use slog::{error, Logger};
+use tokio::sync::mpsc::Sender;
 
 // Each `LightClientProducerEvent` is ~200 bytes. With the light_client server producing only recent
 // updates it is okay to drop some events in case of overloading. In normal network conditions
@@ -12,6 +14,7 @@ pub(crate) const LIGHT_CLIENT_SERVER_CHANNEL_CAPACITY: usize = 32;
 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>,
     log: &Logger,
 ) {
     // Should only receive events for recent blocks, import_block filters by blocks close to clock.
@@ -20,10 +23,17 @@ pub async fn compute_light_client_updates<T: BeaconChainTypes>(
     // Uses a bounded receiver, so may drop some SyncAggregates if very overloaded. This is okay
     // since only the most recent updates have value.
     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);
-            })
+            });
+
+        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)
+        };
     }
 }