fix: ensure that downloader sync gap is only set once (paradigmxyz#16693)

Author: klkvr

crates/ethereum/node/tests/e2e/p2p.rs

@@ -187,7 +187,7 @@ async fn test_reorg_through_backfill() -> eyre::Result<()> {
 
     // Produce an unfinalized fork chain with 5 blocks
     second_node.payload.timestamp = head.header.timestamp;
-    advance_with_random_transactions(&mut second_node, 5, &mut rng, false).await?;
+    advance_with_random_transactions(&mut second_node, 10, &mut rng, false).await?;
 
     // Now reorg second node to the finalized canonical head
     let head = first_provider.get_block_by_number(100.into()).await?.unwrap();

crates/net/p2p/src/headers/downloader.rs

@@ -80,8 +80,8 @@ impl SyncTarget {
 }
 
 /// Represents a gap to sync: from `local_head` to `target`
-#[derive(Clone, Debug)]
-pub struct HeaderSyncGap<H = Header> {
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct HeaderSyncGap<H: Sealable = Header> {
     /// The local head block. Represents lower bound of sync range.
     pub local_head: SealedHeader<H>,
 

crates/stages/stages/src/stages/headers.rs

@@ -214,7 +214,6 @@ where
         let target = SyncTarget::Tip(*self.tip.borrow());
         let gap = HeaderSyncGap { local_head, target };
         let tip = gap.target.tip();
-        self.sync_gap = Some(gap.clone());
 
         // Nothing to sync
         if gap.is_closed() {
@@ -232,7 +231,10 @@ where
         let local_head_number = gap.local_head.number();
 
         // let the downloader know what to sync
-        self.downloader.update_sync_gap(gap.local_head, gap.target);
+        if self.sync_gap != Some(gap.clone()) {
+            self.sync_gap = Some(gap.clone());
+            self.downloader.update_sync_gap(gap.local_head, gap.target);
+        }
 
         // We only want to stop once we have all the headers on ETL filespace (disk).
         loop {
@@ -263,13 +265,17 @@ where
                 }
                 Some(Err(HeadersDownloaderError::DetachedHead { local_head, header, error })) => {
                     error!(target: "sync::stages::headers", %error, "Cannot attach header to head");
+                    self.sync_gap = None;
                     return Poll::Ready(Err(StageError::DetachedHead {
                         local_head: Box::new(local_head.block_with_parent()),
                         header: Box::new(header.block_with_parent()),
                         error,
                     }))
                 }
-                None => return Poll::Ready(Err(StageError::ChannelClosed)),
+                None => {
+                    self.sync_gap = None;
+                    return Poll::Ready(Err(StageError::ChannelClosed))
+                }
             }
         }
     }
@@ -279,7 +285,7 @@ where
     fn execute(&mut self, provider: &Provider, input: ExecInput) -> Result<ExecOutput, StageError> {
         let current_checkpoint = input.checkpoint();
 
-        if self.sync_gap.as_ref().ok_or(StageError::MissingSyncGap)?.is_closed() {
+        if self.sync_gap.take().ok_or(StageError::MissingSyncGap)?.is_closed() {
             self.is_etl_ready = false;
             return Ok(ExecOutput::done(current_checkpoint))
         }