From 4e6745656d8ee17c8b4036deee94224da65b827d Mon Sep 17 00:00:00 2001 From: will-2012 <117156346+will-2012@users.noreply.github.com> Date: Mon, 24 Aug 2026 11:28:03 +0000 Subject: [PATCH 1/3] fix(rpc): stop cancelled blocking IO tasks (#26776) Co-authored-by: Matthias Seitz (cherry picked from commit de401d9b403481fdfe86a6d1841f77d460f41fc9) --- .../rpc-eth-api/src/helpers/blocking_task.rs | 18 +++++++++++++++--- crates/rpc/rpc/src/eth/filter.rs | 16 ++++++++++++---- 2 files changed, 27 insertions(+), 7 deletions(-) diff --git a/crates/rpc/rpc-eth-api/src/helpers/blocking_task.rs b/crates/rpc/rpc-eth-api/src/helpers/blocking_task.rs index de29c934609..5336d016455 100644 --- a/crates/rpc/rpc-eth-api/src/helpers/blocking_task.rs +++ b/crates/rpc/rpc-eth-api/src/helpers/blocking_task.rs @@ -164,6 +164,9 @@ pub trait SpawnBlocking: EthApiTypes + Clone + Send + Sync + 'static { let (tx, rx) = oneshot::channel(); let this = self.clone(); self.io_task_spawner().spawn_blocking_task(async move { + if tx.is_closed() { + return + } let res = f(this); let _ = tx.send(res); }); @@ -184,11 +187,20 @@ pub trait SpawnBlocking: EthApiTypes + Clone + Send + Sync + 'static { F: FnOnce(Self) -> Fut + Send + 'static, R: Send + 'static, { - let (tx, rx) = oneshot::channel(); + let (mut tx, rx) = oneshot::channel(); let this = self.clone(); self.io_task_spawner().spawn_blocking_task(async move { - let res = f(this).await; - let _ = tx.send(res); + let fut = f(this); + tokio::pin!(fut); + let res = tokio::select! { + // Futures executed here may perform blocking work before their first yield. + biased; + _ = tx.closed() => None, + res = &mut fut => Some(res), + }; + if let Some(res) = res { + let _ = tx.send(res); + } }); async move { rx.await.map_err(|_| EthApiError::InternalEthError)? } diff --git a/crates/rpc/rpc/src/eth/filter.rs b/crates/rpc/rpc/src/eth/filter.rs index f201b2f9418..da5a5d5da66 100644 --- a/crates/rpc/rpc/src/eth/filter.rs +++ b/crates/rpc/rpc/src/eth/filter.rs @@ -658,12 +658,20 @@ where return Err(EthFilterError::QueryExceedsMaxBlocks(max_blocks_per_filter)) } - let (tx, rx) = oneshot::channel(); + let (mut tx, rx) = oneshot::channel(); let this = self.clone(); self.task_spawner.spawn_blocking_task(async move { - let res = - this.get_logs_in_block_range_inner(&filter, from_block, to_block, limits).await; - let _ = tx.send(res); + let fut = this.get_logs_in_block_range_inner(&filter, from_block, to_block, limits); + tokio::pin!(fut); + let res = tokio::select! { + // Range scans perform blocking reads before their first yield. + biased; + _ = tx.closed() => None, + res = &mut fut => Some(res), + }; + if let Some(res) = res { + let _ = tx.send(res); + } }); rx.await.map_err(|_| EthFilterError::InternalError)? From 569a53188df908898a4de810f5b07f14d5e9532f Mon Sep 17 00:00:00 2001 From: Matthias Seitz Date: Fri, 4 Sep 2026 12:55:06 +0000 Subject: [PATCH 2/3] fix(rpc): hold a blocking IO permit for eth_getLogs range scans (#26999) (cherry picked from commit 1f94488d1e42866d3f8280702a546d42a7b288af) --- .../rpc-eth-api/src/helpers/blocking_task.rs | 6 +-- crates/rpc/rpc/src/eth/filter.rs | 40 +++++++++++++++++++ 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/crates/rpc/rpc-eth-api/src/helpers/blocking_task.rs b/crates/rpc/rpc-eth-api/src/helpers/blocking_task.rs index 5336d016455..f6918948cf0 100644 --- a/crates/rpc/rpc-eth-api/src/helpers/blocking_task.rs +++ b/crates/rpc/rpc-eth-api/src/helpers/blocking_task.rs @@ -40,7 +40,7 @@ pub trait SpawnBlocking: EthApiTypes + Clone + Send + Sync + 'static { /// Returns handle to semaphore for blocking IO tasks. /// /// This semaphore is used to limit concurrent blocking IO operations like `eth_call`, - /// `eth_estimateGas`, and similar methods that require EVM execution. + /// `eth_estimateGas` and the `eth_getLogs` range scans. fn blocking_io_task_guard(&self) -> &Arc; /// Acquires a permit from the tracing task semaphore. @@ -80,8 +80,8 @@ pub trait SpawnBlocking: EthApiTypes + Clone + Send + Sync + 'static { /// Acquires a permit from the blocking IO request semaphore. /// - /// This should be used for operations like `eth_call`, `eth_estimateGas`, and similar methods - /// that require EVM execution and are spawned as blocking tasks. + /// This should be used for operations like `eth_call`, `eth_estimateGas` and the `eth_getLogs` + /// range scans, which are spawned as blocking tasks. /// /// See also [`Semaphore::acquire_owned`](`tokio::sync::Semaphore::acquire_owned`). fn acquire_owned_blocking_io( diff --git a/crates/rpc/rpc/src/eth/filter.rs b/crates/rpc/rpc/src/eth/filter.rs index da5a5d5da66..1f98b441d85 100644 --- a/crates/rpc/rpc/src/eth/filter.rs +++ b/crates/rpc/rpc/src/eth/filter.rs @@ -658,9 +658,19 @@ where return Err(EthFilterError::QueryExceedsMaxBlocks(max_blocks_per_filter)) } + // The scan occupies a blocking thread until it completes, so it shares the budget for + // blocking IO requests with `eth_call` and friends instead of pinning an unbounded number + // of pool threads. + let permit = self + .eth_api + .acquire_owned_blocking_io() + .await + .map_err(|_| EthFilterError::InternalError)?; + let (mut tx, rx) = oneshot::channel(); let this = self.clone(); self.task_spawner.spawn_blocking_task(async move { + let _permit = permit; let fut = this.get_logs_in_block_range_inner(&filter, from_block, to_block, limits); tokio::pin!(fut); let res = tokio::select! { @@ -2029,4 +2039,34 @@ mod tests { assert_eq!(logs[0].block_hash, Some(expected_hashes[0])); // block 100 assert_eq!(logs[1].block_hash, Some(expected_hashes[2])); // block 102 } + + #[tokio::test] + async fn test_range_scan_waits_for_blocking_io_permit() { + use reth_rpc_eth_api::helpers::SpawnBlocking; + + let provider = MockEthProvider::default(); + provider.add_header(FixedBytes::random(), alloy_consensus::Header::default()); + let eth_api = build_test_eth_api(provider); + + // take every permit so the scan has to wait for one + let guard = eth_api.blocking_io_task_guard().clone(); + let permits = + guard.clone().acquire_many_owned(guard.available_permits() as u32).await.unwrap(); + + let eth_filter = EthFilter::new(eth_api, EthFilterConfig::default(), Runtime::test()); + let scan = eth_filter.inner.clone().get_logs_in_block_range( + Filter::default(), + 0, + 0, + QueryLimits::default(), + ); + tokio::pin!(scan); + assert!( + tokio::time::timeout(Duration::from_millis(100), &mut scan).await.is_err(), + "scan must wait for a blocking IO permit" + ); + + drop(permits); + assert!(scan.await.unwrap().is_empty()); + } } From cf0d35dec240e190c4924eb5445e211e969b681c Mon Sep 17 00:00:00 2001 From: Matthias Seitz Date: Fri, 4 Sep 2026 14:34:50 +0000 Subject: [PATCH 3/3] perf(rpc): scan eth_getLogs ranges window by window (#27000) (cherry picked from commit 055fafc723d1df096e551a14a80e76d2483e97fa) Conflict resolution for v2.5.2: kept the range-mode header buffer at Vec::with_capacity(self.max_range). The remaining-header bound comes from upstream PR 26971, which is not part of this backport. Co-authored-by: Toshi --- crates/rpc/rpc/src/eth/filter.rs | 351 ++++++++++++++++++++++--------- 1 file changed, 248 insertions(+), 103 deletions(-) diff --git a/crates/rpc/rpc/src/eth/filter.rs b/crates/rpc/rpc/src/eth/filter.rs index 1f98b441d85..90986db45a8 100644 --- a/crates/rpc/rpc/src/eth/filter.rs +++ b/crates/rpc/rpc/src/eth/filter.rs @@ -81,12 +81,20 @@ const BLOOM_ADJUSTMENT_MIN_BLOCKS: u64 = 100; /// The maximum number of headers we read at once when handling a range filter. const MAX_HEADERS_RANGE: u64 = 1_000; // with ~530bytes per header this is ~500kb -/// Threshold for enabling parallel processing in range mode +// Cached mode is only reachable for ranges that fit into a single header window, which is what +// keeps the mode decision independent of how a range is split. +const _: () = assert!(CACHED_MODE_BLOCK_THRESHOLD <= MAX_HEADERS_RANGE); + +/// Minimum number of bloom matching blocks in a header window for fetching their receipts in +/// parallel const PARALLEL_PROCESSING_THRESHOLD: usize = 1000; /// Default concurrency for parallel processing const DEFAULT_PARALLEL_CONCURRENCY: usize = 4; +/// Maximum number of blocks whose receipts the parallel fetching holds in memory at once +const MAX_PARALLEL_BATCH_SIZE: usize = 256; + /// `Eth` filter RPC implementation. /// /// This type handles `eth_` rpc requests related to filters (`eth_getLogs`). @@ -703,15 +711,21 @@ where limits: QueryLimits, ) -> Result>, EthFilterError> { let mut all_logs = Vec::new(); - let mut matching_headers = Vec::new(); // get current chain tip to determine processing mode let chain_tip = self.provider().best_block_number()?; - // first collect all headers that match the bloom filter for cached mode decision + // Scan the range window by window so that receipts are fetched while headers are still + // being read: the log limit can end the query after the first window, memory is bounded by + // one window, and a cancelled query stops at the next window. for (from, to) in BlockRangeInclusiveIter::new(from_block..=to_block, self.max_headers_range) { + // reading headers is blocking, this gives the cancellation check a chance to run + tokio::task::yield_now().await; + + // collect the headers of this window that match the bloom filter + let mut matching_headers = Vec::new(); let headers = self.provider().headers_range(from..=to)?; let mut headers_iter = headers.into_iter().peekable(); @@ -736,58 +750,61 @@ where matching_headers.push(SealedHeader::new(header, block_hash)); } - } - // initialize the appropriate range mode based on collected headers - let mut range_mode = RangeMode::new( - self.clone(), - matching_headers, - from_block, - to_block, - self.max_headers_range, - chain_tip, - ); + // initialize the appropriate range mode based on collected headers + let mut range_mode = RangeMode::new( + self.clone(), + matching_headers, + from_block, + to_block, + self.max_headers_range, + chain_tip, + ); - // iterate through the range mode to get receipts and blocks - while let Some(ReceiptBlockResult { receipts, recovered_block, header }) = - range_mode.next().await? - { - let num_hash = header.num_hash(); - append_matching_block_logs( - &mut all_logs, - self.eth_api.converter(), - recovered_block - .map(ProviderOrBlock::Block) - .unwrap_or_else(|| ProviderOrBlock::Provider(self.provider())), - filter, - &header, - &receipts, - false, - )?; - - // size check but only if range is multiple blocks, so we always return all - // logs of a single block - let is_multi_block_range = from_block != to_block; - if let Some(max_logs_per_response) = limits.max_logs_per_response && - is_multi_block_range && - all_logs.len() > max_logs_per_response + // iterate through the range mode to get receipts and blocks + while let Some(ReceiptBlockResult { receipts, recovered_block, header }) = + range_mode.next().await? { - let retry_to_block = - if num_hash.number == from_block { from_block } else { num_hash.number - 1 }; - - debug!( - target: "rpc::eth::filter", - logs_found = all_logs.len(), - max_logs_per_response, - from_block, - to_block = retry_to_block, - "Query exceeded max logs per response limit" - ); - return Err(EthFilterError::QueryExceedsMaxResults { - max_logs: max_logs_per_response, - from_block, - to_block: retry_to_block, - }); + let num_hash = header.num_hash(); + append_matching_block_logs( + &mut all_logs, + self.eth_api.converter(), + recovered_block + .map(ProviderOrBlock::Block) + .unwrap_or_else(|| ProviderOrBlock::Provider(self.provider())), + filter, + &header, + &receipts, + false, + )?; + + // size check but only if range is multiple blocks, so we always return all + // logs of a single block + let is_multi_block_range = from_block != to_block; + if let Some(max_logs_per_response) = limits.max_logs_per_response && + is_multi_block_range && + all_logs.len() > max_logs_per_response + { + let retry_to_block = if num_hash.number == from_block { + from_block + } else { + num_hash.number - 1 + }; + + debug!( + target: "rpc::eth::filter", + logs_found = all_logs.len(), + max_logs_per_response, + from_block, + to_block = retry_to_block, + "Query exceeded max logs per response limit" + ); + return Err(EthFilterError::QueryExceedsMaxResults { + max_logs: max_logs_per_response, + from_block, + to_block: retry_to_block, + }); + } } } @@ -1101,6 +1118,10 @@ impl< let use_cached_mode = Self::should_use_cached_mode(&sealed_headers, block_count, distance_from_tip); + // Fetching receipts in parallel is only worth the extra tasks when most of the window has + // them to read; the sequential path serves the rest from the receipt cache where it can + let parallel = sealed_headers.len() >= PARALLEL_PROCESSING_THRESHOLD; + if use_cached_mode && !sealed_headers.is_empty() { Self::Cached(CachedMode { filter_inner, headers_iter: sealed_headers.into_iter() }) } else { @@ -1108,7 +1129,8 @@ impl< filter_inner, iter: sealed_headers.into_iter().peekable(), next: VecDeque::new(), - max_range: max_headers_range as usize, + max_range: (max_headers_range as usize).min(MAX_PARALLEL_BATCH_SIZE), + parallel, pending_tasks: FuturesOrdered::new(), }) } @@ -1205,7 +1227,10 @@ struct RangeBlockMode< filter_inner: Arc>, iter: Peekable::Header>>>, next: VecDeque>, + /// Maximum number of consecutive blocks fetched by one batch of parallel tasks max_range: usize, + /// Whether receipts are fetched in parallel batches of consecutive blocks + parallel: bool, // Stream of ongoing receipt fetching tasks pending_tasks: FuturesOrdered>, } @@ -1237,6 +1262,15 @@ impl< return Ok(None); }; + if !self.parallel { + // Process the block on its own so that only one block's receipts are held + if let Some(result) = self.process_small_range(vec![next_header]).await? { + return Ok(Some(result)); + } + // Continue loop to check for more work + continue; + } + let mut range_headers = Vec::with_capacity(self.max_range); range_headers.push(next_header); @@ -1262,24 +1296,14 @@ impl< range_headers.push(next_header); } - // Check if we should use parallel processing for large ranges - let remaining_headers = self.iter.len() + range_headers.len(); - if remaining_headers >= PARALLEL_PROCESSING_THRESHOLD { - self.spawn_parallel_tasks(range_headers); - // Continue loop to await the spawned tasks - } else { - // Process small range sequentially and add results to buffer - if let Some(result) = self.process_small_range(range_headers).await? { - return Ok(Some(result)); - } - // Continue loop to check for more work - } + self.spawn_parallel_tasks(range_headers); + // Continue loop to await the spawned tasks } } /// Process a small range of headers sequentially /// - /// This is used when the remaining headers count is below [`PARALLEL_PROCESSING_THRESHOLD`]. + /// This is used for ranges below [`PARALLEL_PROCESSING_THRESHOLD`], one block at a time. async fn process_small_range( &mut self, range_headers: Vec::Header>>, @@ -1318,14 +1342,13 @@ impl< /// Spawn parallel tasks for processing a large range of headers /// - /// This is used when the remaining headers count is at or above - /// [`PARALLEL_PROCESSING_THRESHOLD`]. + /// This is used for ranges of at least [`PARALLEL_PROCESSING_THRESHOLD`] blocks. fn spawn_parallel_tasks( &mut self, range_headers: Vec::Header>>, ) { // Split headers into chunks - let chunk_size = std::cmp::max(range_headers.len() / DEFAULT_PARALLEL_CONCURRENCY, 1); + let chunk_size = range_headers.len().div_ceil(DEFAULT_PARALLEL_CONCURRENCY).max(1); let header_chunks = range_headers .into_iter() .chunks(chunk_size) @@ -1336,47 +1359,61 @@ impl< // Spawn each chunk as a separate task directly into the FuturesOrdered stream for chunk_headers in header_chunks { let filter_inner = self.filter_inner.clone(); - let chunk_task = Box::pin(async move { - let chunk_task = tokio::task::spawn_blocking(move || { - let mut chunk_results = Vec::with_capacity(chunk_headers.len()); - - for header in chunk_headers { - // Fetch directly from provider - RangeMode is used for older blocks - // unlikely to be cached - let receipts = match filter_inner - .provider() - .receipts_by_block(header.hash().into())? - { - Some(receipts) => Arc::new(receipts), - None => continue, // No receipts found - }; - - if !receipts.is_empty() { - chunk_results.push(ReceiptBlockResult { - receipts, - recovered_block: None, - header, - }); - } - } + let fetch = move || Self::fetch_chunk_receipts(&filter_inner, chunk_headers); - Ok(chunk_results) - }); + // A parallel task occupies an additional blocking thread, so it needs its own share + // of the blocking IO budget. Without one the chunk is fetched on this task instead. + let chunk_task: ReceiptFetchFuture = match self + .filter_inner + .eth_api + .blocking_io_task_guard() + .clone() + .try_acquire_owned() + { + Ok(permit) => Box::pin(async move { + let chunk_task = tokio::task::spawn_blocking(move || { + let _permit = permit; + fetch() + }); - // Await the blocking task and handle the result - match chunk_task.await { - Ok(Ok(chunk_results)) => Ok(chunk_results), - Ok(Err(e)) => Err(e), - Err(join_err) => { - trace!(target: "rpc::eth::filter", error = ?join_err, "Task join error"); - Err(EthFilterError::InternalError) + // Await the blocking task and handle the result + match chunk_task.await { + Ok(chunk_results) => chunk_results, + Err(join_err) => { + trace!(target: "rpc::eth::filter", error = ?join_err, "Task join error"); + Err(EthFilterError::InternalError) + } } - } - }); + }), + Err(_) => Box::pin(async move { fetch() }), + }; self.pending_tasks.push_back(chunk_task); } } + + /// Fetches the receipts of the given blocks from the provider. + fn fetch_chunk_receipts( + filter_inner: &EthFilterInner, + chunk_headers: Vec::Header>>, + ) -> Result>, EthFilterError> { + let mut chunk_results = Vec::with_capacity(chunk_headers.len()); + + for header in chunk_headers { + // Fetch directly from provider - RangeMode is used for older blocks + // unlikely to be cached + let receipts = match filter_inner.provider().receipts_by_block(header.hash().into())? { + Some(receipts) => Arc::new(receipts), + None => continue, // No receipts found + }; + + if !receipts.is_empty() { + chunk_results.push(ReceiptBlockResult { receipts, recovered_block: None, header }); + } + } + + Ok(chunk_results) + } } #[cfg(test)] @@ -1455,6 +1492,8 @@ mod tests { iter: headers.into_iter().peekable(), next: VecDeque::new(), max_range, + parallel: false, + pending_tasks: FuturesOrdered::new(), }; @@ -1530,6 +1569,8 @@ mod tests { iter: headers.into_iter().peekable(), next: VecDeque::from([mock_result_1, mock_result_2]), // Queue two results max_range: 100, + parallel: false, + pending_tasks: FuturesOrdered::new(), }; @@ -1597,6 +1638,8 @@ mod tests { iter: headers.into_iter().peekable(), next: VecDeque::new(), max_range: 100, + parallel: false, + pending_tasks: FuturesOrdered::new(), }; @@ -1665,6 +1708,8 @@ mod tests { iter: headers.into_iter().peekable(), next: VecDeque::new(), max_range: 3, // include the 3 blocks in the first queried results + parallel: false, + pending_tasks: FuturesOrdered::new(), }; @@ -1755,6 +1800,8 @@ mod tests { iter: headers.into_iter().peekable(), next: VecDeque::new(), max_range: 1, + parallel: false, + pending_tasks: FuturesOrdered::new(), }; @@ -2069,4 +2116,102 @@ mod tests { drop(permits); assert!(scan.await.unwrap().is_empty()); } + + #[tokio::test] + async fn test_range_scan_across_windows() { + use alloy_consensus::TxLegacy; + use alloy_primitives::{Address, Bloom, Bytes, Log, LogData, Signature}; + use reth_db_api::models::StoredBlockBodyIndices; + use reth_ethereum_primitives::{Block, BlockBody, Receipt, TransactionSigned}; + + let provider = MockEthProvider::default(); + let tx = TransactionSigned::new_unhashed( + TxLegacy { + chain_id: Some(1), + gas_price: 21_000, + gas_limit: 21_000, + ..Default::default() + } + .into(), + Signature::test_signature(), + ); + let receipt = Receipt { + tx_type: TxType::Legacy, + cumulative_gas_used: 21_000, + logs: vec![Log { + address: Address::ZERO, + data: LogData::new_unchecked(vec![], Bytes::new()), + }], + success: true, + }; + + // an empty filter matches every header, so every window is fetched in parallel while + // only three blocks have logs to return + let matching_blocks = [5u64, 1_200, 2_400]; + let mut parent_hash = FixedBytes::default(); + for number in 0..=2_500u64 { + let matches = matching_blocks.contains(&number); + let header = alloy_consensus::Header { + number, + parent_hash, + logs_bloom: if matches { Bloom::from([1u8; 256]) } else { Bloom::default() }, + ..Default::default() + }; + parent_hash = header.hash_slow(); + let transactions = if matches { vec![tx.clone()] } else { Vec::new() }; + provider.add_block( + parent_hash, + Block { header, body: BlockBody { transactions, ..Default::default() } }, + ); + if matches { + let tx_num = matching_blocks.iter().position(|b| *b == number).unwrap() as u64; + provider.add_receipts(number, vec![receipt.clone()]); + provider.add_block_body_indices( + number, + StoredBlockBodyIndices { first_tx_num: tx_num, tx_count: 1 }, + ); + } else { + provider.add_receipts(number, vec![]); + } + } + + let eth_filter = EthFilter::new( + build_test_eth_api(provider), + EthFilterConfig::default(), + Runtime::test(), + ); + let logs = eth_filter + .inner + .clone() + .get_logs_in_block_range(Filter::default(), 0, 2_500, QueryLimits::default()) + .await + .unwrap(); + let blocks = logs.iter().map(|log| log.block_number.unwrap()).collect::>(); + assert_eq!(blocks, matching_blocks); + + // the log limit ends the scan in the window that exceeds it and names the last block that + // fit as the range to retry with + let err = eth_filter + .inner + .clone() + .get_logs_in_block_range( + Filter::default(), + 0, + 2_500, + QueryLimits { max_blocks_per_filter: None, max_logs_per_response: Some(1) }, + ) + .await + .unwrap_err(); + assert!( + matches!( + err, + EthFilterError::QueryExceedsMaxResults { + max_logs: 1, + from_block: 0, + to_block: 1_199 + } + ), + "{err:?}" + ); + } }