From 0c45a4dc2512c510b5e67641c9e31b8aedfe19f3 Mon Sep 17 00:00:00 2001 From: Blake Emerson Date: Wed, 17 Jun 2026 03:05:05 -0500 Subject: [PATCH 1/3] perf(zaino-state): pipeline finalised bulk-sync block fetches The bulk-sync loop in write_blocks_to_height was serial and fetch-latency-bound, awaiting two sequential validator RPCs per block (get_block + get_commitment_tree_roots) with no overlap. build_indexed_block_from_source was split into fetch_block_data (the parallelizable fetch) and assemble_indexed_block (the sequential chainwork + IndexedBlock build), and the v1 bulk path was driven through a futures buffered(N) stream that keeps up to N fetches in flight while yielding them in height order, so only the fetch wait was overlapped. The v0 and experimental paths kept the serial helper. sync_fetch_concurrency (default 32) was added and documented alongside sync_write_batch_bytes in the example config. --- docs/example_configs/zainod.toml | 18 ++++ packages/zaino-common/src/config/storage.rs | 19 ++++ .../src/chain_index/finalised_state.rs | 87 ++++++++++++++++--- .../finalised_state/db/v1/write_core.rs | 65 ++++++++------ 4 files changed, 151 insertions(+), 38 deletions(-) diff --git a/docs/example_configs/zainod.toml b/docs/example_configs/zainod.toml index 136a7b2c3..0c3b57420 100644 --- a/docs/example_configs/zainod.toml +++ b/docs/example_configs/zainod.toml @@ -33,3 +33,21 @@ shard_power = 4 [storage.database] path = '/home/cupress/.cache/zaino' size = 128 + +# Approximate in-memory byte budget for the finalised-state bulk-sync write batch. +# Bulk sync buffers fetched blocks up to this many bytes, then writes the whole batch in one +# LMDB transaction with the random-keyed `spent` / `txid_location` entries inserted in sorted +# key order (a sequential B-tree sweep instead of random faults once the DB exceeds RAM); larger +# batches mean fewer sweeps. Peak RAM is roughly this budget plus the transaction's dirty pages, +# and it competes with the OS page cache the sweep relies on — larger is not always better. +# Defaults to 4 GiB (4294967296); raise it on large-RAM hosts. +sync_write_batch_bytes = 4294967296 + +# Number of blocks fetched concurrently during finalised-state bulk sync. +# The bulk-sync loop is fetch-latency-bound: each block needs sequential round-trips to the +# validator (block + commitment-tree roots), and a strictly serial loop leaves both zaino and the +# validator idle while it waits. This runs up to this many fetches in flight at once (yielded back +# in strict height order, so writes are unaffected), overlapping the latency until the validator's +# RPC handler saturates. Peak fetch memory is roughly this many buffered blocks, so keep it modest. +# Defaults to 32; tune up toward the validator's RPC capacity. +sync_fetch_concurrency = 32 diff --git a/packages/zaino-common/src/config/storage.rs b/packages/zaino-common/src/config/storage.rs index a7bbfc0f2..cd5578817 100644 --- a/packages/zaino-common/src/config/storage.rs +++ b/packages/zaino-common/src/config/storage.rs @@ -80,6 +80,19 @@ pub struct DatabaseConfig { /// better. Defaults to 4 GiB; raise it on large-RAM hosts. #[serde(default = "default_sync_write_batch_bytes")] pub sync_write_batch_bytes: u64, + /// Number of blocks fetched concurrently during finalised-state bulk sync. + /// + /// The bulk-sync ingestion loop is fetch-latency-bound: each block requires sequential + /// round-trips to the source (block + commitment-tree roots), and a strictly serial loop leaves + /// both zaino and the validator idle while it waits on the network. This knob runs up to this + /// many block fetches in flight at once (yielded back in strict height order, so the writer is + /// unaffected), overlapping the latency and multiplying fetch throughput until the source's RPC + /// handler saturates. + /// + /// Peak fetch memory is roughly this many buffered blocks, so keep it modest. Defaults to 32; + /// tune up toward the source's RPC capacity. + #[serde(default = "default_sync_fetch_concurrency")] + pub sync_fetch_concurrency: usize, } /// Default [`DatabaseConfig::sync_write_batch_bytes`]: 4 GiB. @@ -87,12 +100,18 @@ fn default_sync_write_batch_bytes() -> u64 { 4 * 1024 * 1024 * 1024 } +/// Default [`DatabaseConfig::sync_fetch_concurrency`]: 32 concurrent block fetches. +fn default_sync_fetch_concurrency() -> usize { + 32 +} + impl Default for DatabaseConfig { fn default() -> Self { Self { path: resolve_path_with_xdg_cache_defaults("zaino"), size: DatabaseSize::default(), sync_write_batch_bytes: default_sync_write_batch_bytes(), + sync_fetch_concurrency: default_sync_fetch_concurrency(), } } } diff --git a/packages/zaino-state/src/chain_index/finalised_state.rs b/packages/zaino-state/src/chain_index/finalised_state.rs index c8dfeab8d..7afbf5999 100644 --- a/packages/zaino-state/src/chain_index/finalised_state.rs +++ b/packages/zaino-state/src/chain_index/finalised_state.rs @@ -195,20 +195,34 @@ use crate::{ use std::{sync::Arc, time::Duration}; use tokio::time::{interval, MissedTickBehavior}; -/// Fetches the block at `height_int` from `source` and builds its [`IndexedBlock`], threading -/// `parent_chainwork` into the block metadata. +/// A block plus its resolved commitment-tree roots, fetched from a [`BlockchainSource`] but **not** +/// yet assembled into an [`IndexedBlock`]. /// -/// Shared by every backend's [`capability::DbWrite::write_blocks_to_height`] ingestion loop so the -/// fetch + commitment-tree-root + metadata assembly lives in one place regardless of which backend -/// owns the loop. -pub(crate) async fn build_indexed_block_from_source( +/// This is the parallelizable half of block ingestion: producing it needs nothing from the previous +/// height, so the bulk-sync loop fetches many concurrently (see [`fetch_block_data`]). The remaining +/// sequential step — threading `parent_chainwork` and building the [`IndexedBlock`] — is +/// [`assemble_indexed_block`]. +pub(crate) struct FetchedBlock { + /// Height the block was fetched at; kept for ordered progress logging and assembly errors. + pub(crate) height: u32, + block: Arc, + /// Sapling `(root, size)`, already gated on the Sapling activation height (defaults below it). + sapling: (zebra_chain::sapling::tree::Root, u64), + /// Orchard `(root, size)`, already gated on the NU5 activation height (defaults below it). + orchard: (zebra_chain::orchard::tree::Root, u64), +} + +/// Fetches the block at `height_int` and its Sapling/Orchard commitment-tree roots from `source`. +/// +/// This is the fetch-latency-bound, dependency-free half of ingestion (the block at height *h* needs +/// nothing from *h-1*), so the bulk-sync loop runs many of these concurrently. The sequential +/// chainwork + [`IndexedBlock`] assembly is split out into [`assemble_indexed_block`]. +pub(crate) async fn fetch_block_data( source: &S, - network: zaino_common::Network, sapling_activation_height: zebra_chain::block::Height, nu5_activation_height: Option, height_int: u32, - parent_chainwork: ChainWork, -) -> Result { +) -> Result { let block = match source .get_block(zebra_state::HashOrHeight::Height( zebra_chain::block::Height(height_int), @@ -233,7 +247,7 @@ pub(crate) async fn build_indexed_block_from_source( let is_orchard_active = nu5_activation_height .is_some_and(|nu5_activation_height| height_int >= nu5_activation_height.0); - let (sapling_root, sapling_size) = if is_sapling_active { + let sapling = if is_sapling_active { sapling_opt.ok_or_else(|| { FinalisedStateError::BlockchainSourceError(BlockchainSourceError::Unrecoverable( format!("missing Sapling commitment tree root for block {block_hash}"), @@ -243,7 +257,7 @@ pub(crate) async fn build_indexed_block_from_source( (zebra_chain::sapling::tree::Root::default(), 0) }; - let (orchard_root, orchard_size) = if is_orchard_active { + let orchard = if is_orchard_active { orchard_opt.ok_or_else(|| { FinalisedStateError::BlockchainSourceError(BlockchainSourceError::Unrecoverable( format!("missing Orchard commitment tree root for block {block_hash}"), @@ -253,6 +267,27 @@ pub(crate) async fn build_indexed_block_from_source( (zebra_chain::orchard::tree::Root::default(), 0) }; + Ok(FetchedBlock { + height: height_int, + block, + sapling, + orchard, + }) +} + +/// Assembles a [`FetchedBlock`] into an [`IndexedBlock`], threading `parent_chainwork` into the +/// block metadata. +/// +/// This is the sequential half of ingestion — chainwork chains off the previous block — and so must +/// run in strict height order even when [`fetch_block_data`] is parallelized. +pub(crate) fn assemble_indexed_block( + fetched: FetchedBlock, + network: zaino_common::Network, + parent_chainwork: ChainWork, +) -> Result { + let (sapling_root, sapling_size) = fetched.sapling; + let (orchard_root, orchard_size) = fetched.orchard; + let metadata = BlockMetadata::new( sapling_root, sapling_size as u32, @@ -262,14 +297,40 @@ pub(crate) async fn build_indexed_block_from_source( network.to_zebra_network(), ); - let block_with_metadata = BlockWithMetadata::new(block.as_ref(), metadata); + let block_with_metadata = BlockWithMetadata::new(fetched.block.as_ref(), metadata); IndexedBlock::try_from(block_with_metadata).map_err(|_| { FinalisedStateError::BlockchainSourceError(BlockchainSourceError::Unrecoverable(format!( - "error building block data at height {height_int}" + "error building block data at height {}", + fetched.height ))) }) } +/// Fetches the block at `height_int` from `source` and builds its [`IndexedBlock`], threading +/// `parent_chainwork` into the block metadata. +/// +/// A thin serial convenience over [`fetch_block_data`] + [`assemble_indexed_block`], used by the +/// backends whose ingestion loop fetches one block at a time (the v0 backend and the experimental +/// address-history path). The v1 bulk-sync loop calls the two halves directly so it can pipeline the +/// fetch. +pub(crate) async fn build_indexed_block_from_source( + source: &S, + network: zaino_common::Network, + sapling_activation_height: zebra_chain::block::Height, + nu5_activation_height: Option, + height_int: u32, + parent_chainwork: ChainWork, +) -> Result { + let fetched = fetch_block_data( + source, + sapling_activation_height, + nu5_activation_height, + height_int, + ) + .await?; + assemble_indexed_block(fetched, network, parent_chainwork) +} + use super::source::BlockchainSource; #[derive(Debug)] diff --git a/packages/zaino-state/src/chain_index/finalised_state/db/v1/write_core.rs b/packages/zaino-state/src/chain_index/finalised_state/db/v1/write_core.rs index 07a08ff9a..0dac3d321 100644 --- a/packages/zaino-state/src/chain_index/finalised_state/db/v1/write_core.rs +++ b/packages/zaino-state/src/chain_index/finalised_state/db/v1/write_core.rs @@ -60,7 +60,12 @@ impl DbWrite for DbV1 { height: Height, source: &S, ) -> Result<(), FinalisedStateError> { + #[cfg(feature = "transparent_address_history_experimental")] use crate::chain_index::finalised_state::build_indexed_block_from_source; + #[cfg(not(feature = "transparent_address_history_experimental"))] + use crate::chain_index::finalised_state::{assemble_indexed_block, fetch_block_data}; + #[cfg(not(feature = "transparent_address_history_experimental"))] + use futures::stream::StreamExt as _; use zebra_chain::parameters::NetworkUpgrade; let network = self.config.network; @@ -119,44 +124,54 @@ impl DbWrite for DbV1 { #[cfg(not(feature = "transparent_address_history_experimental"))] { let batch_budget = self.config.storage.database.sync_write_batch_bytes.max(1); - let mut next = start_height; + let concurrency = self.config.storage.database.sync_fetch_concurrency.max(1); + + // Pipeline the fetch: a strictly serial loop here is fetch-latency-bound (each block + // needs sequential round-trips to the source for the block + its commitment-tree roots, + // and nothing else runs while it waits — both zaino and the validator sit idle). The + // fetch of height `h` depends on nothing from `h-1`, so we run up to `concurrency` + // fetches at once. `buffered` (NOT `buffer_unordered`) yields them back in strict input + // (height) order, so the writer below still sees consecutive heights — only the wait is + // overlapped. The sequential `parent_chainwork` chaining and the batched write are + // unchanged, downstream of the fetch. + let mut fetched = futures::stream::iter(start_height..=height.0) + .map(|h| { + fetch_block_data(source, sapling_activation_height, nu5_activation_height, h) + }) + .buffered(concurrency); + let mut last_progress_log = std::time::Instant::now(); - while next <= height.0 { - // Fetch blocks (async; an LMDB write txn is `!Send` and cannot be held across the - // await) into a buffer, flushing on the *first* of: byte budget, block-count cap, or - // time cap. The count/time caps keep the first commit (and progress, and crash-loss - // window) prompt even on the tiny early-chain blocks, where the byte budget alone - // would buffer a huge number of blocks before committing. + loop { + // Assemble fetched blocks (in height order) into a buffer, flushing on the *first* + // of: byte budget, block-count cap, or time cap. The count/time caps keep the first + // commit (and progress, and crash-loss window) prompt even on the tiny early-chain + // blocks, where the byte budget alone would buffer a huge number of blocks before + // committing. let mut batch: Vec = Vec::new(); let mut batch_bytes: u64 = 0; + let mut batch_tip_height = 0u32; let batch_started = std::time::Instant::now(); - while next <= height.0 - && batch_bytes < batch_budget + while batch_bytes < batch_budget && batch.len() < SYNC_WRITE_BATCH_MAX_BLOCKS && batch_started.elapsed() < SYNC_WRITE_BATCH_MAX_INTERVAL { - let block = build_indexed_block_from_source( - source, - network, - sapling_activation_height, - nu5_activation_height, - next, - parent_chainwork, - ) - .await?; + let Some(fetched_block) = fetched.next().await else { + break; + }; + let fetched_block = fetched_block?; + // Sequential, ordered: chainwork chains off the previous block. + let block = assemble_indexed_block(fetched_block, network, parent_chainwork)?; parent_chainwork = block.context.chainwork; + batch_tip_height = block.context.index.height.0; batch_bytes = batch_bytes.saturating_add(approx_indexed_block_bytes(&block)); batch.push(block); - next += 1; - // In-flight progress: the block being fetched, throttled by time. (The committed - // tip is reported by the per-batch commit log below.) + // In-flight progress: the height just assembled, throttled by time. (The + // committed tip is reported by the per-batch commit log below.) if last_progress_log.elapsed() >= SYNC_PROGRESS_LOG_INTERVAL { info!( "write_blocks_to_height: syncing height {} / {} on {:?}", - next - 1, - height.0, - network + batch_tip_height, height.0, network ); last_progress_log = std::time::Instant::now(); } @@ -180,7 +195,7 @@ impl DbWrite for DbV1 { self.status.store(StatusType::Ready); info!( "write_blocks_to_height: committed batch to height {} ({} blocks)", - next - 1, + batch_tip_height, batch.len() ); } From 00ffaffe894a04b1fdddeac9ca5c588db40c0d9b Mon Sep 17 00:00:00 2001 From: Blake Emerson Date: Wed, 17 Jun 2026 03:39:27 -0500 Subject: [PATCH 2/3] Use HTTP/2 to reduce connections, retry failures, fix circuit breaker --- Cargo.lock | 1 + Cargo.toml | 1 + .../zaino-fetch/src/jsonrpsee/connector.rs | 30 ++++++-- packages/zaino-state/src/chain_index.rs | 72 +++++++++++++++---- 4 files changed, 87 insertions(+), 17 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d5a898d17..abbfd5553 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3430,6 +3430,7 @@ dependencies = [ "cookie", "cookie_store", "futures-core", + "h2", "http", "http-body", "http-body-util", diff --git a/Cargo.toml b/Cargo.toml index a782b338b..101bde087 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -71,6 +71,7 @@ url = "2.5" reqwest = { version = "0.12", default-features = false, features = [ "cookies", "rustls-tls", + "http2", ] } tower = { version = "0.4", features = ["buffer", "util"] } tonic = { version = "0.14" } diff --git a/packages/zaino-fetch/src/jsonrpsee/connector.rs b/packages/zaino-fetch/src/jsonrpsee/connector.rs index 5175d03b1..765cace5a 100644 --- a/packages/zaino-fetch/src/jsonrpsee/connector.rs +++ b/packages/zaino-fetch/src/jsonrpsee/connector.rs @@ -205,6 +205,14 @@ impl JsonRpSeeConnector { .connect_timeout(Duration::from_secs(2)) .timeout(Duration::from_secs(5)) .redirect(reqwest::redirect::Policy::none()) + // HTTP/2 cleartext (h2c) with prior knowledge: zebra's RPC endpoint speaks HTTP/2, so a + // single TCP connection multiplexes every in-flight request. This removes the HTTP/1.1 + // connection-per-request churn and the stale-keep-alive reap race (the "error sending + // request" / "error decoding response body" class) that the concurrent bulk-sync fetch + // pipeline exposed, and lets fetch concurrency scale on one connection. + // NOTE: prior knowledge has no HTTP/1.1 fallback — if the endpoint does not accept h2c, + // every request fails. + .http2_prior_knowledge() .build() .map_err(TransportError::ReqwestError)?; @@ -225,6 +233,8 @@ impl JsonRpSeeConnector { .timeout(Duration::from_secs(5)) .redirect(reqwest::redirect::Policy::none()) .cookie_store(true) + // See `new_with_basic_auth` for why HTTP/2 prior knowledge (h2c) is used. + .http2_prior_knowledge() .build() .map_err(TransportError::ReqwestError)?; @@ -310,10 +320,22 @@ impl JsonRpSeeConnector { .build_request(method, ¶ms, id) .map_err(RpcRequestError::JsonRpc)?; - let response = request_builder - .send() - .await - .map_err(|e| RpcRequestError::Transport(TransportError::ReqwestError(e)))?; + let response = match request_builder.send().await { + Ok(response) => response, + Err(e) => { + // A `send` failure is a connection-level error: the request never reached the + // handler (connection refused / reset, a stale pooled connection, or an h2 + // GOAWAY when zebra restarts), so retrying is safe for any method, idempotent or + // not. Bounded by `max_attempts` so a genuine outage still surfaces to the caller + // (and the sync loop's give-up ladder). With HTTP/2 prior knowledge this class is + // rare; the retry is defence-in-depth for restarts and transient blips. + if attempts >= max_attempts { + return Err(RpcRequestError::Transport(TransportError::ReqwestError(e))); + } + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + continue; + } + }; let status = response.status(); diff --git a/packages/zaino-state/src/chain_index.rs b/packages/zaino-state/src/chain_index.rs index 9e4ef163f..0b2ae0d9f 100644 --- a/packages/zaino-state/src/chain_index.rs +++ b/packages/zaino-state/src/chain_index.rs @@ -834,6 +834,13 @@ impl NodeBackedChainIndex { let mut change_rx = source.subscribe_to_blocks_received(); let mut consecutive_failures: u32 = 0; let mut current_backoff = timings.initial_backoff; + // Highest finalised tip observed so far. Used to tell a *stuck* sync loop (the + // give-up condition) apart from one that committed batches and then hit a transient + // transport error: a long `sync_to_height` can persist tens of thousands of blocks and + // still return `Err` on a single dropped connection, which is forward progress, not a + // stall. `None` until the db has a tip. + let mut last_committed_tip: Option = + fs.db_height().await.ok().flatten().map(|h| h.0); loop { let source = source.clone(); @@ -915,6 +922,13 @@ impl NodeBackedChainIndex { Ok(()) => { consecutive_failures = 0; current_backoff = timings.initial_backoff; + last_committed_tip = fs + .db_height() + .await + .ok() + .flatten() + .map(|h| h.0) + .or(last_committed_tip); status.store(StatusType::Ready); // Race the post-success wait against cancellation // and a source-change notification. `shutdown()`'s @@ -934,20 +948,48 @@ impl NodeBackedChainIndex { } } Err(e) => { - consecutive_failures += 1; - if consecutive_failures >= timings.max_consecutive_failures { - tracing::error!( - "Sync loop failed {consecutive_failures} consecutive times, \ - giving up: {e:?}" + // Distinguish a stuck loop from one that committed batches and then hit a + // transient transport error. The give-up counter exists to detect a loop + // making *no* progress; a `sync_to_height` that persisted blocks before the + // error advanced the finalised tip, so it must not count toward the cap. + // Without this, a fast genesis sync over flaky HTTP/1.1 keep-alive + // connections marches the counter to `max_consecutive_failures` and crashes + // while healthily syncing. + let current_tip = fs.db_height().await.ok().flatten().map(|h| h.0); + let made_progress = match (current_tip, last_committed_tip) { + (Some(current), Some(last)) => current > last, + (Some(_), None) => true, + (None, _) => false, + }; + last_committed_tip = current_tip.or(last_committed_tip); + + if made_progress { + // Real forward progress despite the error: reset the counter and the + // backoff ladder, then back off briefly to let the connection layer + // recover before resuming from the new tip. + consecutive_failures = 0; + current_backoff = timings.initial_backoff; + tracing::warn!( + "Sync loop iteration failed but committed progress \ + (finalised tip now {current_tip:?}); resetting failure counter, \ + retrying in {current_backoff:?}: {e:?}" + ); + } else { + consecutive_failures += 1; + if consecutive_failures >= timings.max_consecutive_failures { + tracing::error!( + "Sync loop failed {consecutive_failures} consecutive times \ + without progress, giving up: {e:?}" + ); + status.store(StatusType::CriticalError); + return Err(e); + } + tracing::warn!( + "Sync loop iteration failed ({consecutive_failures}/{}), \ + retrying in {current_backoff:?}: {e:?}", + timings.max_consecutive_failures ); - status.store(StatusType::CriticalError); - return Err(e); } - tracing::warn!( - "Sync loop iteration failed ({consecutive_failures}/{}), \ - retrying in {current_backoff:?}: {e:?}", - timings.max_consecutive_failures - ); status.store(StatusType::RecoverableError); // Race the failure-path backoff sleep against // cancellation. Without this, `shutdown()` after @@ -959,7 +1001,11 @@ impl NodeBackedChainIndex { _ = cancel_token.cancelled() => return Ok(()), _ = tokio::time::sleep(current_backoff) => {} } - current_backoff = (current_backoff * 2).min(timings.max_backoff); + // Only escalate the backoff when we made no progress; a progressing loop + // already reset it to `initial_backoff` above. + if !made_progress { + current_backoff = (current_backoff * 2).min(timings.max_backoff); + } } } } From 21437f755bb218afde286a82497510264727e77d Mon Sep 17 00:00:00 2001 From: Blake Emerson Date: Wed, 17 Jun 2026 09:18:10 -0500 Subject: [PATCH 3/3] fix(zaino-fetch): enable HTTP/2 adaptive flow-control window and raise RPC timeout to 30s --- .../zaino-fetch/src/jsonrpsee/connector.rs | 24 ++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/packages/zaino-fetch/src/jsonrpsee/connector.rs b/packages/zaino-fetch/src/jsonrpsee/connector.rs index 765cace5a..72b54d513 100644 --- a/packages/zaino-fetch/src/jsonrpsee/connector.rs +++ b/packages/zaino-fetch/src/jsonrpsee/connector.rs @@ -46,6 +46,15 @@ use crate::jsonrpsee::{ use super::response::{GetDifficultyResponse, GetNetworkSolPsResponse}; +/// Per-request timeout (total: connect excluded, send + full body read). +/// +/// Sized for the bulk-sync worst case: many large sandblast-era blocks (~hundreds of KiB each) +/// fetched concurrently and multiplexed over the single h2c connection. A short timeout +/// guillotines a legitimately large transfer mid-body under that load and surfaces as +/// "error decoding response body"; the sync loop's retry/give-up ladder remains the backstop for a +/// genuinely unresponsive validator. +const HTTP_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); + #[derive(Serialize, Deserialize, Debug)] struct RpcRequest { jsonrpc: String, @@ -203,7 +212,7 @@ impl JsonRpSeeConnector { ) -> Result { let client = ClientBuilder::new() .connect_timeout(Duration::from_secs(2)) - .timeout(Duration::from_secs(5)) + .timeout(HTTP_REQUEST_TIMEOUT) .redirect(reqwest::redirect::Policy::none()) // HTTP/2 cleartext (h2c) with prior knowledge: zebra's RPC endpoint speaks HTTP/2, so a // single TCP connection multiplexes every in-flight request. This removes the HTTP/1.1 @@ -213,6 +222,13 @@ impl JsonRpSeeConnector { // NOTE: prior knowledge has no HTTP/1.1 fallback — if the endpoint does not accept h2c, // every request fails. .http2_prior_knowledge() + // Let the HTTP/2 stack size its connection/stream flow-control windows to the + // bandwidth-delay product. With the default fixed 64 KiB connection window, N + // concurrent large-block fetches (sandblast blocks are ~hundreds of KiB) multiplexed + // over the one h2c connection starve on flow control, stall, and trip the request + // timeout mid-body — surfacing as "error decoding response body". Adaptive sizing keeps + // many large responses in flight at once. + .http2_adaptive_window(true) .build() .map_err(TransportError::ReqwestError)?; @@ -230,11 +246,13 @@ impl JsonRpSeeConnector { let client = ClientBuilder::new() .connect_timeout(Duration::from_secs(2)) - .timeout(Duration::from_secs(5)) + .timeout(HTTP_REQUEST_TIMEOUT) .redirect(reqwest::redirect::Policy::none()) .cookie_store(true) - // See `new_with_basic_auth` for why HTTP/2 prior knowledge (h2c) is used. + // See `new_with_basic_auth` for why HTTP/2 prior knowledge + adaptive flow-control + // window are used. .http2_prior_knowledge() + .http2_adaptive_window(true) .build() .map_err(TransportError::ReqwestError)?;