Debounce UnknownBlockHashFromAttestation events (#5706)

Squashed commit of the following:

commit a050d13
Author: dapplion <[email protected]>
Date:   Tue May 7 11:13:05 2024 +0900

    Re-add dropped comment

commit 729005f
Merge: bc5899a 5135a77
Author: realbigsean <[email protected]>
Date:   Mon May 6 15:39:13 2024 -0400

    Merge branch 'unstable' into debounce-sync-block-unknown

commit bc5899a
Author: dapplion <[email protected]>
Date:   Fri May 3 17:13:20 2024 +0900

    Debounce UnknownBlockHashFromAttestation events

beacon_node/network/src/sync/manager.rs

@@ -56,6 +56,7 @@ use lighthouse_network::rpc::RPCError;
 use lighthouse_network::types::{NetworkGlobals, SyncState};
 use lighthouse_network::SyncInfo;
 use lighthouse_network::{PeerAction, PeerId};
+use lru_cache::LRUTimeCache;
 use slog::{crit, debug, error, info, o, trace, warn, Logger};
 use std::ops::Sub;
 use std::sync::Arc;
@@ -72,6 +73,11 @@ use types::{BlobSidecar, EthSpec, Hash256, SignedBeaconBlock, Slot};
 /// blocks for.
 pub const SLOT_IMPORT_TOLERANCE: usize = 32;
 
+/// Suppress duplicated `UnknownBlockHashFromAttestation` events for some duration of time. In
+/// practice peers are likely to send the same root during a single slot. 30 seconds is a rather
+/// arbitrary number that covers a full slot, but allows recovery if sync get stuck for a few slots.
+const NOTIFIED_UNKNOWN_ROOT_EXPIRY_SECONDS: u64 = 30;
+
 pub type Id = u32;
 
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
@@ -199,6 +205,10 @@ pub struct SyncManager<T: BeaconChainTypes> {
     backfill_sync: BackFillSync<T>,
 
     block_lookups: BlockLookups<T>,
+    /// debounce duplicated `UnknownBlockHashFromAttestation` for the same root peer tuple. A peer
+    /// may forward us thousands of a attestations, each one triggering an individual event. Only
+    /// one event is useful, the rest generating log noise and wasted cycles
+    notified_unknown_roots: LRUTimeCache<(PeerId, Hash256)>,
 
     /// The logger for the import manager.
     log: Logger,
@@ -262,6 +272,9 @@ impl<T: BeaconChainTypes> SyncManager<T> {
                 log.new(o!("service" => "backfill_sync")),
             ),
             block_lookups: BlockLookups::new(log.new(o!("service"=> "lookup_sync"))),
+            notified_unknown_roots: LRUTimeCache::new(Duration::from_secs(
+                NOTIFIED_UNKNOWN_ROOT_EXPIRY_SECONDS,
+            )),
             log: log.clone(),
         }
     }
@@ -622,7 +635,11 @@ impl<T: BeaconChainTypes> SyncManager<T> {
                 );
             }
             SyncMessage::UnknownBlockHashFromAttestation(peer_id, block_root) => {
-                self.handle_unknown_block_root(peer_id, block_root);
+                if !self.notified_unknown_roots.contains(&(peer_id, block_root)) {
+                    self.notified_unknown_roots.insert((peer_id, block_root));
+                    debug!(self.log, "Received unknown block hash message"; "block_root" => ?block_root, "peer" => ?peer_id);
+                    self.handle_unknown_block_root(peer_id, block_root);
+                }
             }
             SyncMessage::Disconnect(peer_id) => {
                 debug!(self.log, "Received disconnected message"; "peer_id" => %peer_id);