From 162e20622b68f7e91b546c37951c3abd710ff174 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Thu, 6 Aug 2026 15:50:24 +0300 Subject: [PATCH 1/3] fix(net): do not penalize peers for empty download responses An empty response to a headers or bodies request is the protocol-correct answer of a peer that does not have the requested data, but the downloaders reported it as a bad message (ReputationChangeKind::BadMessage, -16384): four honest answers cross the ban threshold and disconnect the peer for 12 hours. When no peer can serve the requested range - e.g. a fork choice target ahead of every currently connected peer - the node bans its entire honest peer set within seconds, guaranteeing the download it is waiting on can never complete. This wedged OP Stack nodes in production (Ink Mainnet 2026-07-16, unichain 2026-08-05), where a >32 block gap after a sequencer stall sent all replicas into backfill with no peer able to serve the range. Empty responses are no longer reported; the failed request is resubmitted as before, and the fetcher already deprioritizes peers whose last response was unsatisfactory when picking a peer for the next request. Malformed responses (wrong start block, non-contiguous headers, invalid seals, too many bodies) keep the full penalty. The engine's single-block and block-range fetch paths already treat empty responses as non-reportable. Also promotes the "Penalizing peer" log from trace to debug so reputation destruction is visible closer to default verbosity. Co-Authored-By: Claude Fable 5 --- crates/net/downloaders/src/bodies/request.rs | 35 +++- .../src/headers/reverse_headers.rs | 162 +++++++++++++++++- .../src/test_utils/bodies_client.rs | 8 +- 3 files changed, 200 insertions(+), 5 deletions(-) diff --git a/crates/net/downloaders/src/bodies/request.rs b/crates/net/downloaders/src/bodies/request.rs index 864fa93d998..58251059848 100644 --- 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> { client: Arc, consensus: Arc>, @@ -89,7 +90,12 @@ where fn on_error(&mut self, error: DownloadError, peer_id: Option) { 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::::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 index 6a866f0ad9b..d90bad213de 100644 --- 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, 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>>>, + total_requests: Arc, + bad_messages: Arc, + } + + impl ScriptedHeadersClient { + fn new(responses: Vec>) -> 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>> + 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(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 index e0936d4cd01..fa863bb2d44 100644 --- 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, times_requested: AtomicU64, empty_response_mod: Option, + 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 { From 3f9120f5cd93b4a29213156b0177d5fd3c7389a1 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Thu, 6 Aug 2026 16:29:19 +0300 Subject: [PATCH 2/3] fix(stages): fail loud and recoverable when the header download stalls The Headers stage propagated the downloader's Pending forever: a fork choice target that no connected peer can serve - e.g. after a sequencer stall leaves every replica equally behind - parked the pipeline at stage 1 indefinitely, with no retry, no error and no log line. Combined with fork choice updates only being tracked during backfill, this wedged OP Stack nodes in production until an operator restart (Ink Mainnet 2026-07-16, unichain 2026-08-05). Two bounds, both progress-aware so a slow but healthy sync is never interrupted: - `HeaderStage` now arms a stall deadline (default 30s, configurable via `stages.headers.stall_timeout`) while the downloader is pending and resets it whenever headers arrive. On expiry the stage returns a new recoverable `StageError::Stalled` instead of pending forever. The ETL and downloader state are kept so a retry with an unchanged sync gap resumes the download; the ETL collectors are instead cleared whenever the sync gap changes, which also fixes stale collector entries leaking into a later run with a different target. A timeout wrapped around the `execute_ready` future itself would not work here: the headers stage legitimately downloads the entire gap - potentially for hours during initial sync - before it resolves. - The pipeline now gives a stage a readiness budget (default 5m, `PipelineBuilder::with_execute_ready_timeout`) armed on the first `execute_ready` failure and cleared once the stage becomes ready. When a stage keeps failing readiness beyond the budget, the pipeline emits `PipelineEvent::TimedOut` and gives up on the stage for the current run with `ControlFlow::NoProgress`, so control returns to the caller (the engine re-evaluates the backfill target) instead of retrying a stale target forever. Co-Authored-By: Claude Fable 5 --- crates/cli/commands/src/stage/run.rs | 17 +- crates/config/src/config.rs | 13 ++ crates/stages/api/src/error.rs | 8 + crates/stages/api/src/pipeline/builder.rs | 20 +- crates/stages/api/src/pipeline/event.rs | 6 + crates/stages/api/src/pipeline/mod.rs | 80 ++++++- crates/stages/api/src/test_utils.rs | 18 ++ crates/stages/stages/Cargo.toml | 2 +- crates/stages/stages/src/sets.rs | 20 +- crates/stages/stages/src/stages/headers.rs | 250 ++++++++++++++++++++- docs/vocs/docs/pages/run/configuration.mdx | 4 + 11 files changed, 412 insertions(+), 26 deletions(-) diff --git a/crates/cli/commands/src/stage/run.rs b/crates/cli/commands/src/stage/run.rs index c8601a3cf4b..7f514b4bed7 100644 --- a/crates/cli/commands/src/stage/run.rs +++ b/crates/cli/commands/src/stage/run.rs @@ -196,13 +196,16 @@ impl }; 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 index ad0e71b0c62..8046807f3e1 100644 --- 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/stages/api/src/error.rs b/crates/stages/api/src/error.rs index fab6e849d0e..58ca0619372 100644 --- 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), diff --git a/crates/stages/api/src/pipeline/builder.rs b/crates/stages/api/src/pipeline/builder.rs index 818b037da75..396d978b510 100644 --- a/crates/stages/api/src/pipeline/builder.rs +++ b/crates/stages/api/src/pipeline/builder.rs @@ -1,7 +1,11 @@ -use crate::{pipeline::BoxedStage, MetricEventsSender, Pipeline, Stage, StageId, StageSet}; +use crate::{ + pipeline::{BoxedStage, DEFAULT_EXECUTE_READY_TIMEOUT}, + MetricEventsSender, Pipeline, Stage, StageId, StageSet, +}; use alloy_primitives::{BlockNumber, B256}; use reth_provider::{providers::ProviderNodeTypes, DatabaseProviderFactory, ProviderFactory}; use reth_static_file::StaticFileProducer; +use std::time::Duration; use tokio::sync::watch; /// Builds a [`Pipeline`]. @@ -15,6 +19,7 @@ pub struct PipelineBuilder { tip_tx: Option>, metrics_tx: Option, fail_on_unwind: bool, + execute_ready_timeout: Duration, } impl PipelineBuilder { @@ -67,6 +72,14 @@ impl PipelineBuilder { self } + /// Set the total time budget a stage is given to keep failing [`Stage::execute_ready`] + /// before the pipeline gives up the current run with + /// [`ControlFlow::NoProgress`](crate::ControlFlow::NoProgress). + pub const fn with_execute_ready_timeout(mut self, timeout: Duration) -> Self { + self.execute_ready_timeout = timeout; + self + } + /// Builds the final [`Pipeline`] using the given database. pub fn build( self, @@ -77,7 +90,8 @@ impl PipelineBuilder { N: ProviderNodeTypes, ProviderFactory: DatabaseProviderFactory, { - let Self { stages, max_block, tip_tx, metrics_tx, fail_on_unwind } = self; + let Self { stages, max_block, tip_tx, metrics_tx, fail_on_unwind, execute_ready_timeout } = + self; Pipeline { provider_factory, stages, @@ -90,6 +104,7 @@ impl PipelineBuilder { fail_on_unwind, last_detached_head_unwind_target: None, detached_head_attempts: 0, + execute_ready_timeout, } } } @@ -102,6 +117,7 @@ impl Default for PipelineBuilder { tip_tx: None, metrics_tx: None, fail_on_unwind: false, + execute_ready_timeout: DEFAULT_EXECUTE_READY_TIMEOUT, } } } diff --git a/crates/stages/api/src/pipeline/event.rs b/crates/stages/api/src/pipeline/event.rs index 879725886cf..70a47a1b545 100644 --- a/crates/stages/api/src/pipeline/event.rs +++ b/crates/stages/api/src/pipeline/event.rs @@ -64,6 +64,12 @@ pub enum PipelineEvent { /// The stage that encountered an error. stage_id: StageId, }, + /// Emitted when a stage repeatedly failed to become ready for execution within the + /// pipeline's readiness timeout and the pipeline gave up the current run. + TimedOut { + /// The stage that timed out. + stage_id: StageId, + }, /// Emitted when a stage was skipped due to it's run conditions not being met: /// /// - The stage might have progressed beyond the point of our target block diff --git a/crates/stages/api/src/pipeline/mod.rs b/crates/stages/api/src/pipeline/mod.rs index 142e5ad4aa2..720adbb7d0e 100644 --- a/crates/stages/api/src/pipeline/mod.rs +++ b/crates/stages/api/src/pipeline/mod.rs @@ -34,6 +34,10 @@ use progress::*; use reth_errors::RethResult; pub use set::*; +/// Default total time budget a stage is given to keep failing [`Stage::execute_ready`] before +/// the pipeline gives up the current run with [`ControlFlow::NoProgress`]. +pub const DEFAULT_EXECUTE_READY_TIMEOUT: Duration = Duration::from_secs(5 * 60); + /// A container for a queued stage. pub(crate) type BoxedStage = Box>; @@ -92,6 +96,14 @@ pub struct Pipeline { /// Number of consecutive unwind attempts due to [`StageError::DetachedHead`] for the current /// fork. detached_head_attempts: u64, + /// Total time budget a stage is given to keep failing [`Stage::execute_ready`] before the + /// pipeline gives up on it for the current run with [`ControlFlow::NoProgress`]. + /// + /// This bounds how long the pipeline retries a stage that cannot become ready, e.g. the + /// headers stage waiting on a download no peer can serve, so control returns to the caller + /// instead of retrying a stale target forever. It only counts consecutive readiness + /// failures, so slow but progressing stages are not affected. + execute_ready_timeout: Duration, } impl Pipeline { @@ -439,6 +451,12 @@ impl Pipeline { let mut made_progress = false; let target = self.max_block.or(previous_stage); + // The deadline for the current streak of `execute_ready` failures. It is armed on the + // first failure and cleared once the stage becomes ready, so it only cuts the retry + // loop of a stage that keeps failing readiness and never a long, healthy readiness + // wait, e.g. a large header download. + let mut ready_deadline: Option = None; + loop { let prev_checkpoint = self.provider_factory.get_stage_checkpoint(stage_id)?; @@ -477,10 +495,33 @@ impl Pipeline { self.event_sender.notify(PipelineEvent::Error { stage_id }); match self.on_stage_error(stage_id, prev_checkpoint, err)? { Some(ctrl) => return Ok(ctrl), - None => continue, + None => { + if ready_deadline.is_none() { + // a `checked_add` overflow means the timeout is effectively + // unbounded, so the deadline stays unarmed + ready_deadline = Instant::now().checked_add(self.execute_ready_timeout); + } + if ready_deadline.is_some_and(|deadline| Instant::now() >= deadline) { + error!( + target: "sync::pipeline", + stage = %stage_id, + timeout = ?self.execute_ready_timeout, + "Stage kept failing to become ready within the timeout, giving up on it for this pipeline run" + ); + self.event_sender.notify(PipelineEvent::TimedOut { stage_id }); + return Ok(ControlFlow::NoProgress { + block_number: prev_checkpoint + .map(|checkpoint| checkpoint.block_number), + }) + } + continue + } }; } + // the stage became ready, so end any failure streak + ready_deadline = None; + let stage_started_at = Instant::now(); let provider_rw = self.provider_factory.database_provider_rw()?; @@ -790,6 +831,43 @@ mod tests { assert_eq!(post_unwind_commit_counter_b.load(Ordering::Relaxed), 0); } + /// Gives up the pipeline run when a stage keeps stalling beyond the readiness timeout, + /// returning control to the caller instead of retrying the stage forever. + #[tokio::test] + async fn pipeline_gives_up_run_on_stalled_stage() { + let provider_factory = create_test_provider_factory(); + + let stage = TestStage::new(StageId::Other("A")) + .add_ready(Err(StageError::Stalled("download made no progress".to_owned()))); + + let mut pipeline = Pipeline::::builder() + .add_stage(stage) + .with_execute_ready_timeout(Duration::ZERO) + .build( + provider_factory.clone(), + StaticFileProducer::new(provider_factory.clone(), PruneModes::default()), + ); + let events = pipeline.events(); + + let result = pipeline.run_loop().await.unwrap(); + assert_eq!(result, ControlFlow::NoProgress { block_number: None }); + drop(pipeline); + + assert_eq!( + events.collect::>().await, + vec![ + PipelineEvent::Prepare { + pipeline_stages_progress: PipelineStagesProgress { current: 1, total: 1 }, + stage_id: StageId::Other("A"), + checkpoint: None, + target: None, + }, + PipelineEvent::Error { stage_id: StageId::Other("A") }, + PipelineEvent::TimedOut { stage_id: StageId::Other("A") }, + ] + ); + } + /// Unwinds a simple pipeline. #[tokio::test] async fn unwind_pipeline() { diff --git a/crates/stages/api/src/test_utils.rs b/crates/stages/api/src/test_utils.rs index 1f15e55140e..abb9bb0d85b 100644 --- a/crates/stages/api/src/test_utils.rs +++ b/crates/stages/api/src/test_utils.rs @@ -7,6 +7,7 @@ use std::{ atomic::{AtomicUsize, Ordering}, Arc, }, + task::{Context, Poll}, }; /// A test stage that can be used for testing. @@ -15,6 +16,7 @@ use std::{ #[derive(Debug)] pub struct TestStage { id: StageId, + ready_outcomes: VecDeque>, exec_outputs: VecDeque>, unwind_outputs: VecDeque>, post_execute_commit_counter: Arc, @@ -25,6 +27,7 @@ impl TestStage { pub fn new(id: StageId) -> Self { Self { id, + ready_outcomes: VecDeque::new(), exec_outputs: VecDeque::new(), unwind_outputs: VecDeque::new(), post_execute_commit_counter: Arc::new(AtomicUsize::new(0)), @@ -50,6 +53,13 @@ impl TestStage { self } + /// Queues an outcome for [`Stage::poll_execute_ready`]. Once all queued outcomes are + /// consumed, the stage reports ready. + pub fn add_ready(mut self, outcome: Result<(), StageError>) -> Self { + self.ready_outcomes.push_back(outcome); + self + } + pub fn add_unwind(mut self, output: Result) -> Self { self.unwind_outputs.push_back(output); self @@ -73,6 +83,14 @@ impl Stage for TestStage { self.id } + fn poll_execute_ready( + &mut self, + _cx: &mut Context<'_>, + _input: ExecInput, + ) -> Poll> { + Poll::Ready(self.ready_outcomes.pop_front().unwrap_or(Ok(()))) + } + fn execute(&mut self, _: &Provider, _input: ExecInput) -> Result { self.exec_outputs .pop_front() diff --git a/crates/stages/stages/Cargo.toml b/crates/stages/stages/Cargo.toml index 46eba3e0786..787a407be3c 100644 --- a/crates/stages/stages/Cargo.toml +++ b/crates/stages/stages/Cargo.toml @@ -50,7 +50,7 @@ alloy-consensus.workspace = true alloy-rlp.workspace = true # async -tokio = { workspace = true, features = ["sync"] } +tokio = { workspace = true, features = ["sync", "time"] } futures-util.workspace = true # observability diff --git a/crates/stages/stages/src/sets.rs b/crates/stages/stages/src/sets.rs index 0edb14e0c85..73956d56e1e 100644 --- a/crates/stages/stages/src/sets.rs +++ b/crates/stages/stages/src/sets.rs @@ -254,7 +254,10 @@ where HeaderStage: Stage, { StageSetBuilder::default() - .add_stage(HeaderStage::new(provider, header_downloader, tip, stages_config.etl)) + .add_stage( + HeaderStage::new(provider, header_downloader, tip, stages_config.etl.clone()) + .with_stall_timeout(stages_config.headers.stall_timeout), + ) .add_stage(bodies) } } @@ -278,12 +281,15 @@ where } builder - .add_stage(HeaderStage::new( - self.provider, - self.header_downloader, - self.tip, - self.stages_config.etl.clone(), - )) + .add_stage( + HeaderStage::new( + self.provider, + self.header_downloader, + self.tip, + self.stages_config.etl.clone(), + ) + .with_stall_timeout(self.stages_config.headers.stall_timeout), + ) .add_stage(BodyStage::new(self.body_downloader)) } } diff --git a/crates/stages/stages/src/stages/headers.rs b/crates/stages/stages/src/stages/headers.rs index f9ca2a86f3a..0b8487bff76 100644 --- a/crates/stages/stages/src/stages/headers.rs +++ b/crates/stages/stages/src/stages/headers.rs @@ -25,11 +25,20 @@ use reth_stages_api::{ StageCheckpoint, StageError, StageId, UnwindInput, UnwindOutput, }; use reth_static_file_types::StaticFileSegment; -use std::task::{ready, Context, Poll}; +use std::{ + future::Future, + pin::Pin, + task::{Context, Poll}, + time::Duration, +}; use tokio::sync::watch; use tracing::*; +/// Default duration after which the header download is considered stalled if the downloader +/// yielded no headers while the stage is waiting on it. +pub const DEFAULT_HEADER_DOWNLOAD_STALL_TIMEOUT: Duration = Duration::from_secs(30); + /// The headers stage. /// /// The headers stage downloads all block headers from the highest block in storage to @@ -59,6 +68,15 @@ pub struct HeaderStage { header_collector: Collector, /// Returns true if the ETL collector has all necessary headers to fill the gap. is_etl_ready: bool, + /// Duration after which the download is considered stalled if the downloader yielded no + /// headers while the stage is waiting on it. + stall_timeout: Duration, + /// Deadline armed while the downloader is pending; reset whenever it yields headers. + /// + /// When it elapses the stage fails with a recoverable [`StageError::Stalled`] instead of + /// waiting forever on a download that cannot complete, e.g. because no connected peer can + /// serve the requested range. + stall_deadline: Option>>, } // === impl HeaderStage === @@ -82,15 +100,25 @@ where hash_collector: Collector::new(etl_config.file_size / 2, etl_config.dir.clone()), header_collector: Collector::new(etl_config.file_size / 2, etl_config.dir), is_etl_ready: false, + stall_timeout: DEFAULT_HEADER_DOWNLOAD_STALL_TIMEOUT, + stall_deadline: None, } } + /// Sets the duration after which the download is considered stalled if the downloader + /// yielded no headers while the stage is waiting on it. + pub const fn with_stall_timeout(mut self, stall_timeout: Duration) -> Self { + self.stall_timeout = stall_timeout; + self + } + /// Clear all ETL state. Called on error paths to prevent buffer pollution on retry. fn clear_etl_state(&mut self) { self.sync_gap = None; self.hash_collector.clear(); self.header_collector.clear(); self.is_etl_ready = false; + self.stall_deadline = None; } /// Write downloaded headers to storage from ETL. @@ -232,14 +260,20 @@ where // let the downloader know what to sync if self.sync_gap != Some(gap.clone()) { + // discard any headers collected for a previous gap, e.g. by an attempt that stalled, + // so ranges of different targets are never mixed in the ETL collectors + self.clear_etl_state(); 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 { - match ready!(self.downloader.poll_next_unpin(cx)) { - Some(Ok(headers)) => { + match self.downloader.poll_next_unpin(cx) { + Poll::Ready(Some(Ok(headers))) => { + // the downloader is making progress, so reset the stall deadline + self.stall_deadline = None; + info!(target: "sync::stages::headers", total = headers.len(), from_block = headers.first().map(|h| h.number()), to_block = headers.last().map(|h| h.number()), "Received headers"); for header in headers { let header_number = header.number(); @@ -256,7 +290,11 @@ where } } } - Some(Err(HeadersDownloaderError::DetachedHead { local_head, header, error })) => { + Poll::Ready(Some(Err(HeadersDownloaderError::DetachedHead { + local_head, + header, + error, + }))) => { error!(target: "sync::stages::headers", %error, "Cannot attach header to head"); self.clear_etl_state(); return Poll::Ready(Err(StageError::DetachedHead { @@ -265,10 +303,34 @@ where error, })) } - None => { + Poll::Ready(None) => { self.clear_etl_state(); return Poll::Ready(Err(StageError::ChannelClosed)) } + Poll::Pending => { + // The downloader has nothing to yield. Arm the stall deadline so a download + // that cannot complete - e.g. because no connected peer can serve the + // requested range - fails loud and recoverable instead of pending forever. + let stall_deadline = self + .stall_deadline + .get_or_insert_with(|| Box::pin(tokio::time::sleep(self.stall_timeout))); + if stall_deadline.as_mut().poll(cx).is_ready() { + warn!( + target: "sync::stages::headers", + stall_timeout = ?self.stall_timeout, + ?tip, + "Header download made no progress, aborting stage execution" + ); + // keep the ETL and downloader state intact: a retry with an unchanged + // sync gap resumes the download where it left off + self.stall_deadline = None; + return Poll::Ready(Err(StageError::Stalled(format!( + "header download made no progress for {:?}", + self.stall_timeout + )))) + } + return Poll::Pending + } } } } @@ -388,14 +450,32 @@ where mod tests { use super::*; use crate::test_utils::{ - stage_test_suite, ExecuteStageTestRunner, StageTestRunner, UnwindStageTestRunner, + stage_test_suite, ExecuteStageTestRunner, StageTestRunner, TestStageDB, + UnwindStageTestRunner, }; use alloy_primitives::B256; use assert_matches::assert_matches; - use reth_provider::{DatabaseProviderFactory, ProviderFactory, StaticFileProviderFactory}; + use reth_consensus::test_utils::TestConsensus; + use reth_db::{test_utils::TempDatabase, Database, DatabaseEnv}; + use reth_downloaders::headers::reverse_headers::ReverseHeadersDownloaderBuilder; + use reth_network_p2p::{ + download::DownloadClient, + error::PeerRequestResult, + headers::client::{HeadersClient, HeadersRequest}, + priority::Priority, + test_utils::TestHeadersClient, + }; + use reth_network_peers::PeerId; + use reth_provider::{ + test_utils::MockNodeTypesWithDB, DatabaseProvider, DatabaseProviderFactory, + ProviderFactory, StaticFileProviderFactory, + }; use reth_stages_api::StageUnitCheckpoint; use reth_testing_utils::generators::{self, random_header, random_header_range}; - use std::sync::Arc; + use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }; use test_runner::HeadersTestRunner; mod test_runner { @@ -676,4 +756,158 @@ mod tests { assert!(runner.stage().hash_collector.is_empty()); assert!(runner.stage().header_collector.is_empty()); } + + /// A headers client that never completes requests, simulating a network where no peer can + /// serve the requested range. + #[derive(Debug, Default, Clone)] + struct PendingHeadersClient; + + impl DownloadClient for PendingHeadersClient { + fn report_bad_message(&self, _peer_id: PeerId) {} + + fn num_connected_peers(&self) -> usize { + 0 + } + } + + impl HeadersClient for PendingHeadersClient { + type Header = alloy_consensus::Header; + type Output = Pin< + Box< + dyn std::future::Future>> + + Send + + Sync, + >, + >; + + fn get_headers_with_priority( + &self, + _request: HeadersRequest, + _priority: Priority, + ) -> Self::Output { + Box::pin(std::future::pending()) + } + } + + /// A download that makes no progress must surface a recoverable error within the stall + /// timeout instead of pending forever. + #[tokio::test] + async fn poll_execute_ready_stalls_on_no_download_progress() { + let db = TestStageDB::default(); + let head = random_header(&mut generators::rng(), 0, None); + db.insert_headers(std::iter::once(&head)).unwrap(); + + let (_tip_tx, tip_rx) = watch::channel(B256::random()); + let downloader = ReverseHeadersDownloaderBuilder::default() + .build(Arc::new(PendingHeadersClient), Arc::new(TestConsensus::default())); + let mut stage = + HeaderStage::new(db.factory.clone(), downloader, tip_rx, EtlConfig::default()) + .with_stall_timeout(Duration::from_millis(100)); + let stage: &mut dyn Stage< + DatabaseProvider< as Database>::TXMut, MockNodeTypesWithDB>, + > = &mut stage; + + let input = ExecInput { target: Some(1000), checkpoint: None }; + let result = tokio::time::timeout( + Duration::from_secs(10), + std::future::poll_fn(|cx| stage.poll_execute_ready(cx, input)), + ) + .await + .expect("stage must fail fast instead of waiting on the download forever"); + assert_matches!(result, Err(StageError::Stalled(_))); + } + + /// A headers client that keeps requests pending until it is opened. + #[derive(Debug, Default, Clone)] + struct GatedHeadersClient { + inner: TestHeadersClient, + is_open: Arc, + notify: Arc, + } + + impl GatedHeadersClient { + /// Allows all pending and future requests to be served. + fn open(&self) { + self.is_open.store(true, Ordering::Relaxed); + self.notify.notify_waiters(); + } + } + + impl DownloadClient for GatedHeadersClient { + fn report_bad_message(&self, _peer_id: PeerId) {} + + fn num_connected_peers(&self) -> usize { + 1 + } + } + + impl HeadersClient for GatedHeadersClient { + type Header = alloy_consensus::Header; + type Output = Pin< + Box< + dyn std::future::Future>> + + Send + + Sync, + >, + >; + + fn get_headers_with_priority( + &self, + request: HeadersRequest, + priority: Priority, + ) -> Self::Output { + let this = self.clone(); + Box::pin(async move { + while !this.is_open.load(Ordering::Relaxed) { + this.notify.notified().await; + } + this.inner.get_headers_with_priority(request, priority).await + }) + } + } + + /// A retry after a stall resumes the download of the unchanged sync gap instead of + /// restarting it. + #[tokio::test] + async fn poll_execute_ready_resumes_after_stall() { + let db = TestStageDB::default(); + let mut rng = generators::rng(); + let head = random_header(&mut rng, 0, None); + db.insert_headers(std::iter::once(&head)).unwrap(); + let headers = random_header_range(&mut rng, 1..11, head.hash()); + let tip = headers.last().unwrap(); + + // the client has the headers but cannot serve them until it is opened + let client = GatedHeadersClient::default(); + client.inner.extend(headers.iter().rev().map(|h| h.clone_header())).await; + + let (_tip_tx, tip_rx) = watch::channel(tip.hash()); + let downloader = ReverseHeadersDownloaderBuilder::default() + .build(Arc::new(client.clone()), Arc::new(TestConsensus::default())); + let mut stage = + HeaderStage::new(db.factory.clone(), downloader, tip_rx, EtlConfig::default()) + .with_stall_timeout(Duration::from_millis(100)); + let stage: &mut dyn Stage< + DatabaseProvider< as Database>::TXMut, MockNodeTypesWithDB>, + > = &mut stage; + let input = ExecInput { target: Some(10), checkpoint: None }; + + let result = tokio::time::timeout( + Duration::from_secs(10), + std::future::poll_fn(|cx| stage.poll_execute_ready(cx, input)), + ) + .await + .expect("stage must fail fast instead of waiting on the download forever"); + assert_matches!(result, Err(StageError::Stalled(_))); + + // serve the headers: the retry must resume the download and become ready + client.open(); + let result = tokio::time::timeout( + Duration::from_secs(10), + std::future::poll_fn(|cx| stage.poll_execute_ready(cx, input)), + ) + .await + .expect("stage must become ready once the download can complete"); + assert_matches!(result, Ok(())); + } } diff --git a/docs/vocs/docs/pages/run/configuration.mdx b/docs/vocs/docs/pages/run/configuration.mdx index 64a67e0b82c..281187f3200 100644 --- a/docs/vocs/docs/pages/run/configuration.mdx +++ b/docs/vocs/docs/pages/run/configuration.mdx @@ -83,6 +83,10 @@ downloader_request_limit = 1000 # Lower thresholds correspond to more frequent disk I/O (writes), # but lowers memory usage commit_threshold = 10000 +# The time after which the header download is considered stalled if it made no +# progress, causing the headers stage to fail with a recoverable error instead +# of waiting forever, e.g. when no connected peer can serve the requested range. +stall_timeout = "30s" ``` ### `bodies` From faef83e11ef6f8935cd2513e8c2bd9712d60b156 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Thu, 6 Aug 2026 16:43:32 +0300 Subject: [PATCH 3/3] fix(engine): return the network to idle when backfill finishes A finished backfill only returned the network to `SyncState::Idle` when the debug-only `--debug.startup-sync-state-idle` flag was set (the unconditional transition was removed in #16742 and re-added behind the flag in #19429). A backfill that ends without reaching its target - now reachable via the stage stall handling - therefore left the node marked syncing indefinitely, suppressing transaction gossip, until live sync happened to advance. This contributed to wedged OP Stack nodes in production (Ink Mainnet 2026-07-16, unichain 2026-08-05). Make the idle transition on `BackfillSyncFinished` unconditional again. If the node is still behind, `on_backfill_sync_finished` re-checks the distance against the latest tracked fork choice state and re-triggers backfill (emitting `BackfillSyncStarted`, which marks the network as syncing again) or issues the remaining downloads, so the idle window between consecutive runs is brief. This also means the initial-sync latch flips after the first backfill run instead of at the first live-sync block, matching pre-#16742 behavior; a brief syncing/idle flap between runs is preferable to a node that stays "syncing" forever while wedged. The flag keeps its startup semantics. Also warn when the connected-peer count reaches zero while the node is syncing - a sync that has no peers cannot make progress, and this state was previously invisible at default log levels. Co-Authored-By: Claude Fable 5 --- crates/net/network/src/manager.rs | 7 +++++++ crates/node/builder/src/launch/engine.rs | 10 +++++++--- crates/node/core/src/args/debug.rs | 6 +++--- docs/vocs/docs/pages/cli/reth/node.mdx | 4 ++-- 4 files changed, 19 insertions(+), 8 deletions(-) diff --git a/crates/net/network/src/manager.rs b/crates/net/network/src/manager.rs index 515b27d690b..806e4462596 100644 --- 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 NetworkManager { "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 index 21e8fcdf40d..34382bcacdb 100644 --- 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 index 83074c0f66e..2897723f932 100644 --- 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, - /// 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/docs/vocs/docs/pages/cli/reth/node.mdx b/docs/vocs/docs/pages/cli/reth/node.mdx index fcf9e6f7f01..0abce3a6968 100644 --- a/docs/vocs/docs/pages/cli/reth/node.mdx +++ b/docs/vocs/docs/pages/cli/reth/node.mdx @@ -800,9 +800,9 @@ Debug: The URL of the ethstats server to connect to. Example: `nodename:secret@host:port` --debug.startup-sync-state-idle - 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. Database: --db.log-level