perf(zaino-state): Concurrent block fetches

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" }

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

packages/zaino-common/src/config/storage.rs

@@ -80,19 +80,38 @@ 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.
 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(),
         }
     }
 }

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<T> {
     jsonrpc: String,
@@ -203,8 +212,23 @@ impl JsonRpSeeConnector {
     ) -> Result<Self, TransportError> {
         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
+            // 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()
+            // 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)?;
 
@@ -222,9 +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 + adaptive flow-control
+            // window are used.
+            .http2_prior_knowledge()
+            .http2_adaptive_window(true)
             .build()
             .map_err(TransportError::ReqwestError)?;
 
@@ -310,10 +338,22 @@ impl JsonRpSeeConnector {
                 .build_request(method, &params, 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();
 

packages/zaino-state/src/chain_index.rs

@@ -834,6 +834,13 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
             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<u32> =
+                fs.db_height().await.ok().flatten().map(|h| h.0);
 
             loop {
                 let source = source.clone();
@@ -915,6 +922,13 @@ impl<Source: BlockchainSource> NodeBackedChainIndex<Source> {
                     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<Source: BlockchainSource> NodeBackedChainIndex<Source> {
                         }
                     }
                     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<Source: BlockchainSource> NodeBackedChainIndex<Source> {
                             _ = 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);
+                        }
                     }
                 }
             }