diff --git a/AGENTS.md b/AGENTS.md index 783cff82..a16d906b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -124,7 +124,10 @@ Tag requests (`latest`/`finalized`/`safe`) bind number → hash in their single Below the response cache, a bounded in-memory `BlockData` cache keyed by block hash (`--block-data-cache-max-size`, default 1GB, 0 disables) fronts the DB and RPC tiers; block-number lookups resolve number → hash before touching it, so canonicality is never cached and it needs no reorg invalidation. The cache pins 4 shards (largest cacheable entry = `max_bytes / 4` on any host), counts non-retained inserts, and drops an entry when a trace fails for a data-attributable reason (`TraceError::Data` — bad witness) while request-attributable failures (invalid tracer configs) never evict. In local cache mode with a `--witness-generator-endpoint` plus at least one fallback `--witness-endpoint`, request-serving witness fetches route by block age: blocks at least `--witness-local-window` blocks below the local tip skip the generator (which prunes beyond its `BACKUP` window) and fetch from the fallbacks; without the generator flag, witness endpoints are plain failover and routing is disabled. -With the `--r2-*` flag group (endpoint, bucket, access key id, secret), historical witness fetches try a direct SigV4-signed R2 GET (light decode, capped at half the remaining witness budget) before the RPC chain, falling back on any failure; frontier blocks keep the generator path, and `--r2-max-concurrent-requests` caps R2 GETs separately from the RPC witness semaphore. +With the `--r2-*` flag group (endpoint, bucket, access key id, secret), every request-serving witness fetch tries a direct SigV4-signed R2 GET (light decode, capped at half the remaining witness budget) before the RPC chain, falling back on any failure; `--r2-max-concurrent-requests` caps R2 GETs separately from the RPC witness semaphore. +The bucket is the same store the public gateway reads and can lead the generator at the frontier (uploader and generator RPC server publish from different files), so frontier hits are real; the frontier band is a small near-tip window (`R2_FRONTIER_WINDOW`, 32 blocks of uploader-lag grace on either side of the local tip — deliberately far narrower than the 4096-block routing window, so a stale catching-up tip cannot silence holes above it), hits there are labeled `witness_r2_frontier` (vs `witness_r2` past the band), the speculative frontier probe runs on an eighth of the remaining stage (vs half for blocks R2 must hold, so degraded R2 cannot burn half of every near-tip request's budget), and a `missing` classifies by band: in-band is the expected probe-ahead outcome (excluded from the alarm), below-band feeds `debug_trace_r2_witness_errors_total{kind="missing"}` (the bucket-integrity alarm, still covering recent-but-below-tip holes), and above-band — only reachable behind a stale catching-up tip — lands on its own `kind="missing_above_tip"` series, visible without flooding the alarm on every catch-up. +Any witness-chain RPC attempt under a deadline is capped at the tightest of three bounds — half the full witness stage (`RpcClientConfig::witness_per_attempt_timeout`, derived from `--witness-timeout`), the global `--rpc-per-attempt-timeout-ms` (an explicitly stricter operator setting is honored, never loosened), and — only while the round still has an untried provider to rotate to — half of what the call still has as the attempt starts (recomputed after any concurrency-permit wait, so neither an old-block-clamped stage, a post-R2 remainder, nor a long permit queue defeats the reserve). +The round's last hop, and every hop of a single-provider chain, takes the remainder whole under the ceiling instead: rotation stays protected without structurally condemning a slow-but-honest transfer, and the witness decode runs outside the attempt window (bounded by the deadline alone), so CPU-bound decode neither burns the reserve nor reads as a provider stall while a corrupt payload still rotates as the provider's error; deadline-less chain-sync fetches keep the general 20s cap so a slower-than-cap transfer still completes. When a logical upstream call gives up on its deadline it logs one WARN naming the `phase` it died in (`before_attempt` / `permit_wait_clamped` / `attempt_clamped` / `before_backoff`) with `provider` / `round` / `permit_wait_ms` / `attempt_ms`, and the abandoned attempt is recorded as `outcome="deadline_clamped"` rather than dropped; best-effort internal probes (the throttled upstream tip seed) demote that give-up log to debug while the deadline metric still fires, so a probe whose failure is already degraded cannot page as a user-visible incident. Permit wait is timed separately (`debug_trace_upstream_permit_wait_seconds{method}`) and the acquire is clamped to the deadline (phase `permit_wait_clamped`, cut-short wait still sampled), so queueing behind our own `--witness-max-concurrent-requests` stays distinguishable from endpoint slowness and a saturated queue cannot block a call past its budget unobserved. The background chain-sync prefetch routes by freshness against the last observed remote head: frontier-fresh blocks give the generator a short exclusive grace (its "witness not found" means "not generated yet" — fallbacks are fed by the same pipeline and cannot be ahead) before falling back to the full endpoint chain, while deep catch-up blocks — and any block classified against a stale head observation (older than the grace, as during a long catch-up stretch when the tip is not re-polled) — use the full chain from the first attempt. @@ -147,7 +150,7 @@ The background chain-sync prefetch routes by freshness against the last observed | `bin/debug-trace-server/src/rpc_middleware.rs` | Concurrent execution of inbound JSON-RPC batch entries | | `bin/debug-trace-server/src/data_provider.rs` | Block data fetching with single-flight coalescing | | `bin/debug-trace-server/src/block_data_cache.rs` | Bounded in-memory `BlockData` cache keyed by block hash | -| `bin/debug-trace-server/src/r2_witness.rs` | Direct-from-R2 historical witness source (light decode, deadline-aware) | +| `bin/debug-trace-server/src/r2_witness.rs` | Direct-from-R2 witness source (light decode, deadline-aware) | | `bin/debug-trace-server/src/server_db.rs` | Defines + implements the bin-local `BlockStore` trait (backed by `stateless-db`) | ## Test Organization diff --git a/README.md b/README.md index 2579077c..0f659009 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ The workspace contains two binaries and five library crates: | `stateless-db` | `crates/stateless-db` | redb-backed persistence: table definitions, read/write helpers, bounded `ContractCache` | | `stateless-common` | `crates/stateless-common` | Shared utilities: RPC client, logging, metrics | | `stateless-test-utils` | `crates/stateless-test-utils` | Test fixtures (blocks, witnesses, contracts) and env-var lock for integration tests | -| `stateless-r2` | `crates/stateless-r2` | Shared R2 (S3) witness primitives: SigV4 signer, object-key layout, endpoint parsing, signed PUT/GET with retry; consumed by mega-reth's witness uploaders (write), this repo's validator, and the trace server's historical witness source (read) | +| `stateless-r2` | `crates/stateless-r2` | Shared R2 (S3) witness primitives: SigV4 signer, object-key layout, endpoint parsing, signed PUT/GET with retry; consumed by mega-reth's witness uploaders (write), this repo's validator, and the trace server's witness source (read) | | `stateless-validator` | `bin/stateless-validator` | Main binary: chain sync, parallel validation workers | | `debug-trace-server` | `bin/debug-trace-server` | Standalone RPC server for debug/trace methods | @@ -151,10 +151,14 @@ The head observation is trusted as a freshness anchor only while itself recent ( Deep catch-up blocks (far below the observed head, where the generator may have pruned the witness) keep full failover from the first attempt. Without `--witness-generator-endpoint`, historical routing is disabled and the endpoints are plain failover. -**Direct-from-R2 historical witnesses:** -With `--r2-endpoint`, `--r2-bucket`, `--r2-access-key-id`, and `--r2-secret-access-key` (all four together), request-serving witness fetches for historical blocks try a SigV4-signed GET against the bucket before the RPC witness chain. +**Direct-from-R2 witnesses:** +With `--r2-endpoint`, `--r2-bucket`, `--r2-access-key-id`, and `--r2-secret-access-key` (all four together), every request-serving witness fetch tries a SigV4-signed GET against the bucket before the RPC witness chain. Object storage tolerates far higher parallelism than a shared RPC gateway and the bucket holds full history, so bulk backfill traffic stops competing with everything else on the public endpoint; any R2 failure (missing object, throttle, transport, corrupt payload — counted in `debug_trace_r2_witness_errors_total{kind}`) falls back to the RPC chain on the remaining witness budget, and the R2 attempt is capped at half that budget so a hung endpoint can never starve the fallback. -Frontier blocks keep the generator path (the bucket receives objects only after the uploader PUTs them), and the route needs a local DB (`--data-dir`) to anchor block age. +Frontier probes usually miss — the uploader typically lags the generator — and cost one fast 404; the frontier band is a small near-tip window (32 blocks of uploader-lag grace on either side of the local tip — far narrower than the 4096-block routing window, and a stale, catching-up tip cannot silence holes above it), hits there are labeled `witness_r2_frontier` (vs `witness_r2` past the band) so their hit rate stays separable, and the speculative probe runs on an eighth of the remaining stage (vs half for blocks R2 must hold), so degraded R2 cannot burn half of every near-tip request's budget before the RPC chain runs. +A `missing` classifies by band: in-band is expected probe-ahead (excluded from `debug_trace_r2_witness_errors_total{kind="missing"}`), below-band feeds that bucket-integrity alarm (the object must exist there), and above-band — only reachable behind a stale catching-up tip — lands on its own `kind="missing_above_tip"` series so catch-up windows stay visible without flooding the alarm. +The bucket is the same store the public gateway serves witnesses from, so at the frontier it can lead the generator (whose RPC server publishes from a different file than the uploader reads); the route needs a local DB (`--data-dir`) to anchor block age. +Any single witness-chain RPC attempt under a deadline is additionally capped at the tightest of: half the witness stage budget, the global per-attempt timeout, and — only while the round still has an untried provider to rotate to — half of what the call still has as the attempt starts (recomputed after any permit wait). +The round's last hop, and every hop of a single-provider chain, takes the remainder whole under the ceiling instead, so a stalled endpoint (or a saturated concurrency permit — waits are deadline-bounded too) can never consume the stage while a rotation is still worth reserving for, and a slow-but-honest transfer is never structurally condemned; the witness decode runs outside the attempt window, bounded by the request deadline alone. `--r2-max-concurrent-requests` caps in-flight GETs separately from `--witness-max-concurrent-requests` — the RPC cap sizes a shared gateway, R2 tolerates far more. **Witness routing and sync knobs** (each also settable via its `DEBUG_TRACE_SERVER_*` env var): diff --git a/bin/debug-trace-server/src/data_provider.rs b/bin/debug-trace-server/src/data_provider.rs index 0043544a..4fbd3f8e 100644 --- a/bin/debug-trace-server/src/data_provider.rs +++ b/bin/debug-trace-server/src/data_provider.rs @@ -6,11 +6,12 @@ //! 1. **Local Database** (fast) - Local DB for pre-fetched blocks (if configured) //! 2. **Remote RPC** (slower) - Upstream RPC endpoints as fallback //! -//! Within the RPC fallback, the witness stage routes by block age (see [`WitnessFetchConfig`]): -//! blocks fewer than `local_window` blocks below the local tip use the full witness endpoint -//! chain (internal generator first), while historical blocks — at least `local_window` below, -//! which the generator has long pruned — skip the generator and go straight to the remaining -//! endpoints. +//! Within the RPC fallback, the witness stage first probes a configured R2 bucket for every +//! fetch (see [`fetch_witness`]), then routes the RPC chain by block age (see +//! [`WitnessFetchConfig`]): blocks fewer than `local_window` blocks below the local tip use +//! the full witness endpoint chain (internal generator first), while historical blocks — at +//! least `local_window` below, which the generator has long pruned — skip the generator and +//! go straight to the remaining endpoints. //! //! # Features //! - **Single-flight request coalescing**: concurrent callers for the same block hash share one @@ -126,10 +127,19 @@ pub struct BlockData { /// normal. pub const DEFAULT_WITNESS_TIMEOUT_SECS: u64 = 8; -/// The R2 attempt's share of the remaining witness budget: half, so a hung R2 endpoint can -/// never starve the RPC fallback of its turn; a healthy R2 answers in a fraction of it. +/// The R2 attempt's share of the remaining witness budget for blocks R2 should hold +/// (at or below the frontier band): half, so a hung R2 endpoint can never starve the RPC +/// fallback of its turn; a healthy R2 answers in a fraction of it. const R2_WITNESS_BUDGET_DIVISOR: u32 = 2; +/// The frontier probe's much smaller share: an eighth of the remaining budget. The probe +/// is speculative — the uploader usually lags the generator and a miss is the expected +/// outcome — so degraded R2 (timing-out connects, throttling, each retried) must not be +/// able to burn half the witness stage in front of the RPC chain on every near-tip +/// request. An eighth still leaves comfortable headroom over a healthy single GET in +/// every deployed region while bounding a degraded R2's damage to a sliver of the stage. +const R2_FRONTIER_BUDGET_DIVISOR: u32 = 8; + /// Default local-tip window (in blocks): witnesses at least this far below the local tip are /// historical and skip the internal generator endpoint (see [`witness_route`]). /// @@ -138,6 +148,16 @@ const R2_WITNESS_BUDGET_DIVISOR: u32 = 2; /// so probing it first only burns a failover round trip. pub const DEFAULT_WITNESS_LOCAL_WINDOW: u64 = 4096; +/// Near-tip band (in blocks) inside which an R2 witness `missing` is the expected +/// probe-ahead outcome — the uploader may plausibly not have PUT the object yet — rather +/// than a bucket hole. Sized to comfortably cover the uploader's PUT latency plus the local +/// DB tip's own sync lag (a few seconds each; chain sync's `GENERATOR_WITNESS_GRACE` is the +/// time-based analog), and kept far below [`DEFAULT_WITNESS_LOCAL_WINDOW`]: routing asks +/// "may the generator have pruned this?", this asks "may the uploader not have reached it +/// yet?", and gating the `kind="missing"` alarm on the routing window would silence +/// bucket-integrity alerting across its whole 4096-block span. +const R2_FRONTIER_WINDOW: u64 = 32; + /// Default deadline for the full block-fetch pipeline (header + witness + block + contracts) /// in seconds (13 seconds). /// @@ -337,9 +357,8 @@ pub(crate) struct DataProvider { /// The full-call deadline still dominates; the per-stage budgets cap how much of it the /// witness fetch can burn. witness_cfg: WitnessFetchConfig, - /// Optional direct-from-R2 source for historical witnesses. When present, the witness - /// stage tries it before the RPC chain for blocks the routing window classifies as - /// historical; every R2 failure falls back to the RPC chain on the remaining deadline. + /// Optional direct-from-R2 witness source. When present, every witness fetch tries it + /// before the RPC chain; any R2 failure falls back on the remaining deadline. r2_witness: Option>, /// Wall-clock budget for one user-facing block-data call, from entry through /// header + witness + block + contract resolution. The retry loop in `RpcClient` @@ -429,8 +448,7 @@ impl DataProvider { /// + block + contracts) /// * `canonical_hash_memo_capacity` - Entry cap for the in-memory canonical-hash memo /// - /// The optional direct-from-R2 historical witness source attaches via - /// [`Self::with_r2_witness`]. + /// The optional direct-from-R2 witness source attaches via [`Self::with_r2_witness`]. pub fn new( rpc_client: Arc, db: Option>, @@ -1190,6 +1208,40 @@ fn is_historical(db_tip: Option, block_number: u64, local_window: u64) -> b } } +/// Which band a block falls in for the R2 probe, deciding its metrics label, its budget +/// share, and how a `missing` is classified. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum R2Band { + /// Within [`R2_FRONTIER_WINDOW`] of the local tip on either side (or no tip yet — the + /// cold-start transient `validate_args`' `--data-dir` requirement bounds): the + /// uploader may plausibly not have PUT the object yet, so a `missing` is the expected + /// probe-ahead outcome and the speculative probe gets only the + /// [`R2_FRONTIER_BUDGET_DIVISOR`] budget share. + Frontier, + /// More than the band *above* the local tip — only reachable when chain sync is + /// behind, since a healthy tip tracks the real head and blocks past it do not resolve. + /// The bucket's state is unknowable from a stale tip, so a `missing` records on its + /// own [`crate::r2_witness::KIND_MISSING_ABOVE_TIP`] series: visible (a real hole in + /// the catch-up gap still surfaces there) without flooding the below-band + /// bucket-integrity alarm with routine uploader lag on every catch-up. Like the + /// frontier, the probe is speculative — the object is not guaranteed to exist yet — + /// and gets only the [`R2_FRONTIER_BUDGET_DIVISOR`] budget share. + AboveTip, + /// At least the band *below* the tip: the object must exist, so a `missing` is a + /// bucket hole and feeds the `kind="missing"` bucket-integrity alarm. + Historical, +} + +/// Classifies `block_number` against the local tip; see [`R2Band`] for the semantics. +fn r2_band(db_tip: Option, block_number: u64) -> R2Band { + match db_tip { + None => R2Band::Frontier, + Some(tip) if block_number > tip.saturating_add(R2_FRONTIER_WINDOW) => R2Band::AboveTip, + Some(_) if is_historical(db_tip, block_number, R2_FRONTIER_WINDOW) => R2Band::Historical, + Some(_) => R2Band::Frontier, + } +} + /// Witness route for a block: how many leading witness endpoints to skip, plus the metrics /// source label. Historical blocks skip the internal generator at index 0 — but only with a /// fallback endpoint to skip to (`can_skip_generator`, so the skip-aware fetch never sees an @@ -1207,19 +1259,25 @@ fn witness_route( } } -/// Fetches witness data, routing by block age. The `deadline` is the witness stage's -/// effective deadline (see [`witness_deadline_for`]). +/// Fetches witness data. The `deadline` is the witness stage's effective deadline (see +/// [`witness_deadline_for`]). /// +/// **Every fetch tries R2 first** when a source is configured. The bucket is the very store +/// the public gateway serves witnesses from, reached by a client that is bounded where the +/// gateway hop is not (connect timeout, capped attempts, [`R2_WITNESS_BUDGET_DIVISOR`]'s +/// half-budget share) and tolerant of far more parallelism than the shared gateway. At the +/// frontier the bucket can even lead the generator: the uploader and the generator's RPC +/// server publish from *different files* of the same generation run, so "the generator has +/// no file yet" says nothing about the bucket. A frontier probe usually misses — one fast 404 — and +/// every R2 failure falls back to the RPC chain below on the remaining deadline. +/// +/// The RPC chain then routes by block age: /// - **Recent block** (fewer than `local_window` blocks below the local tip, or tip unknown): the -/// full RPC witness endpoint chain, tried in order — the internal generator first, so near-tip -/// witnesses stay on the fast internal path. -/// - **Historical block** with an R2 source configured: R2 first — object storage tolerates far -/// more parallelism than the shared RPC gateway, and the bucket holds full history while the -/// generator prunes beyond its window. Every R2 failure (missing object, throttle-exhausted, -/// transport, corrupt payload) falls back to the RPC chain below on the remaining deadline. -/// - **Historical block** on the RPC chain, with a declared generator and a fallback endpoint -/// configured: the same chain minus the generator, the guaranteed-miss probe -/// [`DEFAULT_WITNESS_LOCAL_WINDOW`] describes. +/// full chain in order — the internal generator first, so near-tip witnesses stay on the fast +/// internal path. +/// - **Historical block**, with a declared generator and a fallback endpoint configured: the same +/// chain minus the generator, the guaranteed-miss probe [`DEFAULT_WITNESS_LOCAL_WINDOW`] +/// describes. /// /// Uses the zero-validation light decode: the trace server never verifies the witness proof, /// so the full decode's per-point elliptic-curve work bought nothing. The recorded size is @@ -1233,11 +1291,11 @@ async fn fetch_witness( block_hash: B256, deadline: Instant, ) -> DataProviderResult<(LightWitness, MptWitness)> { - if let Some(r2) = r2_witness && - is_historical(db_tip, block_number, cfg.local_window) && - let Some(witness) = try_r2_witness(r2, block_number, block_hash, deadline).await - { - return Ok(witness); + if let Some(r2) = r2_witness { + let band = r2_band(db_tip, block_number); + if let Some(witness) = try_r2_witness(r2, band, block_number, block_hash, deadline).await { + return Ok(witness); + } } let can_skip_generator = cfg.generator_first && rpc_client.witness_provider_count() >= 2; @@ -1273,32 +1331,68 @@ async fn fetch_witness( } } -/// One R2 attempt for a historical witness, on [`R2_WITNESS_BUDGET_DIVISOR`]'s share of the -/// remaining budget. `None` on any failure — the caller falls back to the RPC chain. +/// One R2 attempt for a witness, on its band's share of the remaining budget +/// ([`R2_FRONTIER_BUDGET_DIVISOR`] for the speculative frontier probe, +/// [`R2_WITNESS_BUDGET_DIVISOR`] otherwise). `None` on any failure — the caller falls +/// back to the RPC chain. +/// +/// `band` ([`r2_band`]) also selects the metrics source label (`witness_r2_frontier` +/// inside the band vs `witness_r2` outside) and the `missing` classification: in-band, a +/// miss is the expected speculative-probe outcome, kept separable so its dominant miss +/// rate does not read as R2 health degrading; below the band the object must exist and a +/// miss feeds the `kind="missing"` bucket-integrity alarm; above the band (stale tip) it +/// lands on its own `missing_above_tip` series. async fn try_r2_witness( r2: &R2WitnessSource, + band: R2Band, block_number: u64, block_hash: B256, deadline: Instant, ) -> Option<(LightWitness, MptWitness)> { + let source = if band == R2Band::Frontier { "witness_r2_frontier" } else { "witness_r2" }; + // Only the historical band gets the half share: there R2 is the primary source and + // the object must exist. Both near-tip bands are speculative — in-band the uploader + // may lag, above-band (a stale local tip: a deliberate tip buffer, or a catch-up) the + // object is not guaranteed to exist yet — so neither may burn half of a near-head + // request's budget on degraded R2. + let divisor = if band == R2Band::Historical { + R2_WITNESS_BUDGET_DIVISOR + } else { + R2_FRONTIER_BUDGET_DIVISOR + }; let now = Instant::now(); - let r2_deadline = now + deadline.saturating_duration_since(now) / R2_WITNESS_BUDGET_DIVISOR; - let metrics = WitnessSourceMetrics::new_for_source("witness_r2"); + let r2_deadline = now + deadline.saturating_duration_since(now) / divisor; + let metrics = WitnessSourceMetrics::new_for_source(source); match r2.get_witness_light(block_number, block_hash, r2_deadline).await { Ok(witness) => { - record_witness_success(&metrics, "witness_r2", now, &witness); + record_witness_success(&metrics, source, now, &witness); Some(witness) } Err(e) => { metrics.record_request(false, now.elapsed().as_secs_f64()); - crate::metrics::record_r2_witness_error(e.kind()); - warn!( - block_number, - block_hash = %block_hash, - kind = e.kind(), - error = %e, - "R2 witness fetch failed, falling back to the RPC chain", - ); + if band == R2Band::Frontier && e.is_missing() { + // The per-source counter above still records the miss (what the frontier + // hit rate reads); only the `kind="missing"` alarm skips it. + debug!( + block_number, + block_hash = %block_hash, + "Frontier witness not in R2 yet; trying the RPC chain", + ); + } else { + let kind = if band == R2Band::AboveTip && e.is_missing() { + crate::r2_witness::KIND_MISSING_ABOVE_TIP + } else { + e.kind() + }; + crate::metrics::record_r2_witness_error(kind); + warn!( + block_number, + block_hash = %block_hash, + kind, + error = %e, + "R2 witness fetch failed, falling back to the RPC chain", + ); + } None } } @@ -1534,6 +1628,26 @@ pub(crate) mod test_support { .unwrap(); serve(module).await } + + /// Serves `wire` after `delay` on every call — a healthy-but-slow witness endpoint. + pub(crate) async fn delayed_witness_rpc( + delay: Duration, + wire: String, + ) -> (ServerHandle, String, Arc) { + let hits = Arc::new(AtomicUsize::new(0)); + let mut module = RpcModule::new((hits.clone(), wire)); + module + .register_async_method("mega_getBlockWitness", move |_p, ctx, _| async move { + ctx.0.fetch_add(1, Ordering::Relaxed); + tokio::time::sleep(delay).await; + Ok::<_, ErrorObjectOwned>(ctx.1.clone()) + }) + .unwrap(); + let server = + jsonrpsee::server::ServerBuilder::default().build("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", server.local_addr().unwrap()); + (server.start(module), url, hits) + } } #[cfg(test)] @@ -1584,6 +1698,38 @@ mod tests { assert!(is_historical(Some(100), 50, 0), "zero window: everything at/below tip"); } + /// The R2 frontier band is the uploader-lag grace, not the routing window: a block that + /// is recent for routing but past the band must count an R2 miss as a bucket hole (the + /// `kind="missing"` alarm), not an expected probe-ahead miss. + #[test] + fn r2_frontier_band_is_narrower_than_routing() { + use R2Band::*; + assert_eq!(r2_band(None, 100), Frontier, "unknown tip: nothing known to be uploaded"); + assert_eq!(r2_band(Some(5000), 5000), Frontier, "the tip itself"); + assert_eq!(r2_band(Some(5000), 5000 - R2_FRONTIER_WINDOW + 1), Frontier, "just inside"); + assert_eq!( + r2_band(Some(5000), 5000 - R2_FRONTIER_WINDOW), + Historical, + "just past the band" + ); + assert_eq!(r2_band(Some(5000), 5000 + R2_FRONTIER_WINDOW), Frontier, "just above, in band"); + // A stale, catching-up tip must not flood the bucket-integrity alarm for the gap + // above it — nor silence it: the gap gets its own missing_above_tip series. + assert_eq!( + r2_band(Some(5000), 5000 + R2_FRONTIER_WINDOW + 1), + AboveTip, + "far above a stale tip is unknown territory, not uploader lag", + ); + // The band a routing-window gate would have silenced: recent for routing, far past + // any plausible uploader lag. + let recent_not_tip = 4000; + assert_eq!(r2_band(Some(5000), recent_not_tip), Historical, "a hole here must alarm"); + assert!( + !is_historical(Some(5000), recent_not_tip, DEFAULT_WITNESS_LOCAL_WINDOW), + "yet the same block is recent for witness routing", + ); + } + /// Route selection: the skip-generator chain and the `witness_historical` label apply /// only when the generator may be skipped (declared generator + fallback) AND the block /// is historical; everything else stays on the full chain under the `witness_generator` @@ -1607,11 +1753,25 @@ mod tests { rpc_retry: BackoffPolicy::new(Duration::from_millis(1), Duration::from_millis(2)), ..RpcClientConfig::trace_server() }; + let (rpc_client, mut cfg) = cap_fixture(witness_urls, config); + cfg.generator_first = generator_first; + (rpc_client, cfg) + } + + /// [`routing_fixture`] with the client config under the caller's control — the hop-cap + /// tests vary it, and their budgets come from the explicit `deadline` they pass to + /// [`fetch_witness`]. `witness_urls[0]` doubles as the data endpoint and the generator + /// (first witness provider); `generator_first` stays on, matching the deployment shape + /// the cap protects. + fn cap_fixture( + witness_urls: &[&str], + config: RpcClientConfig, + ) -> (RpcClient, WitnessFetchConfig) { let rpc_client = RpcClient::new_with_config(&witness_urls[..1], witness_urls, config, None).unwrap(); let cfg = WitnessFetchConfig { local_window: 100, - generator_first, + generator_first: true, ..WitnessFetchConfig::with_defaults(1) }; (rpc_client, cfg) @@ -1980,6 +2140,14 @@ mod tests { hb.stop().unwrap(); } + /// Fixture witness encoded as the RPC wire string (`encode_witness_response`). + fn fixture_wire() -> String { + let (salt_witness, mpt_witness): (_, MptWitness) = + TestFixtures::mainnet_shared().first_paired_witness(); + stateless_common::encode_witness_response(&salt_witness, &mpt_witness) + .expect("fixture witness must encode") + } + /// Fixture witness encoded as an R2 object body (the uploader's wire format). fn fixture_r2_payload() -> Vec { let (salt_witness, mpt_witness): (_, MptWitness) = @@ -1989,23 +2157,35 @@ mod tests { .1 } - /// A historical block with R2 configured is served from R2 alone: neither the generator - /// nor the fallback RPC endpoint sees a request. + /// A block whose witness is in the bucket is served from R2 alone — historical and + /// frontier alike — with neither the generator nor the fallback touched. The frontier + /// case is why every fetch probes R2: the bucket can lead the generator at the tip + /// (see [`fetch_witness`]). #[tokio::test] - async fn fetch_witness_historical_prefers_r2_over_the_rpc_chain() { + async fn fetch_witness_serves_from_r2_for_historical_and_frontier() { let (r2_endpoint, r2_hits) = mock_r2(vec![(200, fixture_r2_payload())]).await; let (ha, url_a, hits_a) = scripted_witness_rpc(0, None).await; let (hb, url_b, hits_b) = scripted_witness_rpc(0, None).await; let (rpc_client, cfg) = routing_fixture(&[url_a.as_str(), url_b.as_str()], true); let r2 = crate::r2_witness::test_support::source(&r2_endpoint); - // Historical block (900 + 100 <= 5000). - let deadline = Instant::now() + Duration::from_secs(5); - let result = - fetch_witness(&rpc_client, &cfg, Some(&r2), Some(5000), 900, B256::ZERO, deadline) - .await; - assert!(result.is_ok(), "R2 must serve the historical witness"); - assert_eq!(r2_hits.load(Ordering::SeqCst), 1, "exactly one R2 GET"); + // Historical (900 + 100 <= 5000), then the tip itself — the most frontier a block + // gets; `mock_r2` repeats its last scripted response. + for block_number in [900, 5000] { + let deadline = Instant::now() + Duration::from_secs(5); + let result = fetch_witness( + &rpc_client, + &cfg, + Some(&r2), + Some(5000), + block_number, + B256::ZERO, + deadline, + ) + .await; + assert!(result.is_ok(), "R2 must serve block {block_number}: {:?}", result.err()); + } + assert_eq!(r2_hits.load(Ordering::SeqCst), 2, "exactly one R2 GET per fetch"); assert_eq!(hits_a.load(Ordering::Relaxed), 0, "generator must stay untouched"); assert_eq!(hits_b.load(Ordering::Relaxed), 0, "RPC fallback must stay untouched"); @@ -2044,12 +2224,8 @@ mod tests { #[tokio::test] async fn fetch_witness_hung_r2_leaves_budget_for_the_rpc_fallback() { let (r2_endpoint, _r2_peak) = mock_r2_held(200, Duration::from_secs(30)).await; - let (salt_witness, mpt_witness): (_, MptWitness) = - TestFixtures::mainnet_shared().first_paired_witness(); - let wire = stateless_common::encode_witness_response(&salt_witness, &mpt_witness) - .expect("fixture witness must encode"); let (ha, url_a, hits_a) = scripted_witness_rpc(0, None).await; - let (hb, url_b, hits_b) = scripted_witness_rpc(0, Some(wire)).await; + let (hb, url_b, hits_b) = scripted_witness_rpc(0, Some(fixture_wire())).await; let (rpc_client, cfg) = routing_fixture(&[url_a.as_str(), url_b.as_str()], true); let r2 = crate::r2_witness::test_support::source(&r2_endpoint); @@ -2085,27 +2261,297 @@ mod tests { hb.stop().unwrap(); } - /// Recent blocks never touch R2 (the bucket lags the generator at the frontier): the - /// full RPC chain with the generator first keeps serving them. + /// A frontier R2 miss — the expected case, the uploader usually lags the generator — + /// falls through to the RPC chain with the generator first: exactly the pre-R2 shape, + /// at the cost of one fast 404. #[tokio::test] - async fn fetch_witness_recent_block_ignores_r2() { - let (r2_endpoint, r2_hits) = mock_r2(vec![(200, "never fetched")]).await; - let (ha, url_a, hits_a) = scripted_witness_rpc(0, None).await; - let (hb, url_b, _hits_b) = scripted_witness_rpc(0, None).await; + async fn fetch_witness_frontier_r2_miss_falls_to_the_generator() { + let (r2_endpoint, r2_hits) = mock_r2(vec![(404, "NoSuchKey")]).await; + let (ha, url_a, hits_a) = scripted_witness_rpc(0, Some(fixture_wire())).await; + let (hb, url_b, hits_b) = scripted_witness_rpc(0, None).await; let (rpc_client, cfg) = routing_fixture(&[url_a.as_str(), url_b.as_str()], true); let r2 = crate::r2_witness::test_support::source(&r2_endpoint); - // The tip itself is recent: R2 must stay untouched, the generator probed first. - let deadline = Instant::now() + Duration::from_millis(150); - let _ = fetch_witness(&rpc_client, &cfg, Some(&r2), Some(5000), 5000, B256::ZERO, deadline) - .await; - assert_eq!(r2_hits.load(Ordering::SeqCst), 0, "a recent block must not touch R2"); - assert!(hits_a.load(Ordering::Relaxed) >= 1, "recent fetch must probe the generator"); + let deadline = Instant::now() + Duration::from_secs(5); + let result = + fetch_witness(&rpc_client, &cfg, Some(&r2), Some(5000), 5000, B256::ZERO, deadline) + .await; + assert!(result.is_ok(), "generator must serve after the R2 miss: {:?}", result.err()); + assert_eq!(r2_hits.load(Ordering::SeqCst), 1, "the frontier probe is a single GET"); + assert!(hits_a.load(Ordering::Relaxed) >= 1, "the generator follows the R2 miss"); + assert_eq!(hits_b.load(Ordering::Relaxed), 0, "no gateway when the generator serves"); ha.stop().unwrap(); hb.stop().unwrap(); } + /// A stalled endpoint must not consume the whole witness stage. Replays the incident + /// shape: the generator misses (witness not generated yet), the chain rotates to + /// a gateway that accepts TCP and never answers, and the witness appears at the + /// generator moments later. The per-hop cap (`witness_per_attempt_timeout`) cuts the + /// stalled hop so the loop reaches round 1, where the generator serves inside the + /// original deadline. Goes red without the cap: `min(per_attempt = 20s, remaining)` + /// resolves to `remaining`, the stalled hop eats the entire budget, and round 1 never + /// happens — the caller sees a deadline instead of a witness. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn stalled_hop_is_capped_so_the_next_round_can_serve() { + // Generator: "not generated yet" once, then serves (the file has been written). + let (ha, url_gen, hits_gen) = scripted_witness_rpc(1, Some(fixture_wire())).await; + // Gateway: accepts connections and never replies. + let url_gw = test_support::hanging_url(); + + let config = RpcClientConfig { + rpc_retry: BackoffPolicy::new(Duration::from_millis(20), Duration::from_millis(40)), + witness_per_attempt_timeout: Some(Duration::from_millis(300)), + ..RpcClientConfig::trace_server() + }; + let (rpc_client, cfg) = cap_fixture(&[url_gen.as_str(), url_gw.as_str()], config); + + // Frontier block: above the local tip, so the full chain (generator first) runs. + let budget = Duration::from_secs(3); + let started = Instant::now(); + let result = + fetch_witness(&rpc_client, &cfg, None, Some(5000), 5001, B256::ZERO, started + budget) + .await; + let elapsed = started.elapsed(); + + assert!(result.is_ok(), "round 1 must serve inside the budget: {:?}", result.err()); + assert!( + elapsed >= Duration::from_millis(300), + "the stalled hop must be waited on up to its cap first ({elapsed:?})" + ); + assert!(elapsed < budget, "the fetch must not ride the deadline ({elapsed:?})"); + assert!( + hits_gen.load(Ordering::Relaxed) >= 2, + "the generator must be asked again after the stalled hop was cut" + ); + + ha.stop().unwrap(); + } + + /// The hop cap must bind against what the stage actually has left, not the configured + /// full stage: an old-block clamp or a slow R2 pre-try can shrink the stage below the + /// static cap, and a cap that stops binding exactly there lets one stalled hop consume + /// the entire remainder. Here the stage is 1s while the configured cap is 6s — entry + /// halving must cut the stalled hop at ~0.5s so round 1 still fits. Goes red if the + /// effective cap is only `min(configured, remaining)`. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn hop_cap_tracks_the_shrunken_stage_budget() { + let (ha, url_gen, hits_gen) = scripted_witness_rpc(1, Some(fixture_wire())).await; + let url_gw = test_support::hanging_url(); + // A second fallback behind the stalled gateway: the halving reserve applies only + // to hops with an untried provider remaining, so the stall must not sit last. + let (hb, url_gw2, _hits_gw2) = scripted_witness_rpc(usize::MAX, None).await; + + let config = RpcClientConfig { + rpc_retry: BackoffPolicy::new(Duration::from_millis(20), Duration::from_millis(40)), + // Deliberately larger than the whole stage below: only the live-remaining + // halving can save this fetch. + witness_per_attempt_timeout: Some(Duration::from_secs(6)), + ..RpcClientConfig::trace_server() + }; + let (rpc_client, cfg) = + cap_fixture(&[url_gen.as_str(), url_gw.as_str(), url_gw2.as_str()], config); + + let budget = Duration::from_secs(1); + let started = Instant::now(); + let result = + fetch_witness(&rpc_client, &cfg, None, Some(5000), 5001, B256::ZERO, started + budget) + .await; + + assert!(result.is_ok(), "round 1 must fit inside the shrunken stage: {:?}", result.err()); + assert!(started.elapsed() < budget, "must not ride the deadline ({:?})", started.elapsed()); + assert!(hits_gen.load(Ordering::Relaxed) >= 2, "round 1 must reach the generator"); + + ha.stop().unwrap(); + hb.stop().unwrap(); + } + + /// The inter-round backoff may not consume everything the stage has left: held to half + /// the remainder, round 1 fits and the generator serves. Goes red against + /// `sleep.min(remaining)`, where the sleep wakes exactly at the deadline and forfeits + /// the retry it slept for. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn deadline_pressure_shortens_the_backoff_instead_of_sleeping_into_it() { + let (ha, url_gen, hits_gen) = scripted_witness_rpc(1, Some(fixture_wire())).await; + let url_gw = test_support::hanging_url(); + // A second fallback behind the stalled gateway keeps the stall mid-round, where + // the halving reserve applies (the round's last hop takes the remainder whole). + let (hb, url_gw2, _hits_gw2) = scripted_witness_rpc(usize::MAX, None).await; + + let config = RpcClientConfig { + // Production-shaped backoff: larger than what round 0 leaves of the stage. + rpc_retry: BackoffPolicy::new(Duration::from_millis(600), Duration::from_millis(800)), + witness_per_attempt_timeout: Some(Duration::from_secs(6)), + ..RpcClientConfig::trace_server() + }; + let (rpc_client, cfg) = + cap_fixture(&[url_gen.as_str(), url_gw.as_str(), url_gw2.as_str()], config); + + let budget = Duration::from_millis(800); + let started = Instant::now(); + let result = + fetch_witness(&rpc_client, &cfg, None, Some(5000), 5001, B256::ZERO, started + budget) + .await; + + assert!( + result.is_ok(), + "the shortened backoff must leave room for round 1: {:?}", + result.err() + ); + assert!(started.elapsed() < budget, "must not ride the deadline ({:?})", started.elapsed()); + assert!(hits_gen.load(Ordering::Relaxed) >= 2, "round 1 must reach the generator"); + + ha.stop().unwrap(); + hb.stop().unwrap(); + } + + /// A slow-but-honest single witness endpoint must still succeed: with no rotation to + /// reserve for, every hop takes the remainder whole (bounded by the ceiling), so a + /// transfer longer than half the stage completes exactly as it did before the hop cap + /// existed. Goes red under unconditional halving: the serve time never fits the + /// shrinking half-windows and a currently-succeeding fetch becomes a deterministic + /// client-visible timeout. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn slow_honest_single_provider_survives_the_reserve() { + let (ha, url, hits) = + test_support::delayed_witness_rpc(Duration::from_millis(600), fixture_wire()).await; + + let config = RpcClientConfig { + rpc_retry: BackoffPolicy::new(Duration::from_millis(20), Duration::from_millis(40)), + witness_per_attempt_timeout: Some(Duration::from_secs(6)), + ..RpcClientConfig::trace_server() + }; + let (rpc_client, cfg) = cap_fixture(&[url.as_str()], config); + + let budget = Duration::from_secs(1); + let started = Instant::now(); + let result = + fetch_witness(&rpc_client, &cfg, None, Some(5000), 5001, B256::ZERO, started + budget) + .await; + + assert!(result.is_ok(), "a 600ms serve must fit a 1s stage: {:?}", result.err()); + assert_eq!(hits.load(Ordering::Relaxed), 1, "one attempt, served whole"); + + ha.stop().unwrap(); + } + + /// The round's last hop takes the remainder whole: after the generator misses fast, + /// a fallback needing more than half of what is left must still be allowed to finish — + /// there is no further rotation to reserve for. Goes red under unconditional halving + /// (the serve is cut at the half-window and every retry gets less). + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn slow_honest_last_hop_takes_the_remainder() { + let (ha, url_gen, _hits_gen) = scripted_witness_rpc(usize::MAX, None).await; + let (hb, url_gw, hits_gw) = + test_support::delayed_witness_rpc(Duration::from_millis(600), fixture_wire()).await; + + let config = RpcClientConfig { + rpc_retry: BackoffPolicy::new(Duration::from_millis(20), Duration::from_millis(40)), + witness_per_attempt_timeout: Some(Duration::from_secs(6)), + ..RpcClientConfig::trace_server() + }; + let (rpc_client, cfg) = cap_fixture(&[url_gen.as_str(), url_gw.as_str()], config); + + let budget = Duration::from_secs(1); + let started = Instant::now(); + let result = + fetch_witness(&rpc_client, &cfg, None, Some(5000), 5001, B256::ZERO, started + budget) + .await; + + assert!(result.is_ok(), "the last hop must get the whole remainder: {:?}", result.err()); + assert_eq!(hits_gw.load(Ordering::Relaxed), 1, "served on the first gateway attempt"); + + ha.stop().unwrap(); + hb.stop().unwrap(); + } + + /// The frontier probe runs on the speculative eighth of the stage, not the historical + /// half: with R2 held past the frontier slice, the probe is abandoned early and the + /// generator serves with most of the stage intact. Goes red with one shared divisor — + /// the held probe then burns half the stage before the chain starts. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn frontier_probe_gets_only_the_speculative_budget_share() { + let (r2_endpoint, _r2_hits) = mock_r2_held(200, Duration::from_millis(700)).await; + let (ha, url_gen, hits_gen) = scripted_witness_rpc(0, Some(fixture_wire())).await; + let (rpc_client, cfg) = routing_fixture(&[url_gen.as_str()], true); + let r2 = crate::r2_witness::test_support::source(&r2_endpoint); + + // 1.6s stage: a speculative probe's slice is 200ms (an eighth), where the + // historical share would be 800ms. Both near-tip bands are speculative: the tip + // itself (in-band) and a block above a stale tip (above-band). + for block_number in [5000, 6000] { + let budget = Duration::from_millis(1600); + let started = Instant::now(); + let result = fetch_witness( + &rpc_client, + &cfg, + Some(&r2), + Some(5000), + block_number, + B256::ZERO, + started + budget, + ) + .await; + let elapsed = started.elapsed(); + + assert!( + result.is_ok(), + "the generator must serve block {block_number} after the probe: {:?}", + result.err() + ); + assert!(hits_gen.load(Ordering::Relaxed) >= 1, "the RPC chain must be reached"); + assert!( + elapsed < Duration::from_millis(600), + "a speculative probe must be cut at its eighth-share slice, not the \ + historical half (block {block_number}: {elapsed:?})" + ); + } + + ha.stop().unwrap(); + } + + /// An operator's explicit `--rpc-per-attempt-timeout-ms` below the witness cap asked + /// for stalled attempts to be cut *sooner*; the witness cap is a ceiling and must not + /// quietly loosen it. With a 200ms global per-attempt, the stalled hop is cut at + /// ~200ms and the fetch completes far inside the second — not at the witness cap. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn explicit_per_attempt_timeout_stays_the_tighter_bound() { + let (ha, url_gen, hits_gen) = scripted_witness_rpc(1, Some(fixture_wire())).await; + let url_gw = test_support::hanging_url(); + + let config = RpcClientConfig { + rpc_retry: BackoffPolicy::new(Duration::from_millis(20), Duration::from_millis(40)), + per_attempt_timeout: Duration::from_millis(200), + witness_per_attempt_timeout: Some(Duration::from_secs(10)), + ..RpcClientConfig::trace_server() + }; + let (rpc_client, cfg) = cap_fixture(&[url_gen.as_str(), url_gw.as_str()], config); + + let started = Instant::now(); + let result = fetch_witness( + &rpc_client, + &cfg, + None, + Some(5000), + 5001, + B256::ZERO, + started + Duration::from_secs(4), + ) + .await; + let elapsed = started.elapsed(); + + assert!(result.is_ok(), "the fetch must succeed: {:?}", result.err()); + assert!( + elapsed < Duration::from_secs(1), + "the 200ms global per-attempt must cut the stalled hop, not the witness cap \ + ({elapsed:?})" + ); + assert!(hits_gen.load(Ordering::Relaxed) >= 2, "round 1 must reach the generator"); + + ha.stop().unwrap(); + } + /// Old-block witness budget honors the configurable timeout and clamps to /// `witness_timeout`. #[test] diff --git a/bin/debug-trace-server/src/main.rs b/bin/debug-trace-server/src/main.rs index cfa0c659..48d825f9 100644 --- a/bin/debug-trace-server/src/main.rs +++ b/bin/debug-trace-server/src/main.rs @@ -3,10 +3,10 @@ //! # Overview //! A standalone RPC server for `debug_*` and `trace_*` methods using stateless execution. //! Data can be fetched from upstream RPC endpoints or from a local database with chain sync. -//! Request-serving witness fetches route by block age: historical blocks skip the internal -//! generator endpoint (which only retains a small recent window) and go straight to the -//! fallback endpoints — or, with the `--r2-*` flags, straight to the R2 bucket with the RPC -//! chain as fallback. Chain-sync prefetch always uses the full chain. +//! With the `--r2-*` flags every request-serving witness fetch tries the R2 bucket first, +//! falling back to the RPC chain; the chain itself routes by block age — historical blocks +//! skip the internal generator endpoint (which only retains a small recent window) and go +//! straight to the fallback endpoints. Chain-sync prefetch always uses the full chain. //! //! # Architecture //! ```text @@ -158,7 +158,9 @@ struct Args { #[clap(long, env = "DEBUG_TRACE_SERVER_START_BLOCK")] start_block: Option, - /// Witness fetch timeout in seconds. + /// Witness fetch timeout in seconds. No single witness-chain attempt may run longer + /// than half of what the stage has left when the chain starts, so one stalled endpoint + /// can never consume the whole stage and starve the retry rotation. #[clap( long, env = "DEBUG_TRACE_SERVER_WITNESS_TIMEOUT", @@ -311,9 +313,10 @@ struct Args { witness_old_block_timeout: Option, /// R2 S3 endpoint origin, e.g. `https://.r2.cloudflarestorage.com` (no bucket - /// path). With `--r2-bucket` and the credential flags, historical witnesses are fetched - /// straight from the bucket, with the RPC witness chain as fallback; frontier blocks keep - /// the generator path. Requires a local DB (`--data-dir`) to anchor block age. + /// path). With `--r2-bucket` and the credential flags, every witness fetch tries the + /// bucket first, with the RPC witness chain as fallback — frontier probes usually miss + /// (one fast 404) while historical fetches are served here. Requires a local DB + /// (`--data-dir`) to anchor block age. #[clap(long, env = "DEBUG_TRACE_SERVER_R2_ENDPOINT", requires_all = ["r2_bucket", "r2_access_key_id", "r2_secret_access_key"])] r2_endpoint: Option, @@ -474,13 +477,14 @@ fn validate_args(args: &Args) -> Result<()> { ); } } - // The R2 historical route anchors block age to the local DB tip, so without --data-dir - // it can never fire. An operator who configured R2 asked for that route explicitly — - // fail closed instead of silently running the RPC-only setup R2 exists to replace. + // The R2 route anchors block age (frontier vs historical) to the local DB tip; without + // --data-dir every block would classify as frontier and a genuine bucket hole would + // never reach the `kind="missing"` alarm. An operator who configured R2 asked for the + // real route — fail closed instead of running a blind approximation. if args.r2_endpoint.is_some() && args.data_dir.is_none() { eyre::bail!( - "--r2-endpoint requires --data-dir: the R2 historical witness route anchors \ - block age to the local DB tip and can never fire in stateless mode" + "--r2-endpoint requires --data-dir: the R2 witness route anchors block age \ + (frontier vs historical) to the local DB tip" ); } Ok(()) @@ -550,6 +554,10 @@ async fn main() -> Result<()> { data_max_concurrent_requests: args.data_max_concurrent_requests, witness_max_concurrent_requests: args.witness_max_concurrent_requests, per_attempt_timeout, + // Derived rather than flagged so it moves with --witness-timeout; a ceiling the + // per-entry halving normally undercuts — the full contract lives on + // `RpcClientConfig::witness_per_attempt_timeout`. + witness_per_attempt_timeout: Some(std::time::Duration::from_secs(args.witness_timeout) / 2), ..rpc_defaults } .with_metrics(Arc::new(metrics::TraceRpcMetrics)); @@ -580,7 +588,7 @@ async fn main() -> Result<()> { ), } - // Direct-from-R2 historical witness source. Clap's `requires` wiring makes the four + // Direct-from-R2 witness source. Clap's `requires` wiring makes the four // `--r2-*` flags all-or-nothing and `validate_args` rejects empty values and the // data-dir-less combination, so matching on the quad only splits "configured" from // "absent". Shares the RPC path's per-attempt timeout and retry pacing. diff --git a/bin/debug-trace-server/src/metrics.rs b/bin/debug-trace-server/src/metrics.rs index 704dbc4b..96b92235 100644 --- a/bin/debug-trace-server/src/metrics.rs +++ b/bin/debug-trace-server/src/metrics.rs @@ -483,13 +483,18 @@ pub fn record_request_shape(method: &'static str, shape: &'static str) { counter!(REQUEST_SHAPE_TOTAL, "method" => method, "shape" => shape).increment(1); } -/// R2 historical-witness GET retries (one increment per retried attempt). +/// R2 witness GET retries (one increment per retried attempt). const R2_WITNESS_RETRIES_TOTAL: &str = "debug_trace_r2_witness_retries_total"; -/// R2 historical-witness fetch failures, labeled by `kind` +/// R2 witness fetch failures, labeled by `kind` /// (see `crate::r2_witness::R2WitnessError::KINDS`). Failures here are not user-visible /// errors — the witness stage falls back to the RPC chain — so this counter is the signal -/// that the R2 fast path is degrading. +/// that the R2 fast path is degrading. `kind="missing"` stays the bucket-integrity alarm: +/// a frontier probe's miss is the expected ran-ahead-of-the-uploader outcome and is +/// deliberately not counted here (it still lands on +/// `witness_errors_total{source="witness_r2_frontier"}`); frontier = the near-tip +/// `data_provider::R2_FRONTIER_WINDOW` band, deliberately narrower than the routing +/// window (see its doc). const R2_WITNESS_ERRORS_TOTAL: &str = "debug_trace_r2_witness_errors_total"; /// Records one retried R2 witness GET attempt. @@ -727,12 +732,15 @@ fn pre_register_all_metrics() { let _ = DataSourceMetrics::new_for_source("witness_generator"); let _ = DataSourceMetrics::new_for_source("witness_historical"); let _ = DataSourceMetrics::new_for_source("witness_r2"); + let _ = DataSourceMetrics::new_for_source("witness_r2_frontier"); - // Data Fetch Layer: R2 historical witness source + // Data Fetch Layer: R2 witness source counter!(R2_WITNESS_RETRIES_TOTAL).increment(0); for kind in crate::r2_witness::R2WitnessError::KINDS { counter!(R2_WITNESS_ERRORS_TOTAL, "kind" => *kind).increment(0); } + counter!(R2_WITNESS_ERRORS_TOTAL, "kind" => crate::r2_witness::KIND_MISSING_ABOVE_TIP) + .increment(0); let _ = histogram!(R2_WITNESS_QUEUE_WAIT_SECONDS); // Data Fetch Layer: single-flight @@ -792,6 +800,7 @@ fn pre_register_all_metrics() { let _ = WitnessSourceMetrics::new_for_source("witness_generator"); let _ = WitnessSourceMetrics::new_for_source("witness_historical"); let _ = WitnessSourceMetrics::new_for_source("witness_r2"); + let _ = WitnessSourceMetrics::new_for_source("witness_r2_frontier"); // Execution Layer (per method) let _ = EvmExecutionMetrics::new_for_method(METHOD_DEBUG_TRACE_BLOCK_BY_NUMBER); diff --git a/bin/debug-trace-server/src/r2_witness.rs b/bin/debug-trace-server/src/r2_witness.rs index 863f820e..c2744a97 100644 --- a/bin/debug-trace-server/src/r2_witness.rs +++ b/bin/debug-trace-server/src/r2_witness.rs @@ -1,4 +1,4 @@ -//! Direct-from-R2 historical witness source. +//! Direct-from-R2 witness source. //! //! Fetches the primary witness object straight from the R2 bucket over the S3 API and decodes //! it with the **light** decoder — the trace server never verifies the witness proof, so the @@ -30,7 +30,14 @@ use crate::metrics; /// caller's deadline clamps the loop harder anyway. const MAX_ATTEMPTS: usize = 3; -/// Failure outcome of an R2 historical witness fetch. +/// Synthetic `kind` label for a `missing` above the frontier band — a catch-up-gap probe +/// whose bucket state is unknowable from the stale local tip. Kept off +/// [`R2WitnessError::KINDS`] (no error variant produces it); the band classifier in +/// `data_provider` records it so catch-up bursts stay visible without flooding the +/// below-band `kind="missing"` bucket-integrity alarm. +pub(crate) const KIND_MISSING_ABOVE_TIP: &str = "missing_above_tip"; + +/// Failure outcome of an R2 witness fetch. #[derive(Debug, thiserror::Error)] pub enum R2WitnessError { /// The signed GET failed (absent object, transport, throttle, unexpected status, or @@ -41,6 +48,11 @@ pub enum R2WitnessError { /// witness in R2. Deterministic; not retried. #[error("R2 witness for block {number} (key {key}) failed to decode: {source}")] Decode { number: u64, key: String, source: WitnessDecodingError }, + /// The decode outran what was left of the caller's deadline — an oversized or + /// pathological object. The caller falls back with its reserved share of the stage; + /// the blocking decode itself cannot be cancelled and finishes in the background. + #[error("R2 witness decode for block {number} (key {key}) outran the deadline")] + DecodeTimeout { number: u64, key: String }, /// The decode task panicked. This is a bug in our own decoder, not a problem with the /// data in R2, so it is kept out of [`Self::Decode`]. #[error("R2 witness decode task for block {number} (key {key}) panicked: {source}")] @@ -57,6 +69,7 @@ impl R2WitnessError { "connect", "deadline", "decode", + "decode_timeout", "decode_panicked", ]; @@ -66,12 +79,19 @@ impl R2WitnessError { match self { Self::Get(e) => e.kind(), Self::Decode { .. } => "decode", + Self::DecodeTimeout { .. } => "decode_timeout", Self::DecodePanicked { .. } => "decode_panicked", } } + + /// Whether the object was absent from the bucket — the one failure the frontier probe + /// treats as expected rather than alarming. + pub(crate) const fn is_missing(&self) -> bool { + matches!(self, Self::Get(R2GetError::Missing { .. })) + } } -/// Fetches and light-decodes historical witnesses straight from an R2 bucket. +/// Fetches and light-decodes witnesses straight from an R2 bucket. /// The fetcher's `Debug` redacts the credentials. #[derive(Debug)] pub struct R2WitnessSource { @@ -129,18 +149,36 @@ impl R2WitnessSource { // reported on its own series instead of being subtracted the way the validator's // throughput pipeline does. metrics::record_r2_witness_queue_wait(fetched.queue_wait.as_secs_f64()); - let bytes = fetched.bytes; - - // zstd + bincode over a multi-MB witness is CPU-bound; keep it off the runtime. - let key = || keys::block_object_key(number, hash); - match tokio::task::spawn_blocking(move || decode_witness_payload_light(&bytes)).await { - Ok(Ok(witness)) => { - trace!(number, "R2 witness fetched and light-decoded"); - Ok(witness) - } - Ok(Err(source)) => Err(R2WitnessError::Decode { number, key: key(), source }), - Err(source) => Err(R2WitnessError::DecodePanicked { number, key: key(), source }), + decode_light_with_deadline(fetched.bytes, number, hash, deadline).await + } +} + +/// Light-decodes `bytes` on the blocking pool, bounded by the same `deadline` as the GET — +/// an oversized or pathological object must not eat the RPC fallback's share of the stage. +/// On timeout the blocking task is abandoned (it cannot be cancelled) and finishes in the +/// background. +async fn decode_light_with_deadline( + bytes: bytes::Bytes, + number: u64, + hash: B256, + deadline: Instant, +) -> Result<(LightWitness, MptWitness), R2WitnessError> { + let key = || keys::block_object_key(number, hash); + // A GET that lands right at the deadline gets no decode at all — nothing would wait + // for it. + if Instant::now() >= deadline { + return Err(R2WitnessError::DecodeTimeout { number, key: key() }); + } + // zstd + bincode over a multi-MB witness is CPU-bound; keep it off the runtime. + let decode = tokio::task::spawn_blocking(move || decode_witness_payload_light(&bytes)); + match tokio::time::timeout_at(deadline.into(), decode).await { + Ok(Ok(Ok(witness))) => { + trace!(number, "R2 witness fetched and light-decoded"); + Ok(witness) } + Ok(Ok(Err(source))) => Err(R2WitnessError::Decode { number, key: key(), source }), + Ok(Err(source)) => Err(R2WitnessError::DecodePanicked { number, key: key(), source }), + Err(_) => Err(R2WitnessError::DecodeTimeout { number, key: key() }), } } @@ -241,4 +279,22 @@ mod tests { assert!(matches!(err, R2WitnessError::Decode { .. }), "{err}"); assert_eq!(err.kind(), "decode"); } + + /// A decode that outruns the deadline is abandoned so the caller falls back on its + /// reserved budget share, instead of the object holding the witness stage hostage. + /// Driven through the extracted decode step with the deadline already gone — the + /// GET-succeeds-then-decode-overruns timing cannot be scripted deterministically. + #[tokio::test] + async fn decode_past_the_deadline_surfaces_decode_timeout() { + let (salt_witness, mpt_witness): (_, MptWitness) = + TestFixtures::mainnet_shared().first_paired_witness(); + let (_, payload) = stateless_common::encode_witness_payload(&salt_witness, &mpt_witness) + .expect("fixture witness must encode"); + + let err = decode_light_with_deadline(payload.into(), 1, B256::ZERO, Instant::now()) + .await + .expect_err("an already-elapsed deadline must abandon the decode"); + assert!(matches!(err, R2WitnessError::DecodeTimeout { .. }), "{err}"); + assert_eq!(err.kind(), "decode_timeout"); + } } diff --git a/crates/stateless-common/src/rpc_client.rs b/crates/stateless-common/src/rpc_client.rs index 24e18154..3e7aa323 100644 --- a/crates/stateless-common/src/rpc_client.rs +++ b/crates/stateless-common/src/rpc_client.rs @@ -132,6 +132,30 @@ pub struct RpcClientConfig { /// timing out the attempt rotates `round_robin_with_backoff` to the next provider. /// With `deadline = Some(d)`, each attempt uses `min(per_attempt_timeout, d - now)`. pub per_attempt_timeout: Duration, + /// Witness-specific override of [`Self::per_attempt_timeout`], applied only when the + /// logical call carries a deadline. `None` inherits `per_attempt_timeout`. + /// + /// Exists because on the witness path the general cap is dead code: the witness stage + /// budget sits below `per_attempt_timeout`, so `min(per_attempt, remaining)` always + /// resolves to `remaining` and the first stalled endpoint can legally consume the + /// entire stage — leaving no budget for another rotation, which is what turns one slow + /// hop into a client-visible timeout. Deadline-bound witness attempts therefore run + /// under `AttemptCap::ReserveHalf`: while the round still has an untried provider to + /// rotate to, each attempt's window is `min(cap, per_attempt_timeout, remaining / 2)`, + /// with the halving recomputed as the attempt starts — after any concurrency-permit + /// wait — so neither a stalled hop nor a long permit queue can consume what a + /// rotation still needs. The round's last hop, and every hop of a single-provider + /// chain, takes the remainder whole under the ceiling instead: with no rotation left + /// to reserve for, a slow-but-honest transfer must be allowed to finish. The + /// configured value is a ceiling rather than the operative cap whenever the caller's + /// deadline stays within the budget it was derived from — the halving term is then + /// always at least as tight. + /// + /// Deadline-less witness fetches (the chain-sync prefetch) deliberately keep the + /// general cap: with unbounded rounds, cutting a deterministically-slow-but-honest + /// transfer at a value below its duration would re-cut it every round and wedge the + /// fetch permanently instead of merely making it slow. + pub witness_per_attempt_timeout: Option, /// Bound on the TCP connect phase of a single HTTP attempt, applied to every provider /// through the shared HTTP client. Separate from `per_attempt_timeout` so an unreachable /// endpoint — a dead host, or a firewall silently dropping the handshake — fails and @@ -157,6 +181,7 @@ impl Default for RpcClientConfig { // several seconds; everything else is sub-second), but bounded enough that a // stalled (TCP-accept-no-reply) provider is detected within reasonable time. per_attempt_timeout: Duration::from_secs(20), + witness_per_attempt_timeout: None, // 3s accommodates a WAN handshake that loses its first SYN (the retransmit fires // at ~1s) while keeping the cost of an unreachable provider to a quick rotation. connect_timeout: Duration::from_secs(3), @@ -173,6 +198,7 @@ impl std::fmt::Debug for RpcClientConfig { .field("witness_max_concurrent_requests", &self.witness_max_concurrent_requests) .field("rpc_retry", &self.rpc_retry) .field("per_attempt_timeout", &self.per_attempt_timeout) + .field("witness_per_attempt_timeout", &self.witness_per_attempt_timeout) .field("connect_timeout", &self.connect_timeout) .finish() } @@ -445,13 +471,14 @@ impl RpcClient { &self.data_provider_labels, &self.data_concurrency, &self.config.rpc_retry, - self.config.per_attempt_timeout, + AttemptCap::Fixed(self.config.per_attempt_timeout), rr_start, method, self.config.metrics.as_ref(), deadline, best_effort, |provider, _provider_label| f(provider.clone()), + |v, _provider_label| Box::pin(async move { Ok(v) }), ) .await } @@ -743,7 +770,7 @@ impl RpcClient { /// Shared `mega_getBlockWitness` retry loop: primary-failover rounds (always start from /// the first selected provider so the primary takes all traffic while healthy; backups /// are touched only while it is failing), each attempt one RPC round trip followed by the - /// caller-chosen `decode` (see [`fetch_witness_with`]). + /// caller-chosen `decode` (see [`fetch_witness_wire`] / [`decode_witness_wire`]). /// /// `providers` selects the contiguous index range of witness providers in the rotation; /// the logged endpoint labels stay aligned with the full configured list because each @@ -769,20 +796,36 @@ impl RpcClient { "witness provider range ({providers:?}) must select at least one of {} providers", self.witness_providers.len() ); + // Deadline-bound witness attempts run under the reserve-half policy: the tightest of + // the configured ceiling, the general per-attempt timeout, and — recomputed at each + // attempt, after any permit wait — half of what the call still has, so neither a + // stalled hop nor a long permit queue can leave the rotation nothing + // (`witness_per_attempt_timeout` documents the full contract). A deadline-less fetch + // keeps the general cap, since it must be able to out-wait transfers slower than the + // witness cap. + let attempt_cap = match (deadline, self.config.witness_per_attempt_timeout) { + (Some(_), Some(cap)) => { + AttemptCap::ReserveHalf { ceiling: cap.min(self.config.per_attempt_timeout) } + } + _ => AttemptCap::Fixed(self.config.per_attempt_timeout), + }; round_robin_with_backoff( &self.witness_providers[providers.clone()], &self.witness_provider_labels[providers], &self.witness_concurrency, &self.config.rpc_retry, - self.config.per_attempt_timeout, + attempt_cap, 0, RpcMethod::MegaGetBlockWitness, self.config.metrics.as_ref(), deadline, false, - |provider, provider_label| { + |provider, _provider_label| { + Box::pin(async move { fetch_witness_wire(&provider, number, hash).await }) + }, + move |wire, provider_label| { Box::pin(async move { - fetch_witness_with(&provider, &provider_label, number, hash, decode, trace_msg) + decode_witness_wire(wire, &provider_label, number, hash, decode, trace_msg) .await }) }, @@ -1015,6 +1058,28 @@ impl GiveUpPhase<'_> { } } +/// Per-attempt time-cap policy for [`round_robin_with_backoff`]. +enum AttemptCap { + /// Every attempt gets `min(cap, remaining budget)` — the general shape: the cap detects + /// a stalled provider, and near the deadline an attempt may use everything left. + Fixed(Duration), + /// Deadline-bound attempts get `min(ceiling, remaining / 2)`, recomputed when the + /// attempt starts — after any permit wait — but only while the round still has an + /// untried provider to rotate to: the reserve exists to buy exactly that rotation. + /// The round's last hop, and every hop of a single-provider chain, takes the + /// remainder whole under the ceiling instead — there is nothing left to reserve for, + /// and condemning a slow-but-honest transfer would buy nothing. The halving also + /// yields once half the remainder drops below [`MIN_RESERVE_HALF_ATTEMPT`] (a window + /// that small buys no useful follow-up hop). Without a deadline the ceiling alone + /// applies. + ReserveHalf { ceiling: Duration }, +} + +/// Smallest attempt window worth reserving budget past: below this a witness round trip +/// cannot complete anyway, so [`AttemptCap::ReserveHalf`] hands the tail to one final +/// attempt instead of splitting it into unwinnable micro-attempts. +const MIN_RESERVE_HALF_ATTEMPT: Duration = Duration::from_millis(25); + /// Runs a round-robin RPC call with round-level exponential backoff and an optional deadline. /// /// Each round attempts every provider once in round-robin starting at `rr_start`. If any @@ -1034,28 +1099,33 @@ impl GiveUpPhase<'_> { /// attempts are identified in logs and errors by their label alone, which bakes in the /// endpoint's index in the full list at construction, so slicing never misattributes an /// attempt. -// 11-argument retry primitive. Each field plays a distinct role (providers, their metric/log -// labels, concurrency, backoff policy, per-attempt timeout, starting provider, method label, -// metrics sink, deadline, give-up disposition, per-attempt closure) and bundling them into a struct -// would be ceremony without encapsulation — there are exactly two call sites in this crate. Prefer -// clarity at the definition over fewer commas at the call. `provider_labels` is parallel to -// `providers` by index. +// 12-argument retry primitive. Each field plays a distinct role (providers, their metric/log +// labels, concurrency, backoff policy, per-attempt cap policy, starting provider, method label, +// metrics sink, deadline, give-up disposition, windowed per-attempt closure, finalize closure) +// and bundling them into a struct would be ceremony without encapsulation — there are exactly two +// call sites in this crate. Prefer clarity at the definition over fewer commas at the call. +// `provider_labels` is parallel to `providers` by index. `f` runs under the attempt window; +// `finish` runs after it (bounded by the deadline alone): post-transport CPU work — the witness +// light decode — must neither burn the rotation reserve nor read as a provider stall, while its +// failure still counts as the provider's `Error` so corrupt-payload rotation is preserved. #[allow(clippy::too_many_arguments)] -async fn round_robin_with_backoff( +async fn round_robin_with_backoff( providers: &[RootProvider], provider_labels: &[Arc], semaphore: &Semaphore, policy: &BackoffPolicy, - per_attempt_timeout: Duration, + attempt_cap: AttemptCap, rr_start: usize, method: RpcMethod, metrics: Option<&Arc>, deadline: Option, best_effort: bool, - f: impl Fn(RootProvider, Arc) -> BoxFuture>, + f: impl Fn(RootProvider, Arc) -> BoxFuture>, + finish: impl Fn(W, Arc) -> BoxFuture>, ) -> std::result::Result where N: alloy_provider::Network, + W: Send + 'static, T: Send + 'static, { debug_assert_eq!( @@ -1154,13 +1224,36 @@ where // and success/error counters reflect what actually happened in the retry loop // rather than always showing "success" with the cumulative logical-call time. let attempt_start = Instant::now(); - // Bound every attempt by `per_attempt_timeout` — applied even with `deadline = - // None` so a provider that accepts the TCP connection but never replies cannot - // wedge the retry loop. With `deadline = Some(d)` the attempt is further capped - // by `d - now` so we never sleep past the caller's budget. - let attempt_timeout = match deadline { - Some(d) => d.saturating_duration_since(Instant::now()).min(per_attempt_timeout), - None => per_attempt_timeout, + // Bound every attempt by the cap policy — applied even with `deadline = None` + // so a provider that accepts the TCP connection but never replies cannot wedge + // the retry loop. With `deadline = Some(d)` the window is further clamped so we + // never sleep past the caller's budget; `ReserveHalf` recomputes its halving + // here, post-permit, so queue time already spent cannot inflate the window. + let attempt_timeout = match (deadline, &attempt_cap) { + (None, AttemptCap::Fixed(cap) | AttemptCap::ReserveHalf { ceiling: cap }) => *cap, + (Some(d), AttemptCap::Fixed(cap)) => { + d.saturating_duration_since(Instant::now()).min(*cap) + } + (Some(d), AttemptCap::ReserveHalf { ceiling }) => { + let remaining = d.saturating_duration_since(Instant::now()); + let half = remaining / 2; + // Halve only while this round still has an untried provider — the + // reserve exists to buy exactly that rotation, and only while the + // reserved window stays useful (>= MIN_RESERVE_HALF_ATTEMPT). The + // round's last hop — and every hop of a single-provider chain — takes + // the remainder whole instead: there is no rotation left to reserve + // for, and condemning a slow-but-honest transfer would buy nothing. + // The ceiling still bounds every window, which is what cuts the + // incident-shaped stall on a full stage and keeps a retry round + // reachable there. + let has_untried_provider = offset + 1 < n; + if has_untried_provider && half >= MIN_RESERVE_HALF_ATTEMPT { + half + } else { + remaining + } + .min(*ceiling) + } }; // Classify the attempt into a value or a (typed error, reason) pair, so the failure // reason — a returned error vs a per-attempt stall — survives to the metrics and logs. @@ -1171,7 +1264,31 @@ where ) .await { - Ok(Ok(v)) => Ok(v), + // Transport succeeded inside the window; finalize outside it, bounded + // by the caller's deadline alone. A finalize that outruns the deadline + // classifies as `Timeout` and the deadline check below reclassifies it + // to `DeadlineClamped` — the abandoned blocking decode finishes in the + // background (it cannot be cancelled), the same accepted trade as the + // R2 decode. + Ok(Ok(wire)) => { + let fin = finish(wire, Arc::clone(&provider_labels[slot])); + let fin = match deadline { + Some(d) => tokio::time::timeout_at(d.into(), fin).await, + None => Ok(fin.await), + }; + match fin { + Ok(Ok(v)) => Ok(v), + Ok(Err(e)) => Err((e, RpcAttemptOutcome::Error)), + Err(_) => Err(( + eyre!( + "{} finalize for provider {} outran the caller's deadline", + method.as_str(), + provider_label, + ), + RpcAttemptOutcome::Timeout, + )), + } + } Ok(Err(e)) => Err((e, RpcAttemptOutcome::Error)), Err(_) => Err(( eyre!( @@ -1278,7 +1395,9 @@ where if remaining_ms == 0 { return Err(record_deadline(GiveUpPhase::BeforeBackoff, round)); } - sleep_ms = sleep_ms.min(remaining_ms); + // Half of what is left, not all of it: a sleep that swallows the remainder + // wakes exactly at the deadline and forfeits the retry it was sleeping for. + sleep_ms = sleep_ms.min((remaining_ms / 2).max(1)); } log_at!( warn_level, @@ -1387,17 +1506,16 @@ async fn do_get_header( Ok(header) } -/// Shared single-attempt `mega_getBlockWitness` fetch: one RPC round trip, -/// then the caller-chosen decoder on the blocking pool (zstd + bincode over a -/// multi-MB payload is CPU-bound). -async fn fetch_witness_with( +/// Shared single-attempt `mega_getBlockWitness` transport: one RPC round trip, returning +/// the still-encoded wire string plus the round-trip time for the success trace. Decoding +/// lives in [`decode_witness_wire`], run by the retry loop's finalize step *outside* the +/// attempt window — CPU-bound decode must neither burn the rotation reserve nor be blamed +/// on the provider as a stall. +async fn fetch_witness_wire( provider: &RootProvider, - provider_label: &str, number: u64, hash: B256, - decode: fn(&str) -> std::result::Result, - trace_msg: &'static str, -) -> Result { +) -> Result<(String, u128)> { let keys = WitnessRequestKeys { block_number: U64::from(number), block_hash: hash }; let request_start = Instant::now(); let encoded: String = provider @@ -1405,8 +1523,22 @@ async fn fetch_witness_with( .request("mega_getBlockWitness", (keys,)) .await .map_err(|e| eyre!("mega_getBlockWitness failed for block {number}: {e}"))?; - let request_ms = request_start.elapsed().as_millis(); + Ok((encoded, request_start.elapsed().as_millis())) +} +/// [`fetch_witness_wire`]'s finalize half: the caller-chosen decoder on the blocking pool +/// (zstd + bincode over a multi-MB payload is CPU-bound), then the per-endpoint success +/// trace. A decode failure is a corrupt payload from this provider — the retry loop +/// records it as the provider's `Error` and rotates, exactly like a transport error. +async fn decode_witness_wire( + wire: (String, u128), + provider_label: &str, + number: u64, + hash: B256, + decode: fn(&str) -> std::result::Result, + trace_msg: &'static str, +) -> Result { + let (encoded, request_ms) = wire; let decode_start = Instant::now(); let result = tokio::task::spawn_blocking(move || -> Result { decode(&encoded).map_err(|e| eyre!("failed to decode witness response: {e}")) @@ -2214,6 +2346,102 @@ mod tests { handle.stop().unwrap(); } + /// The witness path's own semaphore feeds the same deadline-clamped acquire as the + /// data path: a witness call queued on a saturated `witness_concurrency` gives up at + /// its deadline — with the give-up and the cut-short permit-wait sample recorded — + /// instead of parking until an unrelated in-flight attempt frees a permit. + /// `test_deadline_bounds_permit_wait` pins the same clamp on the data semaphore. + #[tokio::test] + async fn test_witness_deadline_bounds_permit_wait() { + let (client, metrics) = metered_client( + &[LOCALHOST_A], + &[LOCALHOST_A], + RpcClientConfig { witness_max_concurrent_requests: Some(1), ..Default::default() }, + ); + // Hold the single witness permit for the whole test: the call stays queued and + // nothing is ever dialed. + let _held = client.witness_concurrency.acquire().await.unwrap(); + + let err = tokio::time::timeout( + Duration::from_secs(5), + client.get_witness_light_with_deadline( + 2, + B256::ZERO, + Some(Instant::now() + Duration::from_millis(20)), + ), + ) + .await + .expect("call must give up at its deadline instead of hanging") + .expect_err("no permit ever frees, so the call must exceed its deadline"); + assert_eq!(err.method, RpcMethod::MegaGetBlockWitness); + + assert_eq!( + *metrics.deadlines.lock().unwrap(), + vec![RpcMethod::MegaGetBlockWitness], + "exactly one give-up per logical call", + ); + assert_eq!( + *metrics.permit_waits.lock().unwrap(), + vec![RpcMethod::MegaGetBlockWitness], + "the cut-short wait still records a permit-wait sample", + ); + assert!( + metrics.attempts.lock().unwrap().is_empty(), + "no attempt ran — the whole budget died in the permit queue", + ); + } + + /// A long permit wait must not defeat the rotation reserve: `ReserveHalf` recomputes + /// its halving after the permit is acquired, so a call that spent most of its budget + /// queued still splits what is left across providers. Goes red with the cap frozen at + /// chain entry — post-queue `min(entry_cap, remaining)` hands the stalled primary + /// everything that is left and the fallback is never tried. + #[tokio::test] + async fn test_reserve_half_recomputes_after_permit_wait() { + // Primary stalls (accepts, never answers); the fallback answers junk fast — any + // completed round trip against it proves the rotation reached it. + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let url_a = format!("http://{}/", listener.local_addr().unwrap()); + let _held_open = listener; + let order = Arc::new(std::sync::Mutex::new(Vec::::new())); + let (hb, url_b) = start_ordered_witness_rpc('B', order).await; + + let (client, metrics) = metered_client( + &[url_b.as_str()], + &[url_a.as_str(), url_b.as_str()], + RpcClientConfig { + witness_max_concurrent_requests: Some(1), + witness_per_attempt_timeout: Some(Duration::from_millis(300)), + rpc_retry: BackoffPolicy::new(Duration::from_millis(1), Duration::from_millis(2)), + ..Default::default() + }, + ); + + // Hold the only witness permit for 500ms of the call's 800ms budget. + let permit = Arc::clone(&client.witness_concurrency).acquire_owned().await.unwrap(); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(500)).await; + drop(permit); + }); + + let deadline = Instant::now() + Duration::from_millis(800); + let _ = tokio::time::timeout( + Duration::from_secs(5), + client.get_witness_light_with_deadline(1, B256::ZERO, Some(deadline)), + ) + .await + .expect("call must give up by its deadline"); + + let attempts = metrics.attempts.lock().unwrap(); + let providers: std::collections::HashSet<&str> = + attempts.iter().map(|(_, p, _)| p.as_str()).collect(); + assert!( + providers.len() >= 2, + "post-queue halving must leave the fallback a window: {attempts:?}", + ); + hb.stop().unwrap(); + } + /// A provider that accepts the TCP connection but never replies must be detected by the /// per-attempt timeout — even when the call has no deadline (the chain-sync contract). /// Without per-attempt timing, the retry loop wedges on the stalled provider forever. @@ -2560,6 +2788,68 @@ mod tests { ); } + /// A finalize that outruns the caller's deadline is abandoned and books as + /// `deadline_clamped` — the attempt window must not be re-entered and the loop must + /// not hang on the finish future. Drives `round_robin_with_backoff` directly with an + /// instant transport and a never-finishing finalize, the seam the fixed witness + /// decoders leave uninjectable. Goes red if the finalize await loses its deadline + /// bound (the call hangs to the outer 5s guard) or lands on the wrong outcome. + #[tokio::test] + async fn test_finalize_outrun_books_as_deadline_clamped() { + let provider: RootProvider = + alloy_provider::ProviderBuilder::default().connect_http(LOCALHOST_A.parse().unwrap()); + let labels = [endpoint_label(LOCALHOST_A, 0)]; + let semaphore = Semaphore::new(1); + let policy = BackoffPolicy::new(Duration::from_millis(1), Duration::from_millis(2)); + let metrics = Arc::new(CapturingMetrics::default()); + let metrics_dyn: Arc = metrics.clone(); + + let started = Instant::now(); + let result = tokio::time::timeout( + Duration::from_secs(5), + round_robin_with_backoff( + std::slice::from_ref(&provider), + &labels, + &semaphore, + &policy, + AttemptCap::ReserveHalf { ceiling: Duration::from_secs(10) }, + 0, + RpcMethod::EthBlockNumber, + Some(&metrics_dyn), + Some(Instant::now() + Duration::from_millis(50)), + false, + |_provider, _label| Box::pin(async { Ok(()) }), + |(), _label| { + Box::pin(async { + tokio::time::sleep(Duration::from_secs(30)).await; + Ok(()) + }) + }, + ), + ) + .await + .expect("the finalize must be cut at the deadline, not awaited to completion") + .expect_err("an outrun finalize must surface the deadline give-up"); + assert_eq!(result.method, RpcMethod::EthBlockNumber); + assert!( + started.elapsed() < Duration::from_secs(2), + "gave up late: {:?}", + started.elapsed() + ); + + assert_eq!( + *metrics.attempts.lock().unwrap(), + vec![( + RpcMethod::EthBlockNumber, + endpoint_label(LOCALHOST_A, 0).to_string(), + RpcAttemptOutcome::DeadlineClamped, + )], + "the transport succeeded and the finalize ate the budget — deadline pressure, \ + not a provider stall", + ); + assert_eq!(*metrics.deadlines.lock().unwrap(), vec![RpcMethod::EthBlockNumber]); + } + /// A best-effort caller's deadline give-up must not log the operator-paging WARN. /// The normal path first proves the capture works (same `log_at!` expansion, so the /// same `warn!` callsite the absence assertion depends on), then the best-effort path