From b41ed0759964990bec486e223cc3a25ae2c93377 Mon Sep 17 00:00:00 2001
From: dapplion <35266934+dapplion@users.noreply.github.com>
Date: Wed, 13 Dec 2023 11:01:55 +0100
Subject: [PATCH] revert dropping reprocessing queue

---
 ...t_client_optimistic_update_verification.rs | 15 ++++++
 .../gossip_methods.rs                         | 54 ++++++++++++++++++-
 .../src/network_beacon_processor/mod.rs       |  2 +
 3 files changed, 70 insertions(+), 1 deletion(-)

diff --git a/beacon_node/beacon_chain/src/light_client_optimistic_update_verification.rs b/beacon_node/beacon_chain/src/light_client_optimistic_update_verification.rs
index 6c48cb8ceff..813b112db5a 100644
--- a/beacon_node/beacon_chain/src/light_client_optimistic_update_verification.rs
+++ b/beacon_node/beacon_chain/src/light_client_optimistic_update_verification.rs
@@ -28,6 +28,8 @@ pub enum Error {
     SigSlotStartIsNone,
     /// Failed to construct a LightClientOptimisticUpdate from state.
     FailedConstructingUpdate,
+    /// Unknown block with parent root.
+    UnknownBlockParentRoot(Hash256),
 }
 
 /// Wraps a `LightClientOptimisticUpdate` that has been verified for propagation on the gossip network.
@@ -59,6 +61,19 @@ impl<T: BeaconChainTypes> VerifiedLightClientOptimisticUpdate<T> {
             return Err(Error::TooEarly);
         }
 
+        let head = chain.canonical_head.cached_head();
+        let head_block = &head.snapshot.beacon_block;
+        // check if we can process the optimistic update immediately
+        // otherwise queue
+        let canonical_root = rcv_optimistic_update
+            .attested_header
+            .beacon
+            .canonical_root();
+
+        if canonical_root != head_block.message().parent_root() {
+            return Err(Error::UnknownBlockParentRoot(canonical_root));
+        }
+
         let latest_optimistic_update = chain
             .light_client_server_cache
             .get_latest_optimistic_update()
diff --git a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs
index 42047e39df9..07fc06bc370 100644
--- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs
+++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs
@@ -40,7 +40,8 @@ use types::{
 
 use beacon_processor::{
     work_reprocessing_queue::{
-        QueuedAggregate, QueuedGossipBlock, QueuedUnaggregate, ReprocessQueueMessage,
+        QueuedAggregate, QueuedGossipBlock, QueuedLightClientUpdate, QueuedUnaggregate,
+        ReprocessQueueMessage,
     },
     DuplicateCache, GossipAggregatePackage, GossipAttestationPackage,
 };
@@ -1692,6 +1693,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         message_id: MessageId,
         peer_id: PeerId,
         light_client_optimistic_update: LightClientOptimisticUpdate<T::EthSpec>,
+        reprocess_tx: Option<mpsc::Sender<ReprocessQueueMessage>>,
         seen_timestamp: Duration,
     ) {
         match self.chain.verify_optimistic_update_for_gossip(
@@ -1710,6 +1712,56 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
             }
             Err(e) => {
                 match e {
+                    LightClientOptimisticUpdateError::UnknownBlockParentRoot(parent_root) => {
+                        metrics::inc_counter(
+                            &metrics::BEACON_PROCESSOR_REPROCESSING_QUEUE_SENT_OPTIMISTIC_UPDATES,
+                        );
+                        debug!(
+                            self.log,
+                            "Optimistic update for unknown block";
+                            "peer_id" => %peer_id,
+                            "parent_root" => ?parent_root
+                        );
+
+                        if let Some(sender) = reprocess_tx {
+                            let processor = self.clone();
+                            let msg = ReprocessQueueMessage::UnknownLightClientOptimisticUpdate(
+                                QueuedLightClientUpdate {
+                                    parent_root,
+                                    process_fn: Box::new(move || {
+                                        processor.process_gossip_optimistic_update(
+                                            message_id,
+                                            peer_id,
+                                            light_client_optimistic_update,
+                                            None, // Do not reprocess this message again.
+                                            seen_timestamp,
+                                        )
+                                    }),
+                                },
+                            );
+
+                            if sender.try_send(msg).is_err() {
+                                error!(
+                                    self.log,
+                                    "Failed to send optimistic update for re-processing";
+                                )
+                            }
+                        } else {
+                            debug!(
+                                self.log,
+                                "Not sending light client update because it had been reprocessed";
+                                "peer_id" => %peer_id,
+                                "parent_root" => ?parent_root
+                            );
+
+                            self.propagate_validation_result(
+                                message_id,
+                                peer_id,
+                                MessageAcceptance::Ignore,
+                            );
+                        }
+                        return;
+                    }
                     LightClientOptimisticUpdateError::InvalidLightClientOptimisticUpdate => {
                         metrics::register_optimistic_update_error(&e);
 
diff --git a/beacon_node/network/src/network_beacon_processor/mod.rs b/beacon_node/network/src/network_beacon_processor/mod.rs
index 733a7bc9258..e7d3a7ce213 100644
--- a/beacon_node/network/src/network_beacon_processor/mod.rs
+++ b/beacon_node/network/src/network_beacon_processor/mod.rs
@@ -348,10 +348,12 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
     ) -> Result<(), Error<T::EthSpec>> {
         let processor = self.clone();
         let process_fn = move || {
+            let reprocess_tx = processor.reprocess_tx.clone();
             processor.process_gossip_optimistic_update(
                 message_id,
                 peer_id,
                 light_client_optimistic_update,
+                Some(reprocess_tx),
                 seen_timestamp,
             )
         };