op-rs · nonsense · Aug 6, 2026 · Aug 6, 2026 · Aug 6, 2026 · einar-oplabs
diff --git a/crates/cli/commands/src/stage/run.rs b/crates/cli/commands/src/stage/run.rs
@@ -196,13 +196,16 @@ impl<C: ChainSpecParser<ChainSpec: EthChainSpec + Hardforks + EthereumHardforks>
                     };
                     let (_, rx) = watch::channel(tip.hash_slow());
                     (
-                        Box::new(HeaderStage::new(
-                            provider_factory.clone(),
-                            ReverseHeadersDownloaderBuilder::new(config.stages.headers)
-                                .build(fetch_client, consensus.clone()),
-                            rx,
-                            etl_config,
-                        )),
+                        Box::new(
+                            HeaderStage::new(
+                                provider_factory.clone(),
+                                ReverseHeadersDownloaderBuilder::new(config.stages.headers)
+                                    .build(fetch_client, consensus.clone()),
+                                rx,
+                                etl_config,
+                            )
+                            .with_stall_timeout(config.stages.headers.stall_timeout),
+                        ),
                         None,
                     )
                 }

diff --git a/crates/config/src/config.rs b/crates/config/src/config.rs
@@ -192,6 +192,18 @@ pub struct HeadersConfig {
     pub downloader_request_limit: u64,
     /// The maximum number of headers to download before committing progress to the database.
     pub commit_threshold: u64,
+    /// The time after which the header download is considered stalled if the downloader made
+    /// no progress while the headers stage is waiting on it.
+    ///
+    /// Default: 30 seconds
+    #[cfg_attr(
+        feature = "serde",
+        serde(
+            serialize_with = "humantime_serde::serialize",
+            deserialize_with = "humantime_serde::deserialize"
+        )
+    )]
+    pub stall_timeout: Duration,
 }
 
 impl Default for HeadersConfig {
@@ -202,6 +214,7 @@ impl Default for HeadersConfig {
             downloader_max_concurrent_requests: 100,
             downloader_min_concurrent_requests: 5,
             downloader_max_buffered_responses: 100,
+            stall_timeout: Duration::from_secs(30),
         }
     }
 }

