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/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 { 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/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/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 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`