diff --git a/crates/net/downloaders/src/bodies/request.rs b/crates/net/downloaders/src/bodies/request.rs
@@ -35,7 +35,8 @@ use std::{
 /// does](https://github.com/ethereum/go-ethereum/blob/f53ff0ff4a68ffc56004ab1d5cc244bcb64d3277/les/server_requests.go#L245).
 /// All errors regarding the response cause the peer to get penalized, meaning that adversaries
 /// that try to give us bodies that do not match the requested order are going to be penalized
-/// and eventually disconnected.
+/// and eventually disconnected. The exception is an empty response, which is the
+/// protocol-correct answer of a peer that does not have the requested bodies.
 pub(crate) struct BodiesRequestFuture<B: Block, C: BodiesClient<Body = B::Body>> {
     client: Arc<C>,
     consensus: Arc<dyn Consensus<B>>,
@@ -89,7 +90,12 @@ where
     fn on_error(&mut self, error: DownloadError, peer_id: Option<PeerId>) {
         self.metrics.increment_errors(&error);
         tracing::debug!(target: "downloaders::bodies", ?peer_id, %error, "Error requesting bodies");
-        if let Some(peer_id) = peer_id {
+        // An empty response is the protocol-correct answer of a peer that does not have the
+        // requested bodies and must not accrue ban-worthy reputation; the request is simply
+        // resubmitted.
+        if let Some(peer_id) = peer_id &&
+            !matches!(error, DownloadError::EmptyResponse)
+        {
             self.client.report_bad_message(peer_id);
         }
         self.submit_request(
@@ -308,4 +314,29 @@ mod tests {
             (headers.into_iter().filter(|h| !h.is_empty()).count() as u64).div_ceil(2)
         );
     }
+
+    /// Empty responses are the protocol-correct answer of peers that do not have the requested
+    /// bodies and must not be reported as bad messages while the request is retried.
+    #[tokio::test]
+    async fn empty_responses_are_not_penalized() {
+        // Generate some random blocks
+        let (headers, mut bodies) = generate_bodies(0..=19);
+
+        let client = Arc::new(
+            TestBodiesClient::default()
+                .with_bodies(bodies.clone())
+                .with_max_batch_size(5)
+                .with_empty_responses(2),
+        );
+        let fut = BodiesRequestFuture::<Block, _>::new(
+            client.clone(),
+            Arc::new(TestConsensus::default()),
+            BodyDownloaderMetrics::default(),
+        )
+        .with_headers(headers.clone());
+
+        assert_eq!(fut.await.unwrap(), zip_blocks(headers.iter(), &mut bodies));
+        assert!(client.times_requested() > 1);
+        assert_eq!(client.bad_messages(), 0);
+    }
 }
diff --git a/crates/net/downloaders/src/headers/reverse_headers.rs b/crates/net/downloaders/src/headers/reverse_headers.rs
@@ -539,7 +539,17 @@ where
     fn penalize_peer(&self, peer_id: Option<PeerId>, error: &DownloadError) {
         // Penalize the peer for bad response
         if let Some(peer_id) = peer_id {
-            trace!(target: "downloaders::headers", ?peer_id, %error, "Penalizing peer");
+            if matches!(error, DownloadError::EmptyResponse) {
+                // An empty response is the protocol-correct answer of a peer that does not have
+                // the requested range and must not accrue ban-worthy reputation: if no peer can
+                // serve the range yet (e.g. all peers are equally out of sync), banning honest
+                // peers guarantees the download can never complete. The fetcher already
+                // deprioritizes peers whose last response was unsatisfactory when picking a peer
+                // for the resubmitted request.
+                debug!(target: "downloaders::headers", ?peer_id, %error, "Peer unable to serve requested headers");
+                return
+            }
+            debug!(target: "downloaders::headers", ?peer_id, %error, "Penalizing peer");
             self.client.report_bad_message(peer_id);
         }
     }
@@ -1251,7 +1261,15 @@ mod tests {
     use alloy_eips::{eip1898::BlockWithParent, BlockNumHash};
     use assert_matches::assert_matches;
     use reth_consensus::test_utils::TestConsensus;
-    use reth_network_p2p::test_utils::TestHeadersClient;
+    use reth_network_p2p::{download::DownloadClient, test_utils::TestHeadersClient};
+    use reth_network_peers::WithPeerId;
+    use std::{
+        collections::VecDeque,
+        sync::{
+            atomic::{AtomicU64, Ordering as AtomicOrdering},
+            Mutex,
+        },
+    };
 
     /// Tests that `replace_number` works the same way as `Option::replace`
     #[test]
@@ -1547,4 +1565,144 @@ mod tests {
 
         assert!(downloader.next().await.is_none());
     }
+
+    /// A client that serves a scripted sequence of responses, pends once the script is
+    /// exhausted, and counts how often it was reported for a bad message.
+    #[derive(Clone, Debug)]
+    struct ScriptedHeadersClient {
+        responses: Arc<Mutex<VecDeque<Vec<Header>>>>,
+        total_requests: Arc<AtomicU64>,
+        bad_messages: Arc<AtomicU64>,
+    }
+
+    impl ScriptedHeadersClient {
+        fn new(responses: Vec<Vec<Header>>) -> Self {
+            Self {
+                responses: Arc::new(Mutex::new(responses.into())),
+                total_requests: Arc::new(AtomicU64::new(0)),
+                bad_messages: Arc::new(AtomicU64::new(0)),
+            }
+        }
+
+        fn total_requests(&self) -> u64 {
+            self.total_requests.load(AtomicOrdering::Relaxed)
+        }
+
+        fn bad_messages(&self) -> u64 {
+            self.bad_messages.load(AtomicOrdering::Relaxed)
+        }
+    }
+
+    impl DownloadClient for ScriptedHeadersClient {
+        fn report_bad_message(&self, _peer_id: PeerId) {
+            self.bad_messages.fetch_add(1, AtomicOrdering::Relaxed);
+        }
+
+        fn num_connected_peers(&self) -> usize {
+            1
+        }
+    }
+
+    impl HeadersClient for ScriptedHeadersClient {
+        type Header = Header;
+        type Output = Pin<Box<dyn Future<Output = PeerRequestResult<Vec<Header>>> + Send + Sync>>;
+
+        fn get_headers_with_priority(
+            &self,
+            _request: HeadersRequest,
+            _priority: Priority,
+        ) -> Self::Output {
+            self.total_requests.fetch_add(1, AtomicOrdering::Relaxed);
+            let next = self.responses.lock().unwrap().pop_front();
+            Box::pin(async move {
+                match next {
+                    Some(headers) => Ok(WithPeerId::from((PeerId::default(), headers))),
+                    None => std::future::pending().await,
+                }
+            })
+        }
+    }
+
+    /// Polls the downloader once without awaiting its next item.
+    async fn poll_downloader_once<S: Stream + Unpin>(downloader: &mut S) {
+        std::future::poll_fn(|cx| {
+            let _ = Pin::new(&mut *downloader).poll_next(cx);
+            Poll::Ready(())
+        })
+        .await
+    }
+
+    /// Empty responses are the protocol-correct answer of peers that do not have the requested
+    /// range and must not be reported as bad messages, which would otherwise ban the entire
+    /// honest peer set whenever no peer can serve the range yet.
+    #[tokio::test]
+    async fn empty_responses_are_not_penalized() {
+        reth_tracing::init_test_tracing();
+
+        const EMPTY_RESPONSES: usize = 10;
+
+        let p3 = SealedHeader::default();
+        let p2 = child_header(&p3);
+        let p1 = child_header(&p2);
+        let p0 = child_header(&p1);
+
+        // a valid sync target response followed by only empty responses for the range request
+        let mut responses = vec![vec![p0.as_ref().clone()]];
+        responses.extend(std::iter::repeat_n(Vec::new(), EMPTY_RESPONSES));
+        let client = Arc::new(ScriptedHeadersClient::new(responses));
+
+        let mut downloader = ReverseHeadersDownloaderBuilder::default()
+            .stream_batch_size(3)
+            .request_limit(3)
+            .build(Arc::clone(&client), Arc::new(TestConsensus::default()));
+        downloader.update_local_head(p3);
+        downloader.update_sync_target(SyncTarget::Tip(p0.hash()));
+
+        // drive the downloader until every scripted empty response has been consumed and the
+        // failed request resubmitted
+        let expected_requests = 2 + EMPTY_RESPONSES as u64;
+        for _ in 0..100 {
+            if client.total_requests() >= expected_requests {
+                break
+            }
+            poll_downloader_once(&mut downloader).await;
+        }
+
+        assert_eq!(client.total_requests(), expected_requests);
+        assert_eq!(client.bad_messages(), 0);
+    }
+
+    /// A malformed response (wrong start block) is still reported as a bad message.
+    #[tokio::test]
+    async fn malformed_responses_are_penalized() {
+        reth_tracing::init_test_tracing();
+
+        let p3 = SealedHeader::default();
+        let p2 = child_header(&p3);
+        let p1 = child_header(&p2);
+        let p0 = child_header(&p1);
+
+        // a valid sync target response, then a response that starts at the wrong block: the
+        // range request asks for blocks 2 and 1 but the response starts at block 1
+        let client = Arc::new(ScriptedHeadersClient::new(vec![
+            vec![p0.as_ref().clone()],
+            vec![p2.as_ref().clone(), p3.as_ref().clone()],
+        ]));
+
+        let mut downloader = ReverseHeadersDownloaderBuilder::default()
+            .stream_batch_size(3)
+            .request_limit(3)
+            .build(Arc::clone(&client), Arc::new(TestConsensus::default()));
+        downloader.update_local_head(p3);
+        downloader.update_sync_target(SyncTarget::Tip(p0.hash()));
+
+        for _ in 0..100 {
+            if client.bad_messages() > 0 {
+                break
+            }
+            poll_downloader_once(&mut downloader).await;
+        }
+
+        assert_eq!(client.bad_messages(), 1);
+    }
 }
diff --git a/crates/net/downloaders/src/test_utils/bodies_client.rs b/crates/net/downloaders/src/test_utils/bodies_client.rs
@@ -25,6 +25,7 @@ pub struct TestBodiesClient {
     max_batch_size: Option<usize>,
     times_requested: AtomicU64,
     empty_response_mod: Option<u64>,
+    bad_messages: AtomicU64,
 }
 
 impl TestBodiesClient {
@@ -54,6 +55,11 @@ impl TestBodiesClient {
         self.times_requested.load(Ordering::Relaxed)
     }
 
+    /// Returns how often the client was reported for a bad message.
+    pub(crate) fn bad_messages(&self) -> u64 {
+        self.bad_messages.load(Ordering::Relaxed)
+    }
+
     /// Returns whether or not the client should respond with an empty response.
     ///
     /// This will only return true if `empty_response_mod` is `Some`, and `times_requested %
@@ -69,7 +75,7 @@ impl TestBodiesClient {
 
 impl DownloadClient for TestBodiesClient {
     fn report_bad_message(&self, _peer_id: PeerId) {
-        // noop
+        self.bad_messages.fetch_add(1, Ordering::Relaxed);
     }
 
     fn num_connected_peers(&self) -> usize {

diff --git a/crates/net/network/src/manager.rs b/crates/net/network/src/manager.rs
@@ -50,6 +50,7 @@ use reth_network_api::{
     test_utils::PeersHandle,
     EthProtocolInfo, NetworkEvent, NetworkStatus, PeerInfo, PeerRequest,
 };
+use reth_network_p2p::sync::SyncStateProvider;
 use reth_network_peers::{NodeRecord, PeerId};
 use reth_network_types::ReputationChangeKind;
 use reth_storage_api::BlockNumReader;
@@ -906,6 +907,12 @@ impl<N: NetworkPrimitives> NetworkManager<N> {
                     "Session disconnected"
                 );
 
+                // While syncing, the node is downloading data from its peers: without any
+                // connected peer the sync cannot make progress.
+                if total_active == 0 && self.handle.is_syncing() {
+                    warn!(target: "net", ?peer_id, ?error, "All peers disconnected while the node is syncing");
+                }
+
                 // Capture direction before state is reset to Idle
                 let is_inbound = self.swarm.peers().is_inbound_peer(&peer_id);
 

diff --git a/crates/node/builder/src/launch/engine.rs b/crates/node/builder/src/launch/engine.rs
@@ -346,9 +346,13 @@ impl EngineNodeLauncher {
                                     debug!(target: "reth::cli", "Terminating after initial backfill");
                                     break
                                 }
-                                if startup_sync_state_idle {
-                                    network_handle.update_sync_state(SyncState::Idle);
-                                }
+                                // If the node is still behind, the engine re-evaluates against
+                                // the latest fork choice state and re-triggers backfill or live
+                                // sync, marking the network as syncing again. Until then the
+                                // network must return to idle: staying "syncing" after an
+                                // unproductive backfill suppresses transaction gossip
+                                // indefinitely.
+                                network_handle.update_sync_state(SyncState::Idle);
                             }
                             ChainEvent::BackfillSyncStarted => {
                                 network_handle.update_sync_state(SyncState::Syncing);

diff --git a/crates/node/core/src/args/debug.rs b/crates/node/core/src/args/debug.rs
@@ -118,10 +118,10 @@ pub struct DebugArgs {
     #[arg(long = "ethstats", help_heading = "Debug")]
     pub ethstats: Option<String>,
 
-    /// Set the node to idle state when the backfill is not running.
+    /// Set the node to idle state on startup when no backfill is required.
     ///
-    /// This makes the `eth_syncing` RPC return "Idle" when the node has just started or finished
-    /// the backfill, but did not yet receive any new blocks.
+    /// This makes the `eth_syncing` RPC return "Idle" when the node has just started and does
+    /// not need to run a backfill, but did not yet receive any new blocks.
     #[arg(long = "debug.startup-sync-state-idle", help_heading = "Debug")]
     pub startup_sync_state_idle: bool,
 }

diff --git a/crates/stages/api/src/error.rs b/crates/stages/api/src/error.rs
@@ -82,6 +82,14 @@ pub enum StageError {
     /// Download channel closed
     #[error("download channel closed")]
     ChannelClosed,
+    /// The stage made no progress towards becoming ready for execution, e.g. because no peer
+    /// can currently serve the data it is waiting on.
+    ///
+    /// This error is recoverable: the pipeline retries the stage, and ends the current
+    /// pipeline run with no progress once the stage keeps stalling beyond the pipeline's
+    /// readiness timeout, returning control to the caller for re-targeting.
+    #[error("stage stalled while waiting to become ready: {0}")]
+    Stalled(String),
     /// The stage encountered a database integrity error.
     #[error("database integrity error occurred: {0}")]
     DatabaseIntegrity(#[from] ProviderError),