diff --git a/CHANGELOG.md b/CHANGELOG.md index 1445be56976..2becedf2cc3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org). ## [Unreleased] +### Changed + +- The Zakura block-sync request budget (`max_inflight_block_bytes`) is now + purely an outstanding-request cap: a body's wire reservation is released + when it is received instead of being held until the block is durable, so the + resident look-ahead budget (`max_reorder_lookahead_bytes`) is the single + bound on retained memory. Operators who used `max_inflight_block_bytes` to + bound node memory should set `max_reorder_lookahead_bytes` instead. The + request budget is also no longer raised to one checkpoint range (~802 MB) at + config load: configured values load unchanged, except values too small to + cover a single block request (~32 MiB with default response limits), which + are clamped to just above that floor with a warning rather than failing to + start. The `budget_reserved` trace field and metric keep their names but now + report outstanding request bytes only (previously they included retained + body bytes). + ### Removed - Removed two Zakura block-sync config fields that never needed operator diff --git a/zebra-network/src/config.rs b/zebra-network/src/config.rs index d0449443076..54ce0d9794c 100644 --- a/zebra-network/src/config.rs +++ b/zebra-network/src/config.rs @@ -1050,15 +1050,17 @@ impl<'de> Deserialize<'de> for Config { non_zero_config_field.filter(|config_value| config_value > &0).unwrap_or(default_config_value) }); - // Clamp the in-flight byte budget up to the checkpoint-range floor (with a - // warning) rather than rejecting too-small configs, so older configs keep - // starting while checkpoint sync stays deadlock-free. + // Clamp too-small budgets up (with a warning) rather than rejecting the + // config: the resident look-ahead budget up to one checkpoint range, so + // the resident-memory admission gate cannot deadlock checkpoint sync when + // verified_tip is pinned to the previous checkpoint; and the in-flight + // request budget up to just above one floor request, so stored configs + // written when that field also bounded retention keep starting. let mut zakura = zakura; - zakura.block_sync.clamp_inflight_block_bytes_to_floor(); - // Likewise clamp the resident look-ahead budget (and its block cap) up to one - // checkpoint range, so the resident-memory admission gate cannot deadlock checkpoint - // sync when verified_tip is pinned to the previous checkpoint. zakura.block_sync.clamp_reorder_lookahead_to_floor(); + zakura + .block_sync + .clamp_inflight_block_bytes_to_request_floor(); zakura.block_sync.validate().map_err(|error| { de::Error::custom(format!("invalid zakura.block_sync config: {error}")) })?; diff --git a/zebra-network/src/zakura/block_sync/README.md b/zebra-network/src/zakura/block_sync/README.md index 2221d8b3d85..c06ecf23c31 100644 --- a/zebra-network/src/zakura/block_sync/README.md +++ b/zebra-network/src/zakura/block_sync/README.md @@ -45,7 +45,7 @@ while anything anchored to the verified tip is pinned until real progress commit | Purpose | unblock the contiguous prefix so the committer keeps moving | fill the look-ahead buffer so the bursty committer never starves | | Candidate selection | first pending in `[servable_low, min(servable_high, floor_high)]`, only if this peer is the preferred floor carrier | first pending in the peer's servable range, only when the floor arm produced nothing | | cwnd slots | may borrow up to `floor_bypass_slots` (default 2) beyond a saturated cwnd — bypass slots fund the floor **only** | normal cwnd slots only | -| Byte funding | may **block**: `reserve_request_budget`'s floor path can shed an above-floor reorder body to fund the floor (`FundFloorReservation`) — reachable even at zero in-flight budget | non-blocking `try_reserve`; refused if the in-flight budget is spent | +| Byte funding | never refused: `reserve_request_budget`'s floor path overdrafts the in-flight budget by at most one request when `try_reserve` fails — reachable even at zero in-flight budget | non-blocking `try_reserve`; refused if the in-flight budget is spent | | Request deadline | short fixed leash (`floor_rescue_timeout`, default 2 s); on expiry the height is rescued to a faster carrier, the peer is retry-avoided but **not** disconnected | `request_timeout` (default 8 s) + expected transfer time (`estimated_bytes / measured BtlBw`, rate floored at 256 KiB/s) — patient, since it never gates the floor | **Floor carrier preference:** the floor rides the fastest servable peer. Before taking @@ -64,6 +64,13 @@ for the commit-window exemption, the resident-memory gate, and request sizing fill loop feeds its grant verbatim to the work queue and may not substitute its own sizing. +The one deliberate exception is `admission::admit_received_body`, the retention-only +gate for a body that is already downloaded (the unmatched-fallthrough path). A received +body consumes no request budget, so it applies the same commit-window exemption and +resident gate but never consults `budget_available` — a wire budget saturated by +outstanding requests must not force an already-paid-for body to be dropped and +re-downloaded. + ### The commit window Heights in `(verified_tip, verified_tip + 401]` are the **commit window** @@ -116,17 +123,20 @@ _eventual_ decoded cost, `wire_bytes × DESERIALIZED_MEM_FACTOR` (= 4): Two gates, checked together in `lookahead_over_budget`: -- **Byte gate:** `estimated_resident ≥ effective_max_reorder_lookahead_bytes`, where - `effective = min(max_reorder_lookahead_bytes, max_inflight_block_bytes × 4)`. +- **Byte gate:** `estimated_resident ≥ effective_max_reorder_lookahead_bytes` + (= `max_reorder_lookahead_bytes`; the request budget no longer caps it, since + the request cap does not imply retention). - **Block gate (defense in depth):** reorder + applying + reserved block counts `≥ LOOKAHEAD_BLOCK_HARD_CAP` (a fixed 262,144 — it binds before the byte gate only for tiny bodies averaging under ~6.1 KB wire, where per-entry bookkeeping overhead dominates; never needed operator tuning, so it is a constant, not a config knob). -This is separate from the **in-flight wire budget** (`max_inflight_block_bytes`, -default 6 GiB, tracked by `ByteBudget`): that bounds bytes concurrently on the wire; -the look-ahead gate bounds bytes _retained_ by the pipeline. +This is separate from the **in-flight request budget** (`max_inflight_block_bytes`, +default 6 GiB, tracked by `ByteBudget`): that bounds outstanding request +reservations — charged at issuance, released at receipt (or timeout/watchdog/ +reset/floor GC) — while the look-ahead gate is the single authority over bytes +_retained_ by the pipeline. ### Config clamps @@ -158,12 +168,13 @@ retained wire 807.7–807.9 MB → ×4 = 3.231 GB, +0.7% over budget. look-ahead gates — a pinned checkpoint range can always assemble. 2. **Floor grants never size below one byte**, and `take_in_range_budgeted` always takes its first item regardless of the byte cap — so the floor block is taken even - when the in-flight budget is exactly full, reaching the floor funding path… -3. **…which can shed to fund.** `reserve_request_budget`'s Floor path awaits - `FundFloorReservation`: an above-floor reorder body is shed (returned to pending) - to free bytes for the floor block. The floor is therefore never starved by - speculative work, and this path is deliberately independent of the in-flight byte - budget. + when the in-flight budget is exactly full, reaching the floor reservation path… +3. **…which overdrafts instead of waiting.** When `try_reserve` fails, + `reserve_request_budget`'s Floor path charges the reservation past the max — a + bounded overshoot of at most one request (the WorkQueue single-owner invariant + permits one floor reservation globally), repaid through the normal release + discipline. The floor is therefore never starved by speculative work, with no + cross-task funding round trip. 4. **In-window floor liveness never depends on budget size** — the clamps only stop sub-range configs from thrashing the speculative lane. @@ -191,8 +202,8 @@ borrowed a bypass slot. | Knob | Default | Notes | | --- | --- | --- | -| `max_reorder_lookahead_bytes` | ~6.4 GB | **resident-denominated** (compared against wire × 4); effective value capped at `max_inflight_block_bytes × 4`; clamped up to ~3.208 GB | -| `max_inflight_block_bytes` | 6 GiB | in-flight wire budget (separate from the resident gate) | +| `max_reorder_lookahead_bytes` | ~6.4 GB | **resident-denominated** (compared against wire × 4); clamped up to ~3.208 GB | +| `max_inflight_block_bytes` | 6 GiB | outstanding-request wire budget, released at receipt (separate from the resident gate) | | `max_blocks_per_response` | 1 | count cap per request (effective = min of both sides' advertisements, hard max 128) | | `floor_bypass_slots` | 2 | extra slots past a saturated cwnd, floor lane only | | `request_timeout` / `floor_rescue_timeout` | 8 s / 2 s | above-floor base deadline / floor rescue leash | diff --git a/zebra-network/src/zakura/block_sync/admission.rs b/zebra-network/src/zakura/block_sync/admission.rs index 890a7eacce5..dc68688bb78 100644 --- a/zebra-network/src/zakura/block_sync/admission.rs +++ b/zebra-network/src/zakura/block_sync/admission.rs @@ -69,9 +69,9 @@ pub(super) enum AdmissionOutcome { /// rounds to zero. LookaheadAtCap, /// The look-ahead gate has headroom but zero bytes are fundable right now - /// (the in-flight byte budget is spent). Never returned for floor-priority + /// (the in-flight request budget is spent). Never returned for floor-priority /// starts: their byte cap is floored at one so the floor block always - /// reaches the floor-reservation funding path. + /// reaches the bounded-overdraft reservation path. InflightBudgetEmpty, } @@ -243,6 +243,39 @@ fn lookahead_over_budget(config: &ZakuraBlockSyncConfig, snapshot: &AdmissionSna || held_blocks(snapshot) >= LOOKAHEAD_BLOCK_HARD_CAP } +/// Remaining resident look-ahead headroom, expressed back in wire bytes so one +/// admitted response cannot push resident memory past the budget. The next +/// admitted body will usually become decoded soon, so it is sized as if it +/// costs the decoded multiple. +fn remaining_lookahead_wire_bytes( + config: &ZakuraBlockSyncConfig, + snapshot: &AdmissionSnapshot, +) -> u64 { + config + .effective_max_reorder_lookahead_bytes() + .saturating_sub(estimated_resident_pipeline_bytes(snapshot)) + / DESERIALIZED_MEM_FACTOR +} + +/// Retention-only admission for a body that is already downloaded. +/// +/// A received body consumes no request budget (its wire reservation is +/// released at receipt), so unlike [`admit`] this never consults +/// `budget_available`: only the commit-window exemption and the resident +/// look-ahead gate — the two rules that bound retention — apply. +pub(super) fn admit_received_body( + config: &ZakuraBlockSyncConfig, + snapshot: &AdmissionSnapshot, + height: block::Height, + serialized_bytes: u64, +) -> bool { + if height <= commit_window_high(snapshot) { + return true; + } + !lookahead_over_budget(config, snapshot) + && serialized_bytes <= remaining_lookahead_wire_bytes(config, snapshot) +} + /// Plans one contiguous take starting at `start_height`: the single authority for /// the commit-window exemption, the resident-memory gate, and request sizing. /// @@ -265,9 +298,10 @@ fn lookahead_over_budget(config: &ZakuraBlockSyncConfig, snapshot: &AdmissionSna /// ≈ 3.2 GB; a single in-window response can also exceed the byte gate by up to the /// response cap × the factor) regardless of how far headers/downloads run ahead. /// -/// Floor-priority requests are never blocked just because the normal byte budget is exactly full. -/// If the lowest missing block is needed to let commit move forward, -/// it can still be requested even when speculative/look-ahead work has filled the byte budget. +/// Floor-priority requests are never blocked just because the request budget is exactly +/// full. If the lowest missing block is needed to let commit move forward, it can still +/// be requested even when speculative work has spent the in-flight budget (the routine's +/// bounded floor overdraft funds it). pub(super) fn admit( config: &ZakuraBlockSyncConfig, snapshot: AdmissionSnapshot, @@ -289,13 +323,7 @@ pub(super) fn admit( if lookahead_over_budget(config, &snapshot) { return AdmissionOutcome::LookaheadAtCap; } - // Remaining memory headroom, expressed back in wire bytes for the response cap so a - // single response can't push resident memory past the budget. The next admitted body - // will usually become decoded soon, so it is sized as if it costs the decoded multiple. - let remaining_wire_bytes = config - .effective_max_reorder_lookahead_bytes() - .saturating_sub(estimated_resident_pipeline_bytes(&snapshot)) - / DESERIALIZED_MEM_FACTOR; + let remaining_wire_bytes = remaining_lookahead_wire_bytes(config, &snapshot); if remaining_wire_bytes == 0 { return AdmissionOutcome::LookaheadAtCap; } @@ -445,9 +473,10 @@ mod tests { ), "the final block of a checkpoint range must be admissible under a 1 GiB in-flight budget", ); - // The first height above the window is memory-gated but must still have headroom - // under this budget: (802 MB - 2 MB) * 4 resident < 4 GiB effective. This keeps the - // assertion non-vacuous now that the whole range is window-exempt. + // The first height above the window is memory-gated but must still have headroom: + // the snapshot holds one block less than a full range, so its resident estimate is + // (802 MB - 2 MB) * 4 ~= 3.2 GB, below the default resident budget (~6.4 GB). + // This keeps the assertion non-vacuous now that the whole range is window-exempt. assert!( matches!( admit( diff --git a/zebra-network/src/zakura/block_sync/config.rs b/zebra-network/src/zakura/block_sync/config.rs index c283b14f050..25a270cd9d8 100644 --- a/zebra-network/src/zakura/block_sync/config.rs +++ b/zebra-network/src/zakura/block_sync/config.rs @@ -28,25 +28,26 @@ pub const DEFAULT_BS_INITIAL_INFLIGHT: u32 = 64; pub const MAX_BS_INFLIGHT_REQUESTS: u32 = 32_768; /// Default total response byte target advertised per range response. pub const DEFAULT_BS_MAX_RESPONSE_BYTES: u32 = 32 * 1024 * 1024; -/// Default global byte budget reserved for later block-download scheduling. +/// Default global byte budget for outstanding block-request reservations. pub const DEFAULT_BS_MAX_INFLIGHT_BLOCK_BYTES: u64 = 6 * 1024 * 1024 * 1024; /// Worst-case serialized bytes reserved per requested block body. /// -/// Block-sync reserves this much per requested block at send time and only ever -/// shrinks the reservation toward the actual serialized size on receipt, so a -/// valid, already-downloaded body is never discarded for a full budget. Each +/// Block-sync reserves a per-block size estimate at send time (this worst case +/// when no hint is known) and releases it when the body arrives, so a valid, +/// already-downloaded body is never discarded against the request budget. Each /// body arrives in its own `Block` frame bounded by [`block::MAX_BLOCK_BYTES`] /// at decode (`MAX_BS_MESSAGE_BYTES > MAX_BLOCK_BYTES`), so the actual size can -/// never exceed this worst case and the shrink is always non-negative. +/// never exceed this worst case. pub const BS_PER_BLOCK_WORST_CASE_BYTES: u64 = block::MAX_BLOCK_BYTES; /// Default cap on the estimated *resident* memory of the look-ahead pipeline. /// /// Denominated in resident bytes, not wire bytes: admission compares it against the /// retained and in-flight wire bytes scaled by `DESERIALIZED_MEM_FACTOR` (see /// `admission::estimated_resident_pipeline_bytes`), so the default admits roughly a -/// quarter of its nominal value in wire bytes. The numeric value is kept aligned with -/// the in-flight wire budget minus one advertised response, which under the resident -/// interpretation yields a deep (~1.5 GiB wire) look-ahead buffer. +/// quarter of its nominal value in wire bytes (a deep ~1.5 GiB wire look-ahead). +/// The retention budget is independent of the request cap +/// (`max_inflight_block_bytes` bounds outstanding requests, not retained bodies); +/// the numeric value below is only historically derived from it. pub const DEFAULT_BS_MAX_REORDER_LOOKAHEAD_BYTES: u64 = // `DEFAULT_BS_MAX_RESPONSE_BYTES` is a `u32`, so widening to `u64` is lossless. DEFAULT_BS_MAX_INFLIGHT_BLOCK_BYTES - DEFAULT_BS_MAX_RESPONSE_BYTES as u64; @@ -58,15 +59,14 @@ pub const DEFAULT_BS_MAX_REORDER_LOOKAHEAD_BYTES: u64 = /// maximum checkpoint gap plus the boundary block in flight. pub const MIN_BS_CHECKPOINT_SUBMITTED_BLOCK_APPLIES: usize = zebra_chain::parameters::checkpoint::constants::MAX_CHECKPOINT_HEIGHT_GAP + 1; -/// The byte budget required to hold one full worst-case checkpoint range in -/// flight. +/// The byte size of one full worst-case checkpoint range held in flight. /// /// The checkpoint verifier resolves a block's commit only once the entire -/// contiguous range to the next checkpoint has been submitted, and every -/// submitted body stays reserved against `max_inflight_block_bytes` until it is -/// durable. A budget that cannot hold a whole worst-case range can never -/// complete one: the verifier never commits, nothing becomes durable, and no -/// bytes are ever released. +/// contiguous range to the next checkpoint has been submitted, so the whole +/// range must be co-resident. This floor sizes the resident look-ahead clamp +/// (`clamp_reorder_lookahead_to_floor`) and `validate()`'s lower bound on the +/// request budget, so a configuration can never make a full range +/// unassemblable. pub const BS_CHECKPOINT_RANGE_BYTE_FLOOR: u64 = // `MIN_BS_CHECKPOINT_SUBMITTED_BLOCK_APPLIES` is `MAX_CHECKPOINT_HEIGHT_GAP + 1` // (= 401), which fits `u64` losslessly; the product (~802 MB) cannot overflow. @@ -233,9 +233,12 @@ pub struct ZakuraBlockSyncConfig { pub initial_inflight_requests: u32, /// Maximum total response bytes this node advertises per `GetBlocks` response. pub max_response_bytes: u32, - /// Maximum estimated bytes reserved for in-flight and buffered block bodies. + /// Maximum estimated bytes reserved for outstanding block-body requests: a + /// DoS/pacing bound on in-flight wire data, released at receipt. Received + /// bodies are bounded by `max_reorder_lookahead_bytes` instead. pub max_inflight_block_bytes: u64, - /// Maximum speculative body bytes held above the download floor. + /// Maximum estimated *resident* memory of look-ahead block bodies retained + /// by the download pipeline (wire bytes × `DESERIALIZED_MEM_FACTOR`). pub max_reorder_lookahead_bytes: u64, /// How long to avoid reassigning an expired floor height to the same peer. #[serde(with = "humantime_serde")] @@ -378,16 +381,14 @@ impl ZakuraBlockSyncConfig { .max(MIN_BS_CHECKPOINT_SUBMITTED_BLOCK_APPLIES) } - /// Return the speculative look-ahead byte cap clamped to the global budget. + /// Return the resident look-ahead byte cap. + /// + /// Currently the raw configured value; kept as an accessor for the semantic + /// seam with the clamp interplay. The outstanding-request cap + /// (`max_inflight_block_bytes`) no longer implies retention, so it must not + /// silently shrink this memory budget. pub fn effective_max_reorder_lookahead_bytes(&self) -> u64 { - // This is a resident-memory budget: admission counts each pool's wire bytes scaled by - // `DESERIALIZED_MEM_FACTOR`. Cap it against the resident equivalent of the in-flight - // wire budget, not raw `max_inflight_block_bytes`, so look-ahead depth is not - // unnecessarily starved. - self.max_reorder_lookahead_bytes.min( - self.max_inflight_block_bytes - .saturating_mul(super::admission::DESERIALIZED_MEM_FACTOR), - ) + self.max_reorder_lookahead_bytes } /// Return the floor avoid cooldown clamped to a positive duration. @@ -463,32 +464,6 @@ impl ZakuraBlockSyncConfig { Ok(()) } - /// Raise `max_inflight_block_bytes` up to the checkpoint-range floor when it - /// is configured below it, warning once. - /// - /// A positive budget below [`BS_CHECKPOINT_RANGE_BYTE_FLOOR`] cannot hold one - /// full worst-case checkpoint range. The checkpoint verifier only commits a - /// range once the whole range is submitted, and every submitted body stays - /// reserved against the budget until it is durable, so a budget below the - /// floor would deadlock: the verifier never commits, nothing becomes durable, - /// and no bytes are ever released. Rather than refuse to start -- which would - /// break older configs that set a smaller budget -- clamp the budget up to the - /// floor and warn. Zero is left untouched so [`validate`](Self::validate) - /// still rejects it as an explicit misconfiguration. - pub fn clamp_inflight_block_bytes_to_floor(&mut self) { - if self.max_inflight_block_bytes > 0 - && self.max_inflight_block_bytes < BS_CHECKPOINT_RANGE_BYTE_FLOOR - { - tracing::warn!( - configured_max_inflight_block_bytes = self.max_inflight_block_bytes, - checkpoint_range_byte_floor = BS_CHECKPOINT_RANGE_BYTE_FLOOR, - "zakura.block_sync.max_inflight_block_bytes is below the checkpoint-range \ - floor; clamping it up so checkpoint sync cannot deadlock", - ); - self.max_inflight_block_bytes = BS_CHECKPOINT_RANGE_BYTE_FLOOR; - } - } - /// Clamp the resident look-ahead budget up to one worst-case checkpoint range. /// /// This is defense-in-depth for the speculative above-window lane: checkpoint @@ -510,6 +485,27 @@ impl ZakuraBlockSyncConfig { } } + /// Clamp a positive but sub-floor-request `max_inflight_block_bytes` up to + /// just above one floor request. + /// + /// `validate()` requires the outstanding-request budget to cover at least + /// one floor request (the bounded floor overdraft repays against it). + /// Rather than refuse to start — which would break older configs written + /// when the field also bounded retention and small values were clamped — + /// raise the budget to the smallest valid value and warn. + pub fn clamp_inflight_block_bytes_to_request_floor(&mut self) { + let request_floor = self.floor_request_byte_reservation(); + if self.max_inflight_block_bytes > 0 && self.max_inflight_block_bytes <= request_floor { + tracing::warn!( + configured_max_inflight_block_bytes = self.max_inflight_block_bytes, + floor_request_byte_reservation = request_floor, + "zakura.block_sync.max_inflight_block_bytes cannot cover one floor \ + request; clamping it up so the node can start", + ); + self.max_inflight_block_bytes = request_floor.saturating_add(1); + } + } + /// Build the inert local status used before the block-sync reactor is wired. pub fn initial_status(&self) -> BlockSyncStatus { BlockSyncStatus { diff --git a/zebra-network/src/zakura/block_sync/mod.rs b/zebra-network/src/zakura/block_sync/mod.rs index 2ba9d8f6bc2..ddc5f23a73e 100644 --- a/zebra-network/src/zakura/block_sync/mod.rs +++ b/zebra-network/src/zakura/block_sync/mod.rs @@ -15,7 +15,7 @@ use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt}; use serde::{Deserialize, Serialize}; use thiserror::Error; use tokio::{ - sync::{mpsc, oneshot, watch}, + sync::{mpsc, watch}, task::JoinHandle, time, }; diff --git a/zebra-network/src/zakura/block_sync/peer_registry.rs b/zebra-network/src/zakura/block_sync/peer_registry.rs index f397e9a2a52..31a20acfd2a 100644 --- a/zebra-network/src/zakura/block_sync/peer_registry.rs +++ b/zebra-network/src/zakura/block_sync/peer_registry.rs @@ -146,6 +146,21 @@ impl PeerRegistry { .expect("peer registry mutex is never poisoned") } + /// Diagnostic sizes for the periodic state trace: (peer entries, total + /// published outstanding height entries, parked peers). Cheap map-length + /// sums; used to attribute memory growth to a specific structure. + pub(super) fn diagnostic_sizes(&self) -> (u64, u64, u64) { + let peers = self.lock(); + let outstanding: u64 = peers + .values() + .map(|entry| entry.outstanding.len() as u64) + .sum(); + let peer_count = peers.len() as u64; + drop(peers); + let parked = self.lock_parked().len() as u64; + (peer_count, outstanding, parked) + } + fn lock_parked(&self) -> std::sync::MutexGuard<'_, HashMap> { self.parked_peers .lock() diff --git a/zebra-network/src/zakura/block_sync/peer_routine.rs b/zebra-network/src/zakura/block_sync/peer_routine.rs index ad19a79ddf2..0aeb8132828 100644 --- a/zebra-network/src/zakura/block_sync/peer_routine.rs +++ b/zebra-network/src/zakura/block_sync/peer_routine.rs @@ -23,14 +23,15 @@ use std::collections::BTreeMap; -use tokio::sync::{futures::Notified, mpsc, oneshot, watch}; +use tokio::sync::{futures::Notified, mpsc, watch}; use tokio_util::sync::CancellationToken; use super::events::RoutineToReactor; use super::{ admission::{ - admit, floor_rescue_high, request_deadline, request_priority as classify_priority, - AdmissionOutcome, AdmissionSnapshot, RequestPriority, + admit, admit_received_body, floor_rescue_high, request_deadline, + request_priority as classify_priority, AdmissionOutcome, AdmissionSnapshot, + RequestPriority, }, peer_registry::{hard_outbound_capacity, PeerRegistry}, pipe::block_sync_guard, @@ -40,7 +41,7 @@ use super::{ }, reorder::BufferedBlockBody, request::{BlockRangeRequest, ExpectedBlock}, - sequencer_task::{SequencedBody, SequencerControlInput, SequencerView}, + sequencer_task::{SequencedBody, SequencerView}, state::{ DownloadWindow, LivenessOutcome, OutstandingBlockRange, ReceivedBlockTracker, ThroughputMeter, @@ -215,7 +216,6 @@ pub(super) struct PeerRoutine { received_throughput: Arc>, sequencer_input: mpsc::Sender, sequencer_input_bytes: Arc, - sequencer_control: mpsc::UnboundedSender, actions: mpsc::Sender, /// Shared routine→reactor channel for serving / status-advertise / re-query / /// serving-misbehavior. `try_send` (bounded, never-wedging) so a busy reactor @@ -255,7 +255,6 @@ impl PeerRoutine { received_throughput: Arc>, sequencer_input: mpsc::Sender, sequencer_input_bytes: Arc, - sequencer_control: mpsc::UnboundedSender, actions: mpsc::Sender, routine_to_reactor: mpsc::Sender, sequencer_view: watch::Receiver, @@ -296,7 +295,6 @@ impl PeerRoutine { received_throughput, sequencer_input, sequencer_input_bytes, - sequencer_control, actions, routine_to_reactor, sequencer_view, @@ -836,10 +834,7 @@ impl PeerRoutine { let reserved_bytes = items.iter().fold(0u64, |acc, (_, item)| { acc.saturating_add(item.estimated_bytes) }); - if !self - .reserve_request_budget(request_priority, reserved_bytes) - .await - { + if !self.reserve_request_budget(request_priority, reserved_bytes) { self.return_taken_items(&items); break FillStop::Budget; } @@ -996,37 +991,19 @@ impl PeerRoutine { .unwrap_or(usize::MAX) } - async fn reserve_request_budget( - &mut self, - priority: RequestPriority, - reserved_bytes: u64, - ) -> bool { - if priority == RequestPriority::AboveFloor { - return self.budget.try_reserve(reserved_bytes); + fn reserve_request_budget(&mut self, priority: RequestPriority, reserved_bytes: u64) -> bool { + if self.budget.try_reserve(reserved_bytes) { + return true; } - - loop { - if self.budget.try_reserve(reserved_bytes) { - return true; - } - - let (reply, funded) = oneshot::channel(); - if self - .sequencer_control - .send(SequencerControlInput::FundFloorReservation { - needed_bytes: reserved_bytes, - reply, - }) - .is_err() - { - return false; - } - - match funded.await { - Ok(true) => continue, - Ok(false) | Err(_) => return false, - } + if priority == RequestPriority::Floor { + // The WorkQueue owns each height once, so there can only be one + // floor-priority overdraft globally. Its charge is released by the + // normal reservation paths: receipt, timeout, watchdog, or reset. + self.budget.charge(reserved_bytes); + metrics::counter!("sync.block.budget.floor_overdraft").increment(1); + return true; } + false } /// Refill low-water mark in blocks, computed from a single peer's caps. @@ -1285,23 +1262,19 @@ impl PeerRoutine { metrics::counter!("sync.block.body.received").increment(1); self.record_received(serialized_bytes); - // The block reserved its size estimate at send time; settle to the actual - // size. When the body is no larger than its estimate this frees the - // slack; when it is larger (a stale/under-advertised hint) this charges - // the overshoot so held bodies are never under-counted. - // `mark_received` then stops `reserved_bytes()` counting this height; the - // only bytes still held are the `serialized_bytes` carried into the reorder - // buffer. - let Some(delta) = self - .work - .settle_active_reserved_height(height, serialized_bytes) - else { + // The block reserved its size estimate at send time. Receipt ends the + // wire reservation: the estimate goes back to the request budget below, + // after the body is handed to the sequencer, so a woken issuer's resident + // snapshot already sees the body. Retention of the received body is + // bounded by the resident look-ahead gate, not the in-flight budget. + let Some(reserved_estimate) = self.work.release_active_reserved_height(height) else { // The reservation is gone: a competing peer already delivered this // height first (first-completion-wins), a watchdog released it, or it // committed past the floor. In every case the body is already in the // commit pipeline, so mark it received rather than re-queuing it — a // retry here would phantom-re-fetch a body we already hold, and any - // release belongs to whoever settled it, not to this stale claim. + // release belongs to whoever ended the reservation, not to this stale + // claim. tracing::debug!( peer = ?self.peer, ?height, @@ -1311,7 +1284,6 @@ impl PeerRoutine { self.accept_already_settled_height(index, height); return; }; - self.apply_budget_delta(delta); self.trace_body_received( height, serialized_bytes, @@ -1355,6 +1327,12 @@ impl PeerRoutine { let body = BufferedBlockBody::from_decoded_block(block, raw_block_payload); self.forward_body_to_sequencer(height, hash, body, serialized_bytes, body_permit) .await; + // Release the request reservation only now that the body is counted in + // `sequencer_input_bytes` (or was dropped by a failed forward), so it is + // never invisible to both the limiter and the resident snapshot. Exactly + // once on every path: the ledger released above returns `None` to any + // later claimant. + self.budget.release(reserved_estimate); // This body opened only this peer's slots; the want-work loop runs at the // top of the next iteration. } @@ -1414,10 +1392,11 @@ impl PeerRoutine { self.trace_body_sequencer_sent(height, sequencer_send_started.elapsed(), ok); } - /// Accept a wanted unmatched body whose original requester is gone or whose height - /// is currently reserved by another peer. Queued heights reserve their actual size - /// before buffering; reserved in-flight heights settle the existing reservation to - /// the actual held bytes. + /// Accept a wanted unmatched body whose original requester is gone or whose + /// height is currently reserved by another peer. The resident `admit()` check + /// is the sole gate for queued heights — a received body consumes no request + /// budget; for reserved in-flight heights the arrival ends that request's + /// reservation (first-completion-wins). async fn accept_unmatched_queued_body( &mut self, height: block::Height, @@ -1458,66 +1437,39 @@ impl PeerRoutine { return false; } - if is_pending { + // The reservation this arrival ended (an active competing request, or a + // stale charge on the claimed height); released after the forward below. + let ended_reservation = if is_pending { let sequencer_view = *self.sequencer_view.borrow(); let snapshot = self.admission_snapshot(&sequencer_view); - let admitted_bytes = - match admit(&self.config, snapshot, height, height, serialized_bytes) { - AdmissionOutcome::Admit(grant) => grant.max_request_bytes, - AdmissionOutcome::LookaheadAtCap | AdmissionOutcome::InflightBudgetEmpty => { - tracing::debug!( - peer = ?self.peer, - ?height, - serialized_bytes, - "not buffering unmatched queued block-sync body at look-ahead cap" - ); - return true; - } - }; - if admitted_bytes < serialized_bytes { - tracing::debug!( - peer = ?self.peer, - ?height, - serialized_bytes, - admitted_bytes, - "not buffering unmatched queued block-sync body; insufficient admitted budget" - ); - return true; - } - - // This queued height owns no prior reservation: reserve its actual size - // before buffering. If the budget is genuinely full of other legitimate - // bodies, skip buffering (the height stays queued for retry with its own - // size-estimate reservation, so no valid body is lost overall). - if !self.budget.try_reserve(serialized_bytes) { + if !admit_received_body(&self.config, &snapshot, height, serialized_bytes) { tracing::debug!( peer = ?self.peer, ?height, serialized_bytes, - "not buffering unmatched queued block-sync body; height stays queued for retry" + "not buffering unmatched queued block-sync body at look-ahead cap" ); return true; } - // Claim this height into `in_flight` so it leaves `pending`. + // Claim this height into `in_flight` so it leaves `pending`; if it is + // already `in_flight` the take is a no-op and the Sequencer drops the + // later duplicate. The received body charges no request budget, but any + // stale request reservation the height still owned is released below. let _ = self.work.take_in_range(height, height, 1); - let old_charge = self.work.mark_held_direct(height, serialized_bytes); - self.budget.release(old_charge); metrics::counter!("sync.block.response.unmatched_queued_accepted").increment(1); + self.work.claim_received(height) } else { - // First-completion-wins for a timed-out height already re-issued to another - // peer: convert the active request reservation into held bytes and settle - // only the delta, instead of discarding a valid body because another peer - // currently owns the request slot. - let Some(delta) = self - .work - .settle_active_reserved_height(height, serialized_bytes) - else { + // First-completion-wins for a timed-out height already re-issued to + // another peer: this arrival ends that request's reservation instead of + // discarding a valid body because another peer currently owns the + // request slot. + let Some(estimate) = self.work.release_active_reserved_height(height) else { return false; }; - self.apply_budget_delta(delta); metrics::counter!("sync.block.response.unmatched_active_accepted").increment(1); - } + estimate + }; self.record_received(serialized_bytes); self.trace_body_received(height, serialized_bytes, None, None, None); @@ -1540,6 +1492,11 @@ impl PeerRoutine { let body = BufferedBlockBody::from_decoded_block(block, raw_block_payload); self.forward_body_to_sequencer(height, hash, body, serialized_bytes, body_permit) .await; + // Release the ended reservation only now that the body is counted in + // `sequencer_input_bytes`, mirroring the matched receipt path above, so + // the bytes are never invisible to both the limiter and the resident + // snapshot. + self.budget.release(ended_reservation); true } @@ -1792,16 +1749,6 @@ impl PeerRoutine { .release_reserved_and_return_items(unreceived_heights(outstanding)) } - fn apply_budget_delta(&mut self, delta: i128) { - if delta > 0 { - self.budget - .charge(u64::try_from(delta).expect("positive budget delta fits in u64")); - } else if delta < 0 { - self.budget - .release(u64::try_from(-delta).expect("negative budget delta fits in u64")); - } - } - /// Publish this peer's current *unreceived* in-flight height metadata to the /// registry, so the producer's `!has_outstanding_request` filter and the /// low-water `total_unreceived` gate read the same per-request-granularity @@ -2282,7 +2229,7 @@ mod tests { use super::super::peer_registry::PeerRegistry; use super::super::request::BlockSizeEstimate; - use super::super::sequencer_task::{initial_view, SequencerControlInput}; + use super::super::sequencer_task::initial_view; use super::super::state::{ByteBudget, ThroughputMeter}; use super::super::work_queue::WorkQueue; use super::super::{BlockSyncFrontiers, BlockSyncPeerSession, ZakuraBlockSyncConfig}; @@ -2291,19 +2238,17 @@ mod tests { use crate::zakura::trace::ZakuraTrace; use crate::zakura::ZakuraPeerId; - /// A floor request whose byte reservation cannot be met must still reach the - /// sequencer's floor-funding path so the rescue shed can free room — even when the - /// byte budget is *exactly* full. + /// A floor request whose byte reservation cannot be met overdrafts the in-flight + /// budget and is sent immediately — even when the byte budget is *exactly* full. /// - /// Regression guard for the wedge where `try_fill`'s floor arm sized its take by - /// `budget.available()`: at `available() == 0` the take came back empty, the fill - /// loop broke, and `reserve_request_budget` (the only caller that emits - /// `FundFloorReservation`) was never reached — so the shed that would rescue the - /// floor never fired and the floor wedged permanently. The fix sizes the floor take - /// by one response and lets the reservation shed; here we assert the funding request - /// is emitted with a non-zero need. + /// Regression guard for two prior wedges: (a) `try_fill`'s floor arm once sized + /// its take by `budget.available()`, so at `available() == 0` the take came back + /// empty and the floor wedged permanently; (b) the floor funding used to round-trip + /// a `FundFloorReservation` through the sequencer task, so a wedged task deadlocked + /// the floor. The overdraft path has no cross-task dependency: the request goes out + /// synchronously and the budget records a bounded (one-response) overshoot. #[tokio::test] - async fn exhausted_budget_floor_request_still_reaches_the_funding_path() { + async fn floor_overdraft_is_bounded_and_immediate() { let config = ZakuraBlockSyncConfig::default(); // A byte budget reserved down to exactly zero free: the case that used to wedge. @@ -2324,13 +2269,12 @@ mod tests { ); let cancel = CancellationToken::new(); - let (out_send, _out_recv) = framed_channel(16); + let (out_send, mut out_recv) = framed_channel(16); let (_in_send, in_recv) = framed_channel(16); let peer = ZakuraPeerId::new(vec![7u8; 32]).expect("test peer id is within bounds"); let session = BlockSyncPeerSession::for_test(peer.clone(), out_send, cancel.clone()); let (sequencer_input_tx, _sequencer_input_rx) = mpsc::channel(16); - let (control_tx, mut control_rx) = mpsc::unbounded_channel(); let (actions_tx, _actions_rx) = mpsc::channel(16); let (routine_to_reactor_tx, _routine_to_reactor_rx) = mpsc::channel(16); let (_view_tx, view_rx) = watch::channel(initial_view(BlockSyncFrontiers { @@ -2345,13 +2289,12 @@ mod tests { in_recv, config, 0, - budget, - work, + budget.clone(), + work.clone(), Arc::new(PeerRegistry::new()), Arc::new(Mutex::new(ThroughputMeter::new(Instant::now()))), sequencer_input_tx, Arc::new(AtomicU64::new(0)), - control_tx, actions_tx, routine_to_reactor_tx, view_rx, @@ -2364,42 +2307,41 @@ mod tests { routine.servable_low = block::Height(1); routine.servable_high = block::Height(10); - let fill = tokio::spawn(async move { - routine.try_fill().await; - }); + routine.try_fill().await; - let message = timeout(Duration::from_secs(5), control_rx.recv()) + // The floor request went out synchronously (no funding round trip)… + let frame = timeout(Duration::from_secs(5), out_recv.recv()) .await - .expect("an exhausted floor must request funding within the timeout") - .expect("the sequencer-control channel stays open"); - match message { - SequencerControlInput::FundFloorReservation { - needed_bytes, - reply, - } => { - assert!( - needed_bytes > 0, - "the floor reservation funds a non-zero request", - ); - // No reorder body to shed in this unit-level test: deny the funding. The - // routine returns the taken floor height and exits the pass cleanly. - let _ = reply.send(false); - } - other => panic!("expected FundFloorReservation, got {other:?}"), - } - - fill.await - .expect("try_fill completes after the funding decision"); + .expect("the floor GetBlocks is sent within the timeout"); + assert!( + frame.is_some(), + "an exhausted budget must not block the floor request", + ); + // …and the budget recorded a bounded overdraft: exactly the floor request's + // marked size-estimate reservation past the configured maximum. + let marked_estimate = work.reserved_bytes(); + assert!( + marked_estimate > 0, + "the floor request marked a reservation" + ); + assert_eq!( + budget.reserved(), + 8_192 + marked_estimate, + "the floor reservation overdrafts by one request's estimate", + ); + assert!( + !work.pending_contains(block::Height(1)), + "the floor height was taken, not returned", + ); } /// First-completion-wins can settle a height a routine still owns to `Held` when - /// a competing peer delivers it first. This routine's teardown (`Drop`) must be - /// Held-aware: the held body is owned by the Sequencer, so `Drop` must neither - /// release its bytes a second time (the Sequencer releases them on commit) nor - /// re-queue a body already in the commit pipeline. The pre-fix `Drop` used - /// `release_and_return_items`, which for a `Held(actual)` height returned - /// `actual` — double-releasing the `ByteBudget` and re-queuing the height into - /// `pending`. + /// a competing peer delivers it first. This routine's teardown (`Drop`) must + /// skip a height whose reservation already ended: the winner owns the release + /// (after its forward) and the body is in the commit pipeline, so `Drop` must + /// neither release the estimate a second time nor re-queue the height. The + /// pre-fix `Drop` used `release_and_return_items`, which double-released the + /// `ByteBudget` and re-queued the height into `pending`. #[tokio::test] async fn routine_drop_leaves_a_body_won_by_another_peer_to_the_sequencer() { let config = ZakuraBlockSyncConfig::default(); @@ -2428,7 +2370,6 @@ mod tests { let session = BlockSyncPeerSession::for_test(peer.clone(), out_send, cancel.clone()); let (sequencer_input_tx, _sequencer_input_rx) = mpsc::channel(16); - let (control_tx, _control_rx) = mpsc::unbounded_channel(); let (actions_tx, _actions_rx) = mpsc::channel(16); let (routine_to_reactor_tx, _routine_to_reactor_rx) = mpsc::channel(16); let (_view_tx, view_rx) = watch::channel(initial_view(BlockSyncFrontiers { @@ -2449,7 +2390,6 @@ mod tests { Arc::new(Mutex::new(ThroughputMeter::new(Instant::now()))), sequencer_input_tx, Arc::new(AtomicU64::new(0)), - control_tx, actions_tx, routine_to_reactor_tx, view_rx, @@ -2473,13 +2413,13 @@ mod tests { assert_eq!(budget_probe.reserved(), 1_000); assert_eq!(routine.window.outstanding.len(), 1); - // A competing peer delivers height 1 first: settle the shared reservation to - // `Held(actual)`. The estimate matches the actual, so the budget is unchanged - // and now holds the body's actual bytes. - let delta = work - .settle_active_reserved_height(block::Height(1), 1_000) + // A competing peer delivers height 1 first: its receipt ends the shared + // request reservation. The winner releases the estimate to the ByteBudget + // only after its forward, so it is still charged here. + let estimate = work + .release_active_reserved_height(block::Height(1)) .expect("height 1 still owns its active reservation"); - assert_eq!(delta, 0); + assert_eq!(estimate, 1_000); assert_eq!(budget_probe.reserved(), 1_000); // Tear the routine down while it still lists height 1 as unreceived. `Drop` diff --git a/zebra-network/src/zakura/block_sync/reactor.rs b/zebra-network/src/zakura/block_sync/reactor.rs index 2a2a4f66807..593aba55f46 100644 --- a/zebra-network/src/zakura/block_sync/reactor.rs +++ b/zebra-network/src/zakura/block_sync/reactor.rs @@ -65,11 +65,10 @@ pub fn spawn_block_sync_reactor( let (lifecycle_tx, lifecycle_rx) = mpsc::unbounded_channel(); // Size the action channel so the Sequencer can dispatch a full checkpoint // window of `SubmitBlock`s (`submitted_apply_limit`) plus the query/misbehavior - // spare pool. The byte - // budget — not this channel — bounds in-flight body memory, and `SubmitBlock` - // only carries an `Arc` already accounted in `applying`, so the larger - // channel costs negligible memory while removing a head-of-line stall that - // throttled body intake behind commit submission. + // spare pool. The resident look-ahead gate — not this channel — bounds body + // memory, and `SubmitBlock` only carries an `Arc` already accounted in + // `applying`, so the larger channel costs negligible memory while removing a + // head-of-line stall that throttled body intake behind commit submission. let actions_capacity = startup .config .submitted_apply_limit() @@ -133,7 +132,6 @@ pub fn spawn_block_sync_reactor( received_throughput: state.received_throughput.clone(), sequencer_input: sequencer_input_tx.clone(), sequencer_input_bytes: sequencer_input_bytes.clone(), - sequencer_control: sequencer_control_tx.clone(), actions: actions_tx.clone(), routine_to_reactor: routine_to_reactor_tx, view: sequencer_view_rx.clone(), @@ -807,10 +805,9 @@ impl BlockSyncReactor { } else if tip_advanced { // A non-destructive frontier advance does NOT respawn and does NOT // proactively drop outstanding through the tip: a still-open request - // for a now-committed height releases its budget on delivery (Sequencer - // `Redundant`) or on its own timeout, so there is no leak, only a - // slightly later release. Trace the change - // only when the tip actually moved. + // for a now-committed height releases its reservation on delivery or + // on its own timeout, so there is no leak, only a slightly later + // release. Trace the change only when the tip actually moved. self.trace_frontiers_changed(view.verified_tip); } self.prune_needed_below_floor(); @@ -1093,8 +1090,8 @@ impl BlockSyncReactor { local_frontier: Option, ) { // The whole commit-pipeline body (token validate, embedded local-frontier - // advance, applying removal, budget release, throughput record, rollback + - // misbehavior, drain + submit) runs on the Sequencer task. The reactor + // advance, applying removal, throughput record, rollback + misbehavior, + // drain + submit) runs on the Sequencer task. The reactor // forwards the completion and reacts to the resulting progress view // (serving/status/query/schedule) on the `view` arm. self.trace_apply_finished(height, token, result, self.state.budget.reserved()); @@ -1197,10 +1194,10 @@ impl BlockSyncReactor { // bounded by the in-flight byte budget and per-peer slots, not by how fast // commit/verify drains. Including reorder/applying here would pace // downloads to commit speed: a slow commit lets those buffers grow, the - // low-water gate stops refilling, and `outstanding` collapses. The byte - // budget already bounds memory because reorder/applying hold their - // reservation until apply-finish, so downloads may legitimately run far - // ahead of commit up to that budget. + // low-water gate stops refilling, and `outstanding` collapses. Memory is + // already bounded because the resident look-ahead gate counts retained + // reorder/applying bodies, so downloads may legitimately run far ahead of + // commit up to that budget. self.state .work_queue .pending_len() @@ -1662,6 +1659,13 @@ impl BlockSyncReactor { if let Some(start) = self.state.work_queue.min_pending() { bs_insert_height(row, bs_trace::QUEUE_MIN_START, start); } + // Structure-size diagnostics for memory-growth attribution: if RSS + // climbs while the byte pools stay bounded, the growing map names + // itself here. + let (diag_peers, diag_outstanding, diag_parked) = self.registry.diagnostic_sizes(); + bs_insert_u64(row, "diag_registry_peers", diag_peers); + bs_insert_u64(row, "diag_registry_outstanding", diag_outstanding); + bs_insert_u64(row, "diag_registry_parked", diag_parked); bs_insert_u64( row, bs_trace::ASSIGNED_LEN, diff --git a/zebra-network/src/zakura/block_sync/reorder.rs b/zebra-network/src/zakura/block_sync/reorder.rs index 33ff6b73ac8..def12637989 100644 --- a/zebra-network/src/zakura/block_sync/reorder.rs +++ b/zebra-network/src/zakura/block_sync/reorder.rs @@ -22,13 +22,6 @@ impl ReorderBuffer { self.blocks.len() } - /// Highest buffered height, if any. The shed-for-floor-starvation path drops - /// this (the body furthest from the committed floor) to free budget for a - /// lower, commit-unblocking request. - pub(super) fn max_height(&self) -> Option { - self.blocks.keys().next_back().copied() - } - pub(super) fn contains(&self, height: block::Height) -> bool { self.blocks.contains_key(&height) } @@ -41,12 +34,10 @@ impl ReorderBuffer { self.blocks.get(&height).map(|buffered| buffered.hash) } - /// Buffer a received body that already owns its `bytes` reservation. + /// Buffer a received body with its wire `bytes` size. /// - /// The caller reserved worst-case bytes for this height at send time and - /// shrank that reservation to `bytes` on receipt, so the reorder buffer takes - /// ownership of the existing reservation without touching the budget and can - /// never fail on budget. A `Duplicate` height is left to the caller to release. + /// Retained bodies carry no wire-budget charge (the resident look-ahead gate + /// bounds them via `buffered_bytes`), so an insert can never fail on budget. #[cfg(test)] pub(super) fn insert( &mut self, @@ -118,10 +109,8 @@ impl ReorderBuffer { released } - /// Drop every buffered body and return the total bytes they held, so the - /// caller releases exactly that reservation. The reorder buffer is owned by - /// the `Sequencer`, which does not touch the byte budget; it returns the - /// freed bytes to the reactor instead. + /// Drop every buffered body and return the total bytes they held (the + /// retained-size accounting the resident view reads; not a budget charge). pub(crate) fn clear(&mut self) -> u64 { self.drop_from(block::Height::MIN) } diff --git a/zebra-network/src/zakura/block_sync/sequencer.rs b/zebra-network/src/zakura/block_sync/sequencer.rs index e510ff25a0d..9907b4d60a7 100644 --- a/zebra-network/src/zakura/block_sync/sequencer.rs +++ b/zebra-network/src/zakura/block_sync/sequencer.rs @@ -5,9 +5,10 @@ //! never touches download-side state — the byte budget, the work scheduler, //! peers, emitted actions, or state queries. Two rules keep that boundary clean: //! -//! - every method that frees reserved bytes *returns* the freed count, so the -//! reactor releases it against the budget (the budget stays in the reactor for -//! the budget is shared), and +//! - every method that drops held bodies *returns* the freed byte count. Held +//! bodies carry no wire-budget charge (retention is bounded by the resident +//! look-ahead gate), so the count is accounting for callers and tests, not a +//! budget-release obligation, and //! - every download-side consequence (mark a height covered, clear covered, //! re-query, attribute misbehavior) is expressed as a value the reactor acts //! on, not performed here. @@ -34,13 +35,13 @@ pub(super) struct ApplyingBlock { /// Outcome of offering a received body to the commit pipeline. #[derive(Clone, Debug, Eq, PartialEq)] pub(super) enum AcceptOutcome { - /// The body was buffered and now owns its byte reservation. The reactor must - /// mark `covered` covered in the download scheduler so the retry path stops - /// re-requesting it. + /// The body was buffered. The reactor must mark `covered` covered in the + /// download scheduler so the retry path stops re-requesting it. Buffered { covered: block::Height }, /// The body was not buffered (already at/below the floor, held elsewhere in - /// the commit pipeline, or a duplicate). The reactor must release the - /// `release_bytes` the body still reserved. + /// the commit pipeline, or a duplicate). `release_bytes` is the dropped + /// body's size; held bodies carry no wire-budget charge, so it is + /// informational. Redundant { release_bytes: u64 }, } @@ -57,7 +58,9 @@ pub(super) struct SubmitItem { /// Sequencer half of a verified-tip advance (frontier growth/commit). #[derive(Copy, Clone, Debug)] pub(super) struct AdvanceOutcome { - /// Bytes freed from the reorder/applying buffers that the reactor releases. + /// Bytes freed from the reorder/applying buffers (informational; held bodies + /// carry no wire-budget charge). + #[cfg_attr(not(test), allow(dead_code))] // asserted by sequencer unit tests pub(super) release_bytes: u64, /// Whether the verified tip actually moved. The reactor drops download state /// (scheduler/outstanding) and re-drains only when it did. @@ -156,11 +159,6 @@ impl Sequencer { self.reorder.buffered_bytes() } - /// Highest buffered reorder height, for shed-for-floor-starvation. - pub(super) fn reorder_max_height(&self) -> Option { - self.reorder.max_height() - } - pub(super) fn unsubmitted_applying_count(&self) -> usize { // Derived: every applying body is either submitted or not. self.applying @@ -243,8 +241,8 @@ impl Sequencer { // ---- body acceptance ---- /// Offer a received body to the commit pipeline. Runs the redundancy checks - /// and (when not redundant) buffers it in the reorder buffer, which takes - /// ownership of the body's existing `bytes` reservation. + /// and (when not redundant) buffers it in the reorder buffer with its wire + /// `bytes` for the retained-size accounting. #[cfg(test)] pub(super) fn accept_body( &mut self, diff --git a/zebra-network/src/zakura/block_sync/sequencer_task.rs b/zebra-network/src/zakura/block_sync/sequencer_task.rs index 9bb71d2cdf1..cf2acc6e830 100644 --- a/zebra-network/src/zakura/block_sync/sequencer_task.rs +++ b/zebra-network/src/zakura/block_sync/sequencer_task.rs @@ -25,69 +25,6 @@ use super::{ *, }; -/// Favor the lowest needed height over the speculative high tail. -/// -/// While the byte budget cannot fund even one worst-case request yet the lowest -/// needed height (pending or outstanding) sits *below* the highest buffered body, -/// drop that top body: release its bytes to the budget and return its height to -/// `pending` (it was held, hence in `work.in_flight` per the `held ⟺ in_flight` -/// invariant) for later re-fetch. Because another top can always be shed, a low -/// retry never blocks on budget; the floor can never wedge behind a full buffer, -/// and under a stall the speculative tail is shed and the chain fills bottom-up, -/// which also bounds the reorder backlog. The rescue path is purely demand-driven: -/// it runs inline after each accepted body and synchronously when a floor requester -/// needs budget through [`SequencerControlInput::FundFloorReservation`]. Returns -/// whether it shed anything. -pub(super) fn shed_top_until_available( - budget: &mut ByteBudget, - work: &WorkQueue, - sequencer: &mut Sequencer, - target_available: u64, -) -> bool { - let mut shed_any = false; - while budget.available() < target_available { - let lowest_needed = match (work.min_pending(), work.min_in_flight()) { - (Some(pending), Some(in_flight)) => pending.min(in_flight), - (Some(pending), None) => pending, - (None, Some(in_flight)) => in_flight, - (None, None) => break, - }; - let Some(top) = sequencer.reorder_max_height() else { - break; - }; - // Only shed a body that sits above a starved lower height: we trade a - // far-from-floor body for the ability to fetch a nearer, higher-value one. - if lowest_needed >= top { - break; - } - let freed = sequencer.drop_reorder_from(top); - if freed == 0 { - break; - } - let released = work.release_and_return_items([top]); - debug_assert!( - released == 0 || released == freed, - "shed reorder release must match the per-height budget ledger when present" - ); - budget.release(if released == 0 { freed } else { released }); - shed_any = true; - } - shed_any -} - -pub(super) fn shed_top_for_floor_starvation( - budget: &mut ByteBudget, - work: &WorkQueue, - sequencer: &mut Sequencer, -) -> bool { - shed_top_until_available( - budget, - work, - sequencer, - super::config::BS_PER_BLOCK_WORST_CASE_BYTES, - ) -} - /// A received body a peer routine matched (or accepted unmatched) and forwards /// to the commit pipeline. This is the only bounded Sequencer input: a slow /// verifier can backpressure body intake, but must not block apply/frontier @@ -105,9 +42,9 @@ pub(super) struct SequencedBody { /// Progress-critical Sequencer events forwarded by the reactor. /// /// These events must not sit behind downloaded bodies. `ApplyFinished` releases -/// budget and verifier slots, and frontier/reset events can release or discard -/// stale body work. They are locally generated and tiny, so they use a separate -/// unbounded channel and are prioritized by the Sequencer task. +/// retained bodies and verifier slots, and frontier/reset events can release or +/// discard stale body work. They are locally generated and tiny, so they use a +/// separate unbounded channel and are prioritized by the Sequencer task. #[derive(Debug)] pub(super) enum SequencerControlInput { /// A verified-tip advance (frontier growth/commit). @@ -137,12 +74,6 @@ pub(super) enum SequencerControlInput { result: BlockApplyResult, local_frontier: Option, }, - /// Synchronously pop the speculative high tail until a floor request can - /// reserve `needed_bytes`, then wake the requester to retry the reservation. - FundFloorReservation { - needed_bytes: u64, - reply: oneshot::Sender, - }, } /// The progress view the reactor reacts to. A `watch` (latest-wins) send never @@ -286,11 +217,6 @@ impl SequencerTask { Some(body) => { self.release_body_input_bytes(body.bytes); self.handle_accept_body(body).await; - shed_top_for_floor_starvation( - &mut self.budget, - &self.work, - &mut self.sequencer, - ); self.publish_view(); } None => body_open = false, @@ -339,19 +265,6 @@ impl SequencerTask { self.handle_apply_finished(token, height, hash, result, local_frontier) .await } - SequencerControlInput::FundFloorReservation { - needed_bytes, - reply, - } => { - let shed = shed_top_until_available( - &mut self.budget, - &self.work, - &mut self.sequencer, - needed_bytes, - ); - let _ = reply.send(self.budget.available() >= needed_bytes); - shed - } } } @@ -363,8 +276,8 @@ impl SequencerTask { ); } - /// Body-acceptance tail: offer the body to the reorder buffer, release its - /// bytes on a `Redundant` outcome, then drain the ready contiguous prefix into + /// Body-acceptance tail: offer the body to the reorder buffer (a `Redundant` + /// body is simply dropped), then drain the ready contiguous prefix into /// applying and submit it. async fn handle_accept_body(&mut self, body: SequencedBody) { let queued_elapsed = body.received_at.elapsed(); @@ -376,10 +289,9 @@ impl SequencerTask { body.peer, ) { AcceptOutcome::Buffered { .. } => "buffered", - AcceptOutcome::Redundant { release_bytes } => { - self.budget.release(release_bytes); - "redundant" - } + // A dropped redundant body frees no wire budget: its request + // reservation was already released at receipt. + AcceptOutcome::Redundant { .. } => "redundant", }; self.trace_body_accepted(body.height, queued_elapsed, outcome); self.release_contiguous_blocks().await; @@ -403,10 +315,12 @@ impl SequencerTask { return; } self.verified_block_hash = frontiers.verified_block_hash; + // `advance_verified_tip` drops superseded bodies; those never charged the + // wire budget, so only the work queue's still-reserved request estimates + // (unreceived heights GC'd by the floor) are returned to it. let advance = self .sequencer .advance_verified_tip(frontiers.verified_block_tip, release_applied); - self.budget.release(advance.release_bytes); if advance.changed { let released = self.work.advance_floor(frontiers.verified_block_tip); self.budget.release(released); @@ -479,15 +393,14 @@ impl SequencerTask { self.verified_block_hash = frontiers.verified_block_hash; // The Sequencer pins its verified tip and floor to the reset target and - // clears the reorder/applying buffers, returning the freed bytes for - // release. - let released = self + // clears the reorder/applying buffers. Their body bytes never charged + // the wire budget; nothing to release for them. + let _ = self .sequencer .reset_to(frontiers.verified_block_tip, remember_released_applies); - self.budget.release(released); // Drop every download work item above the reset target (their buffers // were cleared by `reset_to`); the reactor's `query_needed_blocks` - // re-fills. + // re-fills. Their unreceived request reservations return to the budget. let released = self.work.reset_above(self.sequencer.floor()); self.budget.release(released); // A destructive reset: bump the epoch so the reactor drops *all* @@ -495,7 +408,7 @@ impl SequencerTask { self.reset_epoch = self.reset_epoch.saturating_add(1); } - /// Handle a verifier apply completion: release the body's bytes and verifier + /// Handle a verifier apply completion: release the body and its verifier /// slot, fold in any embedded `local_frontier` as a frontier advance with /// `release_applied: false`, and on a rejection roll the floor back below the /// bad block so its range is re-requestable. Returns whether the reactor needs @@ -546,7 +459,6 @@ impl SequencerTask { .remove_applying(height) .expect("applying entry exists because it was just checked"); - self.budget.release(applying.bytes); // A `Committed` result is a body that newly extended the chain; count it // toward commit throughput (the apply rate the download path is racing). if matches!(result, BlockApplyResult::Committed) { @@ -562,13 +474,13 @@ impl SequencerTask { // reorder), roll the floor back below it, and drop the WorkQueue // entries above the rolled-back floor so the heights are // re-requestable (the reactor's `query_needed_blocks` re-fills). - let released = self.sequencer.release_applying_blocks_from(height); - self.budget.release(released); + // Only the work queue's unreceived request reservations return + // to the wire budget; the dropped bodies never charged it. + let _ = self.sequencer.release_applying_blocks_from(height); self.sequencer.reset_floor_below(height); let released = self.work.reset_above(self.sequencer.floor()); self.budget.release(released); - let dropped = self.sequencer.drop_reorder_from(height); - self.budget.release(dropped); + let _ = self.sequencer.drop_reorder_from(height); // A `Rejected` result means consensus found the body invalid. // Attribute it to the delivering peer so repeat offenders are // scored and eventually disconnected. `TimedOut` is a local apply @@ -584,10 +496,9 @@ impl SequencerTask { BlockApplyResult::Rejected | BlockApplyResult::TimedOut => {} } if let Some(frontiers) = accepted_local_frontier { - let released = self + let _ = self .sequencer .release_applied_through(frontiers.verified_block_tip); - self.budget.release(released); } self.release_contiguous_blocks().await; @@ -735,20 +646,13 @@ impl SequencerTask { self.committed_throughput.sample(Instant::now()); let reorder_buffered_bytes = self.sequencer.reorder_buffered_bytes(); let applying_buffered_bytes = self.sequencer.applying_buffered_bytes(); - let body_input_bytes = self - .body_input_bytes - .load(std::sync::atomic::Ordering::Relaxed); // Cross-layer drift check: the independently-maintained `ByteBudget` total - // must equal the sum of the component counters. `work.reserved_bytes()` is - // now an O(1) counter, so this runs on every event without a work-queue scan. - let expected_budget = self - .work - .reserved_bytes() - .saturating_add(reorder_buffered_bytes) - .saturating_add(applying_buffered_bytes) - .saturating_add(body_input_bytes); + // must cover the work queue's outstanding-request reservations (retained + // bodies never charge it; a transient excess is expected receipt-handoff + // skew, see `ByteBudget::audit`). `work.reserved_bytes()` is an O(1) + // counter, so this runs on every event without a work-queue scan. self.budget - .audit(expected_budget, "block-sync sequencer view"); + .audit(self.work.reserved_bytes(), "block-sync sequencer view"); let _ = self.view_tx.send_replace(SequencerView { verified_tip: self.sequencer.verified_tip(), verified_hash: self.verified_block_hash, diff --git a/zebra-network/src/zakura/block_sync/service.rs b/zebra-network/src/zakura/block_sync/service.rs index 11f342f3706..ba731408b19 100644 --- a/zebra-network/src/zakura/block_sync/service.rs +++ b/zebra-network/src/zakura/block_sync/service.rs @@ -485,7 +485,6 @@ impl Service for BlockSyncService { wiring.received_throughput, wiring.sequencer_input, wiring.sequencer_input_bytes, - wiring.sequencer_control, wiring.actions, wiring.routine_to_reactor, wiring.view, diff --git a/zebra-network/src/zakura/block_sync/state.rs b/zebra-network/src/zakura/block_sync/state.rs index e00f5123d2b..4ee0f1de8b2 100644 --- a/zebra-network/src/zakura/block_sync/state.rs +++ b/zebra-network/src/zakura/block_sync/state.rs @@ -153,8 +153,6 @@ pub(super) struct RoutineWiring { pub(super) received_throughput: Arc>, pub(super) sequencer_input: mpsc::Sender, pub(super) sequencer_input_bytes: Arc, - pub(super) sequencer_control: - mpsc::UnboundedSender, pub(super) actions: mpsc::Sender, pub(super) routine_to_reactor: mpsc::Sender, pub(super) view: watch::Receiver, @@ -815,10 +813,9 @@ pub(super) struct DeliverySnapshot { impl OutstandingBlockRange { /// Bytes still reserved for this request: the sum of the per-height size - /// estimates for every requested height not yet received. Each received body - /// shrinks its estimate toward the actual size, so releasing this (on - /// timeout/disconnect/short response) never over-releases bytes already handed - /// to the reorder buffer. + /// estimates for every requested height not yet received. Received heights + /// released their estimate at receipt, so releasing this (on + /// timeout/disconnect/short response) never double-releases their share. pub(super) fn reserved_bytes(&self) -> u64 { self.request .expected_blocks @@ -876,12 +873,12 @@ impl OutstandingBlockRange { /// Pure per-height byte-accounting state. /// /// The shared [`ByteBudget`] is just the atomic sink. This ledger owns the -/// lifecycle arithmetic for one requested height: -/// `Reserved(estimate) -> Held(actual) -> Released`. +/// lifecycle of one height's in-flight request-estimate reservation: +/// `Reserved(estimate) -> Released`. Received bodies carry no charge here — +/// retention is bounded by the resident look-ahead gate, not the wire budget. #[derive(Copy, Clone, Debug, Eq, PartialEq)] pub(super) enum BlockBudgetLedger { Reserved(u64), - Held(u64), Released, } @@ -890,26 +887,10 @@ impl BlockBudgetLedger { Self::Reserved(estimate) } - pub(super) fn current_charge(self) -> u64 { - match self { - Self::Reserved(bytes) | Self::Held(bytes) => bytes, - Self::Released => 0, - } - } - - pub(super) fn release_reserved(&mut self) -> u64 { - let released = match *self { - Self::Reserved(bytes) => bytes, - Self::Held(_) | Self::Released => 0, - }; - *self = Self::Released; - released - } - pub(super) fn reserved_charge(self) -> u64 { match self { Self::Reserved(bytes) => bytes, - Self::Held(_) | Self::Released => 0, + Self::Released => 0, } } @@ -917,25 +898,11 @@ impl BlockBudgetLedger { matches!(self, Self::Reserved(_)) } - /// Move a reserved height to held bytes and return the signed budget delta. - /// - /// Positive means charge more bytes; negative means release bytes. - pub(super) fn settle(&mut self, actual: u64) -> i128 { - match *self { - Self::Reserved(reserved) => { - *self = Self::Held(actual); - i128::from(actual) - i128::from(reserved) - } - Self::Released => 0, - Self::Held(_) => 0, - } - } - - /// Release the current charge exactly once. - pub(super) fn release(&mut self) -> u64 { - let charge = self.current_charge(); + /// Release the reserved estimate exactly once, returning the released bytes. + pub(super) fn release_reserved(&mut self) -> u64 { + let released = self.reserved_charge(); *self = Self::Released; - charge + released } } diff --git a/zebra-network/src/zakura/block_sync/tests.rs b/zebra-network/src/zakura/block_sync/tests.rs index c19999495fa..4ba36ad72db 100644 --- a/zebra-network/src/zakura/block_sync/tests.rs +++ b/zebra-network/src/zakura/block_sync/tests.rs @@ -995,10 +995,9 @@ fn config_validate_rejects_degenerate_values() { }; assert!(config.validate().is_err()); - // A positive budget below the checkpoint-range floor is no longer rejected by - // `validate`: it is clamped up to the floor (with a warning) at load instead, - // so older configs keep starting. See - // `config_clamps_below_floor_inflight_block_bytes`. + // A request budget below the checkpoint-range floor is fine: submitted bodies + // release their reservation at receipt, so the budget only has to exceed one + // floor request (checked below the floor-request bound in `validate`). config = ZakuraBlockSyncConfig { max_inflight_block_bytes: BS_CHECKPOINT_RANGE_BYTE_FLOOR - 1, ..ZakuraBlockSyncConfig::default() @@ -1038,58 +1037,45 @@ fn config_validate_rejects_degenerate_values() { } #[test] -fn config_clamps_below_floor_inflight_block_bytes() { - // A positive budget below the checkpoint-range floor is clamped up to the - // floor so checkpoint sync cannot deadlock (instead of refusing to start). - let mut below = ZakuraBlockSyncConfig { - // 256 MiB, the historical `v4.5.0-zakura-blocksync.toml` value, which is - // below the ~802 MB checkpoint-range floor. - max_inflight_block_bytes: 256 * 1024 * 1024, - ..ZakuraBlockSyncConfig::default() - }; - assert!(below.max_inflight_block_bytes < BS_CHECKPOINT_RANGE_BYTE_FLOOR); - below.clamp_inflight_block_bytes_to_floor(); - assert_eq!( - below.max_inflight_block_bytes, - BS_CHECKPOINT_RANGE_BYTE_FLOOR - ); - - // A budget at or above the floor is left untouched. - let mut at_floor = ZakuraBlockSyncConfig { - max_inflight_block_bytes: BS_CHECKPOINT_RANGE_BYTE_FLOOR + 1, - ..ZakuraBlockSyncConfig::default() - }; - at_floor.clamp_inflight_block_bytes_to_floor(); - assert_eq!( - at_floor.max_inflight_block_bytes, - BS_CHECKPOINT_RANGE_BYTE_FLOOR + 1 - ); - - // Zero is left untouched so `validate` still rejects it as a misconfiguration. - let mut zero = ZakuraBlockSyncConfig { - max_inflight_block_bytes: 0, - ..ZakuraBlockSyncConfig::default() - }; - zero.clamp_inflight_block_bytes_to_floor(); - assert_eq!(zero.max_inflight_block_bytes, 0); - assert!(zero.validate().is_err()); +fn config_deserialize_keeps_below_range_floor_inflight_block_bytes() { + // An older config with a small `max_inflight_block_bytes` (e.g. the stored + // `v4.5.0-zakura-blocksync.toml` 256 MiB value) loads unchanged: the request + // budget no longer bounds retention, so it needs no checkpoint-range clamp — + // only `validate`'s one-floor-request lower bound applies. + let config: crate::Config = toml::from_str( + r#" + [zakura.block_sync] + max_inflight_block_bytes = 268435456 + "#, + ) + .expect("a below-range-floor max_inflight_block_bytes config still loads"); + const { + assert!(268435456u64 < BS_CHECKPOINT_RANGE_BYTE_FLOOR); + } + assert_eq!(config.zakura.block_sync.max_inflight_block_bytes, 268435456); } #[test] -fn config_deserialize_clamps_below_floor_inflight_block_bytes() { - // Regression: an older config with a too-small `max_inflight_block_bytes` - // (e.g. the stored `v4.5.0-zakura-blocksync.toml`) must still load -- clamped - // up to the checkpoint-range floor -- rather than being rejected at startup. +fn config_deserialize_clamps_sub_floor_request_inflight_block_bytes() { + // A stored config whose request budget cannot cover one floor request + // (e.g. an old 16 MiB value) loads clamped to just above the floor instead + // of failing `validate`'s one-floor-request lower bound, so older nodes + // keep starting. let config: crate::Config = toml::from_str( r#" [zakura.block_sync] - max_inflight_block_bytes = 268435456 + max_inflight_block_bytes = 16777216 "#, ) - .expect("a below-floor max_inflight_block_bytes config still loads"); + .expect("a sub-floor-request max_inflight_block_bytes config still loads"); + let request_floor = config.zakura.block_sync.floor_request_byte_reservation(); + assert!( + 16_777_216 <= request_floor, + "test premise: the stored value is at or below one floor request" + ); assert_eq!( config.zakura.block_sync.max_inflight_block_bytes, - BS_CHECKPOINT_RANGE_BYTE_FLOOR, + request_floor + 1 ); } @@ -2037,20 +2023,21 @@ fn work_queue_reserved_bytes_counter_matches_scan_across_transitions() { assert_eq!(queue.reserved_bytes(), 600); check("mark_reserved"); - // settle: Reserved -> Held drops the reservation for height 1. - queue - .settle_active_reserved_height(block::Height(1), 80) - .expect("height 1 is reserved"); + // receipt: Reserved -> Released drops the reservation for height 1. + assert_eq!( + queue.release_active_reserved_height(block::Height(1)), + Some(100), + ); assert_eq!(queue.reserved_bytes(), 500); - check("settle"); + check("release_active_reserved_height"); - // mark_held_direct: Reserved -> Held drops height 2's reservation. - queue.mark_held_direct(block::Height(2), 90); + // claim_received: Reserved -> Released drops height 2's reservation. + assert_eq!(queue.claim_received(block::Height(2)), 100); assert_eq!(queue.reserved_bytes(), 400); - check("mark_held_direct"); + check("claim_received"); // release_heights: Reserved -> Released for height 3. - queue.release_heights([block::Height(3)]); + queue.release_reserved_heights([block::Height(3)]); assert_eq!(queue.reserved_bytes(), 300); check("release_heights"); @@ -2060,9 +2047,9 @@ fn work_queue_reserved_bytes_counter_matches_scan_across_transitions() { assert_eq!(queue.reserved_bytes(), 200); check("release_reserved_and_return_items"); - // advance_floor drops the committed `<= floor` prefix. Heights 1 (Held) and 2 - // (Held) contribute no reservation; height 3 is already Released. Only heights - // 5 and 6 remain reserved above the new floor. + // advance_floor drops the committed `<= floor` prefix. Heights 1 and 2 + // (received) and 3 (released) contribute no reservation. Only heights 5 and 6 + // remain reserved above the new floor. let released = queue.advance_floor(block::Height(4)); assert_eq!(released, 0, "heights <= 4 owned no live reservation"); assert!(!queue.pending_contains(block::Height(1))); @@ -2108,37 +2095,63 @@ fn work_queue_advance_floor_drops_only_committed_prefix() { } #[test] -fn watchdog_after_held_settle_releases_once() { +fn receipt_releases_wire_reservation_while_body_retained() { + // Receipt ends the wire reservation: the estimate returns to the budget while + // the caller still owns the body (the height stays claimed in `in_flight`). let queue = work_queue_with(0, [needed(1, BlockSizeEstimate::Advertised(100))]); let mut budget = ByteBudget::new(1_000); let taken = queue.take_in_range(block::Height(1), block::Height(1), 1); assert_eq!(taken.len(), 1); assert!(budget.try_reserve(100)); assert_eq!(queue.mark_reserved([block::Height(1)]), 100); + assert_eq!(queue.reserved_bytes(), 100); - let delta = queue - .settle_active_reserved_height(block::Height(1), 80) - .expect("active reserved height settles"); - assert_eq!(delta, -20); - budget.release(20); - assert_eq!(budget.reserved(), 80); + let reserved_estimate = queue + .release_active_reserved_height(block::Height(1)) + .expect("active reserved height releases at receipt"); + assert_eq!(reserved_estimate, 100); + budget.release(reserved_estimate); + assert_eq!(queue.reserved_bytes(), 0); + assert_eq!(budget.reserved(), 0); + assert!( + queue.in_flight_contains(block::Height(1)), + "the received height stays claimed; the body handoff owns it" + ); +} + +#[test] +fn floor_watchdog_skips_received_heights() { + // After receipt released the reservation, the watchdog must neither requeue + // the height (the body handoff owns it) nor release a second time. + let queue = work_queue_with(0, [needed(1, BlockSizeEstimate::Advertised(100))]); + let mut budget = ByteBudget::new(1_000); + let taken = queue.take_in_range(block::Height(1), block::Height(1), 1); + assert_eq!(taken.len(), 1); + assert!(budget.try_reserve(100)); + assert_eq!(queue.mark_reserved([block::Height(1)]), 100); + + let reserved_estimate = queue + .release_active_reserved_height(block::Height(1)) + .expect("active reserved height releases at receipt"); + budget.release(reserved_estimate); + assert_eq!(budget.reserved(), 0); let watchdog_released = queue.release_reserved_and_return_items([block::Height(1)]); budget.release(watchdog_released); assert_eq!( watchdog_released, 0, - "watchdog must not release a held body owned by the sequencer handoff" + "watchdog must not release a received height owned by the sequencer handoff" + ); + assert!( + queue.in_flight_contains(block::Height(1)), + "the received height must not be requeued for a re-fetch" ); - assert!(queue.in_flight_contains(block::Height(1))); - assert_eq!(budget.reserved(), 80); - - budget.release(80); - assert_eq!(queue.advance_floor(block::Height(1)), 0); assert_eq!(budget.reserved(), 0); + assert_eq!(queue.advance_floor(block::Height(1)), 0); } #[test] -fn late_delivery_after_watchdog_cancellation_does_not_resurrect_released_claim() { +fn late_body_does_not_resurrect_charge() { let queue = work_queue_with(0, [needed(1, BlockSizeEstimate::Advertised(100))]); let mut budget = ByteBudget::new(1_000); let taken = queue.take_in_range(block::Height(1), block::Height(1), 1); @@ -2152,15 +2165,17 @@ fn late_delivery_after_watchdog_cancellation_does_not_resurrect_released_claim() assert!(queue.pending_contains(block::Height(1))); assert_eq!( - queue.settle_active_reserved_height(block::Height(1), 80), + queue.release_active_reserved_height(block::Height(1)), None, - "a late body cannot settle a claim the watchdog already released" + "a late body cannot release a claim the watchdog already released" ); assert_eq!(budget.reserved(), 0); } #[test] -fn mark_held_direct_does_not_orphan_a_charge() { +fn claim_received_does_not_orphan_a_charge() { + // An unmatched queued body claims its height without a new reservation, but + // any stale request reservation it still owned is handed back exactly once. let queue = work_queue_with(0, [needed(1, BlockSizeEstimate::Advertised(100))]); let mut budget = ByteBudget::new(1_000); let taken = queue.take_in_range(block::Height(1), block::Height(1), 1); @@ -2168,20 +2183,19 @@ fn mark_held_direct_does_not_orphan_a_charge() { assert!(budget.try_reserve(100)); assert_eq!(queue.mark_reserved([block::Height(1)]), 100); - assert!(budget.try_reserve(80)); - let old_charge = queue.mark_held_direct(block::Height(1), 80); + let old_charge = queue.claim_received(block::Height(1)); budget.release(old_charge); assert_eq!(old_charge, 100); - assert_eq!(budget.reserved(), 80); + assert_eq!(budget.reserved(), 0); - let released = queue.release_heights([block::Height(1)]); + let released = queue.release_reserved_heights([block::Height(1)]); budget.release(released); - assert_eq!(released, 80); + assert_eq!(released, 0, "a received height owns no further charge"); assert_eq!(budget.reserved(), 0); } #[test] -fn release_reserved_mixed_reserved_held_conserves_budget() { +fn release_reserved_mixed_received_and_reserved_conserves_budget() { let queue = work_queue_with( 0, [ @@ -2198,34 +2212,33 @@ fn release_reserved_mixed_reserved_held_conserves_budget() { 200 ); - let delta = queue - .settle_active_reserved_height(block::Height(1), 80) - .expect("active height settles"); - assert_eq!(delta, -20); - budget.release(20); - assert_eq!(budget.reserved(), 180); + // Height 1 is received: its whole estimate returns to the budget. + let reserved_estimate = queue + .release_active_reserved_height(block::Height(1)) + .expect("active height releases at receipt"); + assert_eq!(reserved_estimate, 100); + budget.release(reserved_estimate); + assert_eq!(budget.reserved(), 100); + // Floor GC releases only the still-reserved (unreceived) height 2. let released = queue.advance_floor(block::Height(2)); budget.release(released); assert_eq!( released, 100, - "WorkQueue releases only the still-reserved height; held bytes are released by Sequencer" + "WorkQueue releases only the still-reserved height; received bodies hold no charge" ); - assert_eq!(budget.reserved(), 80); - - budget.release(80); assert_eq!(budget.reserved(), 0); } #[test] -fn release_reserved_heights_skips_held_body_owned_by_sequencer() { +fn release_reserved_heights_skips_received_body_owned_by_sequencer() { // The owner's routine GC / stale-trim cleanup (`gc_committed_outstanding`, // `stale_adjusted_disposition`, `finish_detached`) releases via - // `release_reserved_heights`. When a competing peer delivered a height late, it - // settled to `Held(actual)` in the shared queue; the owner must release only - // the still-reserved estimate and leave the held body for the Sequencer — never - // double-releasing its bytes. The non-Held-aware `release_heights` would return - // the held `actual` here and saturate the budget. + // `release_reserved_heights`. When a competing peer delivered a height late, + // its receipt ended that height's reservation in the shared queue; the owner + // must release only the still-reserved estimates and leave the received body + // to the winner (which releases the ended estimate after its forward) — never + // double-releasing. let queue = work_queue_with( 0, [ @@ -2242,28 +2255,28 @@ fn release_reserved_heights_skips_held_body_owned_by_sequencer() { 200 ); - // A competing peer's late body settles height 1 to Held(80); height 2 stays - // reserved (never delivered). - let delta = queue - .settle_active_reserved_height(block::Height(1), 80) + // A competing peer's late body ends height 1's reservation; height 2 stays + // reserved (never delivered). The winner releases the estimate after its + // forward. + let estimate = queue + .release_active_reserved_height(block::Height(1)) .expect("height 1 is reserved"); - assert_eq!(delta, -20); - budget.release(20); - assert_eq!(budget.reserved(), 180); + assert_eq!(estimate, 100); + budget.release(estimate); + assert_eq!(budget.reserved(), 100); - // GC both heights: only the still-reserved height 2 releases. The Held height 1 - // is skipped (owned by the Sequencer) and stays in `in_flight`. + // GC both heights: only the still-reserved height 2 releases. The received + // height 1 is skipped (its reservation already ended) and stays in `in_flight`. let released = queue.release_reserved_heights([block::Height(1), block::Height(2)]); budget.release(released); assert_eq!( released, 100, - "release_reserved_heights frees only the still-reserved estimate, never held body bytes" + "release_reserved_heights frees only the still-reserved estimate, exactly once" ); assert!(queue.in_flight_contains(block::Height(1))); - assert_eq!(budget.reserved(), 80); + assert_eq!(budget.reserved(), 0); - // The Sequencer releases the held body on commit; nothing drifts. - budget.release(80); + // Floor GC clears the bookkeeping; nothing drifts. assert_eq!(queue.advance_floor(block::Height(2)), 0); assert_eq!(budget.reserved(), 0); } @@ -3326,24 +3339,20 @@ fn work_queue_keeps_pending_ordered_by_height() { } #[test] -fn reorder_drains_only_contiguous_prefix_without_releasing_budget() { +fn reorder_drains_only_contiguous_prefix_and_reports_dropped_bytes() { + // The reorder buffer never touches the byte budget: it maintains only the + // retained-size accounting (`buffered_bytes`) the resident gate reads, and + // its drop paths report the freed bytes for that accounting. let mut reorder = ReorderBuffer::new(); - let mut budget = ByteBudget::new(10_000); let block = mainnet_block(&BLOCK_MAINNET_1_BYTES); - // The reorder buffer no longer touches the budget on insert: a received body - // already owns its (shrunk) reservation, so the caller reserves and the buffer - // only takes ownership. Model that by reserving the actual bytes here. - assert!(budget.try_reserve(300)); assert_eq!( reorder.insert(block::Height(3), block.clone(), 300, peer(0)), ReorderInsertResult::Inserted ); assert!(reorder.drain_contiguous_prefix(block::Height(0)).is_empty()); assert_eq!(reorder.buffered_bytes(), 300); - assert_eq!(budget.reserved(), 300); - assert!(budget.try_reserve(100)); assert_eq!( reorder.insert(block::Height(1), block.clone(), 100, peer(0)), ReorderInsertResult::Inserted @@ -3356,13 +3365,10 @@ fn reorder_drains_only_contiguous_prefix_without_releasing_budget() { .collect::>(), vec![(block::Height(1), 100)] ); - // Draining the contiguous prefix hands bytes to the apply stage; it does not - // release the budget, which the apply finish releases later. + // Draining the contiguous prefix hands bytes to the apply stage; height 3 + // stays buffered behind the gap at 2. assert_eq!(reorder.buffered_bytes(), 300); - assert_eq!(budget.reserved(), 400); - budget.release(100); - assert!(budget.try_reserve(200)); assert_eq!( reorder.insert(block::Height(2), block.clone(), 200, peer(0)), ReorderInsertResult::Inserted @@ -3375,29 +3381,22 @@ fn reorder_drains_only_contiguous_prefix_without_releasing_budget() { .collect::>(), vec![(block::Height(2), 200), (block::Height(3), 300)] ); - assert_eq!(budget.reserved(), 500); - budget.release(500); + assert_eq!(reorder.buffered_bytes(), 0); - assert!(budget.try_reserve(200)); assert_eq!( reorder.insert(block::Height(2), block.clone(), 200, peer(0)), ReorderInsertResult::Inserted ); - assert!(budget.try_reserve(300)); assert_eq!( reorder.insert(block::Height(3), block, 300, peer(0)), ReorderInsertResult::Inserted ); // `drop_from`/`drop_through`/`clear` return the bytes their dropped bodies - // held; the reactor releases that reservation (the reorder buffer, owned by - // the `Sequencer`, never touches the budget directly). - budget.release(reorder.drop_from(block::Height(3))); + // held, keeping `buffered_bytes` consistent for the resident view. + assert_eq!(reorder.drop_from(block::Height(3)), 300); assert_eq!(reorder.buffered_bytes(), 200); - assert_eq!(budget.reserved(), 200); - budget.release(reorder.drop_through(block::Height(2))); + assert_eq!(reorder.drop_through(block::Height(2)), 200); assert_eq!(reorder.buffered_bytes(), 0); - assert_eq!(budget.reserved(), 0); - assert!(budget.try_reserve(300)); assert_eq!( reorder.insert( block::Height(3), @@ -3407,182 +3406,8 @@ fn reorder_drains_only_contiguous_prefix_without_releasing_budget() { ), ReorderInsertResult::Inserted ); - budget.release(reorder.clear()); + assert_eq!(reorder.clear(), 300); assert_eq!(reorder.buffered_bytes(), 0); - assert_eq!(budget.reserved(), 0); -} - -#[test] -fn shed_top_for_floor_starvation_funds_lowest_pending_by_dropping_top() { - let worst = BS_PER_BLOCK_WORST_CASE_BYTES; - // Budget holds exactly two worst-case blocks, both consumed by buffered bodies - // (heights 5 and 6). Height 1 — the commit-unblocking floor gap — is pending - // and unfunded: this is the "download ran ahead, budget full, low height - // needs (re-)requesting" shape that wedged sync. - let mut budget = ByteBudget::new(2 * worst); - let work = work_queue_with( - 0, - [ - needed(1, BlockSizeEstimate::Unknown), - needed(5, BlockSizeEstimate::Unknown), - needed(6, BlockSizeEstimate::Unknown), - ], - ); - let mut seq = test_sequencer(0, 100); - let block = mainnet_block(&BLOCK_MAINNET_1_BYTES); - - // Heights 5 and 6 are taken (pending -> in_flight) and buffered, each holding a - // worst-case block of budget; height 1 stays pending and unfunded. - work.take_in_range(block::Height(5), block::Height(6), 2); - for height in [5u32, 6] { - assert!(budget.try_reserve(worst)); - seq.accept_body( - block::Height(height), - block::Hash([height as u8; 32]), - block.clone(), - worst, - peer(0), - ); - } - assert_eq!(budget.available(), 0, "budget saturated by buffered bodies"); - assert!(work.pending_contains(block::Height(1))); - - // The floor-reservation rescue drops the top buffered body (6) — the - // one furthest from the floor — releasing its budget and returning its height - // to `pending` for later re-fetch, so the lower floor-gap request can now be - // funded. Without this the budget stays full and height 1 can never be - // requested (the wedge). - let shed = super::sequencer_task::shed_top_for_floor_starvation(&mut budget, &work, &mut seq); - assert!(shed, "the top buffered body is shed"); - assert!( - budget.available() >= worst, - "freed budget can now fund the floor-gap request" - ); - assert!( - !seq.reorder_contains(block::Height(6)), - "the top body is evicted" - ); - assert!( - seq.reorder_contains(block::Height(5)), - "the lower buffered body is kept" - ); - assert!( - work.pending_contains(block::Height(1)), - "the floor gap is still pending and now fundable" - ); - assert!( - work.pending_contains(block::Height(6)), - "the evicted height is returned to pending for re-fetch" - ); -} - -#[test] -fn shed_top_for_floor_starvation_funds_outstanding_floor_by_dropping_top() { - let worst = BS_PER_BLOCK_WORST_CASE_BYTES; - let mut budget = ByteBudget::new(2 * worst); - let work = work_queue_with( - 0, - [ - needed(1, BlockSizeEstimate::Unknown), - needed(5, BlockSizeEstimate::Unknown), - needed(6, BlockSizeEstimate::Unknown), - ], - ); - let mut seq = test_sequencer(0, 100); - let block = mainnet_block(&BLOCK_MAINNET_1_BYTES); - - // Height 1 is outstanding on a slow peer, so it is not pending. The top - // reorder bodies saturate the budget and must still be shed to make the - // floor watchdog's re-request fundable. - work.take_in_range(block::Height(1), block::Height(1), 1); - for height in [5u32, 6] { - work.take_in_range(block::Height(height), block::Height(height), 1); - assert!(budget.try_reserve(worst)); - seq.accept_body( - block::Height(height), - block::Hash([height as u8; 32]), - block.clone(), - worst, - peer(0), - ); - } - assert_eq!(budget.available(), 0, "budget saturated by buffered bodies"); - assert!( - !work.pending_contains(block::Height(1)), - "the floor gap is outstanding, not pending" - ); - - let shed = super::sequencer_task::shed_top_for_floor_starvation(&mut budget, &work, &mut seq); - assert!( - shed, - "the top buffered body is shed even while the floor gap is outstanding" - ); - assert!( - budget.available() >= worst, - "freed budget can now fund the watchdog floor re-request" - ); - assert!( - !seq.reorder_contains(block::Height(6)), - "the top body is evicted" - ); - assert!( - seq.reorder_contains(block::Height(5)), - "the lower buffered body is kept" - ); - assert!( - work.pending_contains(block::Height(6)), - "the evicted height is returned to pending for re-fetch" - ); -} - -#[test] -fn shed_top_until_available_self_funds_floor_reservation() { - let worst = BS_PER_BLOCK_WORST_CASE_BYTES; - let mut budget = ByteBudget::new(3 * worst); - let work = work_queue_with( - 0, - [ - needed(1, BlockSizeEstimate::Unknown), - needed(8, BlockSizeEstimate::Unknown), - needed(9, BlockSizeEstimate::Unknown), - needed(10, BlockSizeEstimate::Unknown), - ], - ); - let mut seq = test_sequencer(0, 100); - let block = mainnet_block(&BLOCK_MAINNET_1_BYTES); - - // The floor has already been taken for a request whose reservation lost a - // budget race. The speculative bodies saturate the budget, so the floor - // reservation path must pop enough high-tail bodies and retry. - work.take_in_range(block::Height(1), block::Height(1), 1); - for height in [8u32, 9, 10] { - work.take_in_range(block::Height(height), block::Height(height), 1); - assert!(budget.try_reserve(worst)); - seq.accept_body( - block::Height(height), - block::Hash([height as u8; 32]), - block.clone(), - worst, - peer(0), - ); - } - assert_eq!(budget.available(), 0); - - let shed = - super::sequencer_task::shed_top_until_available(&mut budget, &work, &mut seq, 2 * worst); - assert!(shed, "the high tail is popped to fund the floor request"); - assert!( - budget.available() >= 2 * worst, - "the pop frees the requested floor bytes" - ); - assert!( - !seq.reorder_contains(block::Height(10)) && !seq.reorder_contains(block::Height(9)), - "the furthest buffered bodies are evicted first" - ); - assert!( - seq.reorder_contains(block::Height(8)), - "nearer buffered body is kept once the request is fundable" - ); } // ---- Sequencer commit pipeline ---- @@ -4044,43 +3869,6 @@ fn sequencer_unsubmit_ignores_stale_or_mismatched_token() { ); } -#[test] -fn sequencer_reorder_max_height_reports_highest_buffered() { - let mut seq = test_sequencer(0, 8); - let blocks = mainnet_blocks_1_to_3(); - // Empty reorder buffer has no top. - assert_eq!(seq.reorder_max_height(), None); - // Buffer heights 1 and 3, leaving a gap at 2; the top is the highest buffered. - seq.accept_body( - block::Height(1), - blocks[0].hash(), - blocks[0].clone(), - 100, - peer(0), - ); - seq.accept_body( - block::Height(3), - blocks[2].hash(), - blocks[2].clone(), - 300, - peer(0), - ); - assert_eq!(seq.reorder_max_height(), Some(block::Height(3))); - // Draining the contiguous prefix removes height 1 but leaves the top (3). - seq.drain_ready_into_applying(); - assert_eq!(seq.reorder_max_height(), Some(block::Height(3))); - // Filling the gap drains 2 and 3 out, emptying the reorder buffer. - seq.accept_body( - block::Height(2), - blocks[1].hash(), - blocks[1].clone(), - 200, - peer(0), - ); - seq.drain_ready_into_applying(); - assert_eq!(seq.reorder_max_height(), None); -} - #[test] fn sequencer_keeps_whole_body_for_contiguous_height() { // The retain-for-backlog trim only applies to *non-contiguous* bodies. A body @@ -4125,22 +3913,19 @@ fn reorder_fuzzes_arrival_order_as_parent_first() { for order in orders { let mut reorder = ReorderBuffer::new(); - let mut budget = ByteBudget::new(10_000); let block = mainnet_block(&BLOCK_MAINNET_1_BYTES); let mut tip = block::Height(0); let mut released_all = Vec::new(); for height in order { - assert!(budget.try_reserve(100)); assert_eq!( reorder.insert(block::Height(height), block.clone(), 100, peer(0)), ReorderInsertResult::Inserted ); - for (released, _, bytes, _) in reorder.drain_contiguous_prefix(tip) { + for (released, _, _, _) in reorder.drain_contiguous_prefix(tip) { assert_eq!(released, block::Height(tip.0 + 1)); tip = released; released_all.push(released); - budget.release(bytes); } } @@ -4153,13 +3938,13 @@ fn reorder_fuzzes_arrival_order_as_parent_first() { block::Height(4) ] ); - assert_eq!(budget.reserved(), 0); + assert_eq!(reorder.buffered_bytes(), 0); } } -/// Build an outstanding three-block range whose worst-case reservation is already -/// held against `budget`, mirroring what the scheduler does at send time. -/// Per-height size-estimate reservation used by the budget-accounting tests. +/// Build an outstanding three-block range whose summed size-estimate reservation +/// is already held against `budget`, mirroring what the fill loop does at send +/// time. Per-height size-estimate reservation used by the budget-accounting tests. const THREE_BLOCK_ESTIMATE: u64 = 1_000; fn outstanding_three_block_range(budget: &mut ByteBudget) -> OutstandingBlockRange { @@ -4219,27 +4004,19 @@ fn outstanding_range_accumulates_delivered_bytes_for_bbr_sample() { } #[test] -fn block_budget_ledger_settles_and_releases_current_charge() { - let mut under = BlockBudgetLedger::reserved(1_000); - assert_eq!(under.current_charge(), 1_000); - assert_eq!(under.settle(700), -300); - assert_eq!(under.current_charge(), 700); - assert_eq!(under.release(), 700); - assert_eq!(under.release(), 0); - - let mut equal = BlockBudgetLedger::reserved(1_000); - assert_eq!(equal.settle(1_000), 0); - assert_eq!(equal.release(), 1_000); - - let mut over = BlockBudgetLedger::reserved(1_000); - assert_eq!(over.settle(1_300), 300); - assert_eq!(over.current_charge(), 1_300); - assert_eq!(over.release(), 1_300); +fn block_budget_ledger_releases_reserved_exactly_once() { + let mut ledger = BlockBudgetLedger::reserved(1_000); + assert!(ledger.is_reserved()); + assert_eq!(ledger.reserved_charge(), 1_000); + assert_eq!(ledger.release_reserved(), 1_000); + assert!(!ledger.is_reserved()); + assert_eq!(ledger.reserved_charge(), 0); + // Releasing twice yields nothing: the reservation lifecycle is one-shot. + assert_eq!(ledger.release_reserved(), 0); let mut released = BlockBudgetLedger::Released; - assert_eq!(released.settle(900), 0); - assert_eq!(released.current_charge(), 0); - assert_eq!(released.release(), 0); + assert!(!released.is_reserved()); + assert_eq!(released.release_reserved(), 0); } #[test] @@ -4247,14 +4024,66 @@ fn budget_audit_catches_injected_drift() { let mut budget = ByteBudget::new(1_000); assert!(budget.try_reserve(400)); assert!(budget.audit(400, "test matching audit")); + // A budget excess is expected receipt-handoff skew (paired call sites drain + // the source ledger before releasing the budget), never drift. + assert!(budget.audit(300, "test transient handoff excess")); + // A budget shortfall cannot be produced by any healthy interleaving + // (a double release or lost charge): drift. assert!( - !budget.audit(300, "test injected drift"), - "audit reports mismatched derived accounting" + !budget.audit(500, "test injected drift"), + "audit reports a budget shortfall as drift" ); budget.release(400); assert!(budget.audit(0, "test released audit")); } +#[test] +fn received_body_admission_ignores_request_budget() { + // A received body consumes no request budget, so a wire budget saturated by + // outstanding requests must not cause it to be dropped and re-downloaded: + // only the commit-window exemption and the resident gate apply. + let config = ZakuraBlockSyncConfig { + max_reorder_lookahead_bytes: 1_000, + ..ZakuraBlockSyncConfig::default() + }; + let snapshot = super::admission::AdmissionSnapshot { + download_floor: block::Height(600), + verified_block_tip: block::Height(10), + reorder_buffered_bytes: 0, + reorder_buffered_blocks: 0, + applying_buffered_bytes: 0, + applying_buffered_blocks: 0, + sequencer_input_queued_bytes: 0, + reserved_above_floor_bytes: 0, + reserved_above_floor_blocks: 0, + // Outstanding requests have spent the entire in-flight budget. + budget_available: 0, + }; + let factor = super::admission::DESERIALIZED_MEM_FACTOR; + // In-window heights (<= verified_tip 10 + 401 = 411) stay exempt. + assert!(super::admission::admit_received_body( + &config, + &snapshot, + block::Height(411), + 2_000_000, + )); + // Above the window the resident gate is the sole authority: a body within + // the remaining headroom is retained even with zero request budget... + assert!(super::admission::admit_received_body( + &config, + &snapshot, + block::Height(602), + 1_000 / factor, + )); + // ...and one that would breach the resident budget once decoded is refused. + assert!(!super::admission::admit_received_body( + &config, + &snapshot, + block::Height(602), + 1_000 / factor + 1, + )); +} + proptest::proptest! { #![proptest_config(proptest::test_runner::Config::with_cases(256))] @@ -4368,51 +4197,37 @@ proptest::proptest! { #[test] fn block_budget_ledger_mirrors_byte_budget( estimate in 1u64..1_000_000, - actual in 1u64..1_000_000, release_before_receipt in proptest::bool::ANY, ) { + // Whichever terminal path runs first (receipt, or a timeout/watchdog + // release) drains the mirrored budget to zero; any later path is a no-op. let mut ledger = BlockBudgetLedger::reserved(estimate); let mut budget = ByteBudget::new(u64::MAX); prop_assert!(budget.try_reserve(estimate)); if release_before_receipt { - let released = ledger.release(); - budget.release(released); - prop_assert_eq!(budget.reserved(), 0); - let delta = ledger.settle(actual); - prop_assert_eq!(delta, 0); + budget.release(ledger.release_reserved()); prop_assert_eq!(budget.reserved(), 0); - prop_assert_eq!(ledger.current_charge(), 0); - let released = ledger.release(); - budget.release(released); - prop_assert_eq!(budget.reserved(), 0); - } else { - let delta = ledger.settle(actual); - if delta >= 0 { - budget.charge(u64::try_from(delta).expect("positive test delta fits in u64")); - } else { - budget.release(u64::try_from(-delta).expect("negative test delta fits in u64")); - } - prop_assert_eq!(ledger.current_charge(), actual); - prop_assert_eq!(budget.reserved(), actual); - - let released = ledger.release(); - budget.release(released); - prop_assert_eq!(budget.reserved(), 0); - prop_assert_eq!(ledger.current_charge(), 0); + prop_assert!(!ledger.is_reserved()); } + budget.release(ledger.release_reserved()); + prop_assert_eq!(budget.reserved(), 0); + prop_assert_eq!(ledger.reserved_charge(), 0); + // A third release still yields nothing. + budget.release(ledger.release_reserved()); + prop_assert_eq!(budget.reserved(), 0); } } -/// The global reservation settles to the current held body bytes across the -/// download -> buffer -> apply -> commit path, and releases exactly once across -/// timeout/duplicate/short-response paths. +/// The global reservation tracks only unreceived request estimates: each height +/// releases its whole estimate exactly once (receipt or timeout), retained bodies +/// charge nothing, and the counter never exceeds the max. #[test] -fn budget_reservation_never_exceeds_max_and_only_shrinks_per_block() { +fn budget_reservation_tracks_only_unreceived_estimates() { let estimate = THREE_BLOCK_ESTIMATE; let max = estimate * 3; - // Happy path: download -> shrink-on-receipt -> buffer -> apply -> commit. + // Happy path: download -> release-on-receipt -> buffer (uncharged) -> commit. { let mut budget = ByteBudget::new(max); let mut reorder = ReorderBuffer::new(); @@ -4421,82 +4236,75 @@ fn budget_reservation_never_exceeds_max_and_only_shrinks_per_block() { assert!(budget.reserved() <= budget.max_bytes_for_test()); let block = mainnet_block(&BLOCK_MAINNET_1_BYTES); - // Receive each height: release `estimate - actual`, keep `actual` reserved, - // and hand `actual` to the reorder buffer without re-reserving. - let actuals = [700u64, 800, 900]; // each < its estimate, varies per block + // Receive each height: release its whole estimate and hand the body's + // actual size to the reorder buffer without any budget charge. + let actuals = [700u64, 800, 900]; for (index, height) in [block::Height(1), block::Height(2), block::Height(3)] .into_iter() .enumerate() { - let before = budget.reserved(); - let actual = actuals[index]; - budget.settle(estimate, actual); + budget.release(estimate); outstanding.mark_received(height); assert_eq!( - reorder.insert(height, block.clone(), actual, peer(0)), + reorder.insert(height, block.clone(), actuals[index], peer(0)), ReorderInsertResult::Inserted ); - // Per-block reservation only shrank (estimate -> actual), never grew. - assert!(budget.reserved() <= before); - assert!(budget.reserved() <= budget.max_bytes_for_test()); + // The budget mirrors the unreceived estimates exactly. + assert_eq!(budget.reserved(), outstanding.reserved_bytes()); } assert!(outstanding.is_complete()); assert_eq!(outstanding.reserved_bytes(), 0); - assert_eq!(budget.reserved(), 700 + 800 + 900); - - // Commit: draining to apply carries the actual bytes; the apply finish - // releases them. - let mut floor = block::Height(0); - let mut applied_bytes = 0; - for (_height, _block, bytes, _peer) in reorder.drain_contiguous_prefix(floor) { - applied_bytes += bytes; - floor = block::Height(floor.0 + 1); - } + assert_eq!( + budget.reserved(), + 0, + "retained bodies charge nothing; the wire budget is fully free" + ); + assert_eq!(reorder.buffered_bytes(), 700 + 800 + 900); + + // Commit: draining to apply carries the actual bytes for the retained + // view; no budget release is owed anywhere downstream. + let applied_bytes: u64 = reorder + .drain_contiguous_prefix(block::Height(0)) + .iter() + .map(|(_, _, bytes, _)| *bytes) + .sum(); assert_eq!(applied_bytes, 700 + 800 + 900); - budget.release(applied_bytes); assert_eq!(budget.reserved(), 0); } - // Timeout / short-response path: heights that never buffer release exactly - // their size-estimate share, with no leak and no double-release. + // Timeout / short-response path: the received height released its estimate at + // receipt; the unreceived heights release theirs exactly once on timeout. { let mut budget = ByteBudget::new(max); let mut outstanding = outstanding_three_block_range(&mut budget); assert_eq!(budget.reserved(), estimate * 3); - // A short response delivers only height 1; release its estimate's slack. - let actual = 700u64; - budget.settle(estimate, actual); + // A short response delivers only height 1; its estimate is released. + budget.release(estimate); outstanding.mark_received(block::Height(1)); // The remaining two unreceived heights still reserve their estimate each. assert_eq!(outstanding.reserved_bytes(), estimate * 2); - assert!(budget.reserved() <= budget.max_bytes_for_test()); - // On timeout the outstanding range releases its still-reserved estimate. + assert_eq!(budget.reserved(), estimate * 2); + // On timeout the outstanding range releases its still-reserved estimates. budget.release(outstanding.reserved_bytes()); - // Plus the actual bytes held for the one received-but-not-buffered height. - budget.release(actual); assert_eq!(budget.reserved(), 0); } - // Duplicate path: a duplicate body reserves nothing extra and releases its - // actual bytes, so the buffer's reservation is unchanged. + // Duplicate path: neither the first body nor a dropped duplicate touches the + // budget; only the retained-size accounting sees them. { - let mut budget = ByteBudget::new(max); + let budget = ByteBudget::new(max); let mut reorder = ReorderBuffer::new(); let block = mainnet_block(&BLOCK_MAINNET_1_BYTES); - assert!(budget.try_reserve(1_000)); assert_eq!( reorder.insert(block::Height(1), block.clone(), 1_000, peer(0)), ReorderInsertResult::Inserted ); - // A second body for the same height is a duplicate; reserve-then-release - // leaves the reservation exactly where it was. - assert!(budget.try_reserve(1_000)); assert_eq!( reorder.insert(block::Height(1), block, 1_000, peer(0)), ReorderInsertResult::Duplicate ); - budget.release(1_000); - assert_eq!(budget.reserved(), 1_000); + assert_eq!(reorder.buffered_bytes(), 1_000); + assert_eq!(budget.reserved(), 0); assert!(budget.reserved() <= budget.max_bytes_for_test()); } } @@ -4540,10 +4348,10 @@ fn received_tracker_handles_a_full_range_at_the_bitset_boundary() { } /// A body whose actual serialized size exceeds its advertised size hint is still -/// accepted and buffered, and the byte budget charges the overshoot so it cannot -/// issue more work while under-counting held bodies. +/// accepted and buffered, and receipt releases exactly the reserved hint — never +/// more — so an oversized body cannot underflow or leak the request budget. #[test] -fn underestimated_body_is_buffered_and_charges_budget_delta() { +fn underestimated_body_is_buffered_and_releases_only_its_estimate() { let hint = 1_000u64; // Budget holds several hint-sized shares: enough to reserve this body's hint. let mut budget = ByteBudget::new(hint * 4); @@ -4579,9 +4387,10 @@ fn underestimated_body_is_buffered_and_charges_budget_delta() { assert!(actual < BS_PER_BLOCK_WORST_CASE_BYTES); assert!(actual > hint); - // Receipt: settle toward the actual size. Because actual exceeds the reserved - // hint, the budget charges the delta even though the body is already admitted. - budget.settle(hint, actual); + // Receipt: release the reserved hint. The oversized body is retained at its + // actual size by the reorder buffer, invisible to the request budget (the + // resident look-ahead gate accounts for it instead). + budget.release(hint); outstanding.mark_received(block::Height(1)); let block = mainnet_block(&BLOCK_MAINNET_1_BYTES); assert_eq!( @@ -4590,14 +4399,15 @@ fn underestimated_body_is_buffered_and_charges_budget_delta() { "an underestimated body must still buffer; a received body is never dropped" ); assert_eq!(reorder.buffered_bytes(), actual); - assert_eq!(budget.reserved(), actual); - assert_eq!(budget.available(), 0); + assert_eq!(budget.reserved(), 0); + assert_eq!(outstanding.reserved_bytes(), 0); assert!( - !budget.try_reserve(hint), - "charging the underestimated delta closes the budget gate" + budget.try_reserve(hint), + "the freed reservation funds the next request immediately" ); - budget.release(reorder.drop_through(block::Height(1))); - assert_eq!(budget.reserved(), 0); + let dropped = reorder.drop_through(block::Height(1)); + assert_eq!(dropped, actual); + assert_eq!(reorder.buffered_bytes(), 0); } #[test] @@ -5125,7 +4935,7 @@ async fn reactor_drives_tip_to_getblocks_to_submit_over_framed_path() { } #[tokio::test] -async fn reactor_keeps_submitted_body_budget_until_apply_finishes() { +async fn reactor_releases_request_budget_at_receipt_not_apply() { let blocks = mainnet_blocks_1_to_3(); let block1_size = block_size(&blocks[0]); let mut config = immediate_body_download_config(); @@ -5225,21 +5035,14 @@ async fn reactor_keeps_submitted_body_budget_until_apply_finishes() { }; assert_eq!(handle.local_status().servable_high, block::Height(0)); - // The submitted-but-unapplied block must keep the body budget full, so no new - // GetBlocks may issue. a routine that consumed work pings the reactor to - // re-query, so a budget-orthogonal `QueryNeededBlocks` may appear in this - // window (the producer self-gates; it is idempotent and downloads nothing). - // Tolerate only that; any GetBlocks/SubmitBlock here would be the regression. - let deadline = tokio::time::Instant::now() + Duration::from_millis(100); - loop { - match tokio::time::timeout_at(deadline, actions.recv()).await { - Err(_) => break, - Ok(Some(BlockSyncAction::QueryNeededBlocks { .. })) => {} - Ok(other) => panic!( - "submitted-but-not-applied block should keep body budget full; got {other:?}" - ), - } - } + // Receipt — not apply-finish — released the request reservation, so with the + // one-block-hint budget and the one-slot window both freed by the delivery, + // the height-2 GetBlocks must go out while height 1 is still + // submitted-but-unapplied (retention is bounded by the resident gate, not + // the request budget). + let (start_height, count) = wait_for_outbound_getblocks(&mut outbound_rx).await; + assert_eq!(start_height, block::Height(2)); + assert_eq!(count, 1); handle .send(BlockSyncEvent::BlockApplyFinished { @@ -5266,10 +5069,6 @@ async fn reactor_keeps_submitted_body_budget_until_apply_finishes() { .await .expect("apply completion frontier advances advertised status"); - let (start_height, count) = wait_for_outbound_getblocks(&mut outbound_rx).await; - assert_eq!(start_height, block::Height(2)); - assert_eq!(count, 1); - reactor_task.abort(); } @@ -5436,13 +5235,14 @@ async fn reactor_does_not_requeue_held_height_reported_still_needed() { } /// A real body whose serialized size dwarfs its advertised size hint is still -/// accepted, buffered, and submitted. Worst case (not the hint) is reserved at -/// send time, so shrink-on-receipt always has room and never drops the body. +/// accepted, buffered, and submitted: a received body never touches the budget +/// gate (its reservation is simply released at receipt), so it cannot be dropped +/// for lack of budget. /// /// A/B: under the old release-then-reserve scheme the request reserved only the /// hint-sized estimate, so a body larger than the hint re-reserved more than was /// released and, against a tight budget, hit `BudgetFull` and dropped a valid -/// body. With worst-case reservation this path is unreachable. +/// body. That path is unreachable now. #[tokio::test] async fn reactor_buffers_body_larger_than_its_size_hint() { let blocks = mainnet_blocks_1_to_3(); @@ -5803,6 +5603,8 @@ async fn reactor_keeps_applying_body_after_non_advancing_duplicate_result() { .expect("needed metadata after duplicate queues"); // A re-request would land on this peer's own real outbound, so watch the wire. + // Height 2 may legitimately be requested (receipt already freed height 1's + // request reservation); only a height-1 re-request would be the regression. let no_duplicate_request = tokio::time::timeout(Duration::from_millis(100), async { while let Some(frame) = outbound_rx.recv().await { if let BlockSyncMessage::GetBlocks { @@ -5810,7 +5612,9 @@ async fn reactor_keeps_applying_body_after_non_advancing_duplicate_result() { count, } = BlockSyncMessage::decode_frame(frame).expect("outbound frame decodes") { - panic!("non-advancing duplicate result re-requested {start_height:?}/{count}"); + if start_height == block::Height(1) { + panic!("non-advancing duplicate result re-requested {start_height:?}/{count}"); + } } } }) @@ -6055,89 +5859,95 @@ async fn reactor_ignores_unmatched_body_for_currently_needed_height() { #[tokio::test] async fn reactor_accepts_unmatched_body_for_queued_height() { let blocks = mainnet_blocks_1_to_3(); - let mut config = immediate_body_download_config(); - config.max_inflight_block_bytes = u64::from(block_size(&blocks[0])); - let (_tip_tx, tip_rx) = watch::channel((block::Height(1), blocks[0].hash())); + let block1_size = block_size(&blocks[0]); + let block2_size = block_size(&blocks[1]); + let config = immediate_body_download_config(); + let (_tip_tx, tip_rx) = watch::channel((block::Height(2), blocks[1].hash())); let startup = BlockSyncStartup::new( BlockSyncFrontiers { finalized_height: block::Height(0), verified_block_tip: block::Height(0), verified_block_hash: block::Hash([0; 32]), }, - (block::Height(1), blocks[0].hash()), + (block::Height(2), blocks[1].hash()), tip_rx, config.clone(), ); let (handle, mut actions, reactor_task) = spawn_block_sync_reactor(startup); let service = BlockSyncService::new_with_handle_for_test(config, handle.clone()); + // One request slot and a response byte cap of one block-1 hint, so the single + // GetBlocks below covers only height 1 and height 2 stays queued without an + // outstanding request — the gap the unmatched-queued acceptance path fills. + // (The old setup starved the request on the byte budget with an oversized + // hint; the floor overdraft now funds a floor request regardless, so the gap + // is created on slots/response bytes instead.) let (_peer_id, inbound_tx, mut outbound_rx) = connect_peer_with_status( &service, &mut actions, 143, - block::Height(1), - blocks[0].hash(), - 1, + block::Height(2), + blocks[1].hash(), 1, + block1_size, ) .await; - // Queue height 1 with a deliberately oversized advertised hint so its send-time - // reservation (the estimate) exceeds the one-actual-block budget: no GetBlocks - // can issue, leaving the height queued without an outstanding request. The body, - // arriving unsolicited, reserves only its small actual size and is still - // buffered. Under the old worst-case reservation this gap was implicit; with - // size-based reservation the oversized hint recreates it. handle - .send(BlockSyncEvent::NeededBlocks(vec![BlockSyncBlockMeta { - height: block::Height(1), - hash: blocks[0].hash(), - size: BlockSizeEstimate::Advertised( - u32::try_from(BS_PER_BLOCK_WORST_CASE_BYTES).expect("worst case fits u32"), - ), - }])) + .send(BlockSyncEvent::NeededBlocks(vec![ + BlockSyncBlockMeta { + height: block::Height(1), + hash: blocks[0].hash(), + size: BlockSizeEstimate::Advertised(block1_size), + }, + BlockSyncBlockMeta { + height: block::Height(2), + hash: blocks[1].hash(), + size: BlockSizeEstimate::Advertised(block2_size), + }, + ])) .await .expect("needed metadata queues"); - // No GetBlocks may issue (the body must stay queued without an outstanding - // request); a request would land on this peer's own real outbound. - let no_getblocks = tokio::time::timeout(Duration::from_millis(100), async { - loop { - if let Some(frame) = outbound_rx.recv().await { - if let BlockSyncMessage::GetBlocks { .. } = - BlockSyncMessage::decode_frame(frame).expect("outbound frame decodes") - { - return; - } - } else { - return; - } - } - }) - .await; - assert!( - no_getblocks.is_err(), - "test setup requires the body to remain queued without an outstanding request", + assert_eq!( + wait_for_outbound_getblocks(&mut outbound_rx).await, + (block::Height(1), 1), + "the byte-capped request must cover only height 1", ); + // Height 2's body arrives without a matching outstanding request. It is a + // queued (pending) height in the peer's servable range, so the routine must + // claim and buffer it — no request reservation is consumed for it. inbound_tx .send( - BlockSyncMessage::Block(blocks[0].clone()) + BlockSyncMessage::Block(blocks[1].clone()) .encode_frame() .expect("block encodes"), ) .await .expect("unmatched queued block queues"); - let submitted = tokio::time::timeout(Duration::from_secs(1), async { - loop { + // Deliver height 1 on its live request; both bodies must submit in order, + // proving the unmatched height-2 body was accepted and buffered. + inbound_tx + .send( + BlockSyncMessage::Block(blocks[0].clone()) + .encode_frame() + .expect("block encodes"), + ) + .await + .expect("matched block queues"); + + let mut submitted = Vec::new(); + tokio::time::timeout(Duration::from_secs(1), async { + while submitted.len() < 2 { if let BlockSyncAction::SubmitBlock { block, .. } = next_action(&mut actions).await { - return block.hash(); + submitted.push(block.hash()); } } }) .await - .expect("unmatched queued body is submitted"); - assert_eq!(submitted, blocks[0].hash()); + .expect("both bodies are submitted"); + assert_eq!(submitted, vec![blocks[0].hash(), blocks[1].hash()]); reactor_task.abort(); } @@ -6540,8 +6350,8 @@ async fn reactor_keeps_issuing_far_above_floor_with_no_near_tip_pause() { async fn routine_refills_after_budget_release_no_missed_wake() { // missed-wake guard at the routine level: a routine blocked on an // exhausted byte budget must re-fill when budget is freed. The budget holds - // exactly one worst-case block, so the first GetBlocks exhausts it; delivering - // that body releases budget (shrink + commit) and the routine must issue the + // exactly one worst-case block, so the first GetBlocks exhausts it; receipt + // of that body releases the whole reservation and the routine must issue the // next GetBlocks without any external nudge. let mut config = immediate_body_download_config(); config.max_inflight_block_bytes = BS_PER_BLOCK_WORST_CASE_BYTES; @@ -6597,10 +6407,9 @@ async fn routine_refills_after_budget_release_no_missed_wake() { let (first_start, _count) = wait_for_outbound_getblocks(&mut outbound_rx).await; assert_eq!(first_start, block::Height(1)); - // Deliver height 1's body. Its worst-case reservation shrinks to the actual - // size, but the remaining reservation still leaves under one worst-case block - // free, so the routine cannot yet issue height 2 — the budget is freed only - // when height 1 *commits* and the Sequencer releases its reservation. + // Deliver height 1's body. Receipt releases its whole worst-case reservation + // (retention charges nothing), so the routine can issue height 2 as soon as + // its next fill pass runs. inbound_tx .send( BlockSyncMessage::Block(blocks[0].clone()) @@ -6610,10 +6419,10 @@ async fn routine_refills_after_budget_release_no_missed_wake() { .await .expect("block frame queues"); - // Drive height 1 to commit: drain its SubmitBlock and report it applied. The - // Sequencer task then releases the byte budget and the capacity notify must - // wake the budget-blocked routine to issue the next GetBlocks (the // missed-wake guarantee — a release between the routine's fill-check and its - // await must not be lost). + // Drive height 1 through submit and apply so the pipeline finishes cleanly; + // the height-2 GetBlocks below must arrive regardless (the receipt-time + // release must wake the budget-blocked fill — a release between the + // routine's fill-check and its await must not be lost). let token = loop { match next_action(&mut actions).await { BlockSyncAction::SubmitBlock { token, block } => { @@ -6639,7 +6448,7 @@ async fn routine_refills_after_budget_release_no_missed_wake() { assert_eq!( second_start, block::Height(2), - "freeing the byte budget must wake the routine to issue the next request (no missed wake)" + "freeing the request budget must wake the routine to issue the next request (no missed wake)" ); reactor_task.abort(); diff --git a/zebra-network/src/zakura/block_sync/work_queue.rs b/zebra-network/src/zakura/block_sync/work_queue.rs index 587b87edf50..ceb922d04ff 100644 --- a/zebra-network/src/zakura/block_sync/work_queue.rs +++ b/zebra-network/src/zakura/block_sync/work_queue.rs @@ -39,13 +39,13 @@ pub(super) struct WorkItem { /// The block's size estimate. Used for request budget reservation and the /// receive-path `SizeMismatch` tolerance check. pub(super) estimated_bytes: u64, - /// Current byte-budget charge owned by this height. + /// Request-estimate reservation owned by this height. /// /// Pending items normally have `Released`; issued-but-unreceived items have - /// `Reserved(estimate)`; received bodies held by the commit pipeline have - /// `Held(actual)`. All terminal release paths go through this ledger so a - /// stale local owner cannot release a charge that another path already - /// returned. + /// `Reserved(estimate)`. Receipt releases the reservation — retained bodies + /// never charge the wire budget. All terminal release paths go through this + /// ledger so a stale local owner cannot release a charge that another path + /// already returned. pub(super) budget: BlockBudgetLedger, } @@ -299,7 +299,7 @@ impl WorkQueue { let Some(item) = inner.in_flight.get_mut(&height) else { continue; }; - if item.budget.current_charge() != 0 { + if item.budget.is_reserved() { continue; } item.budget = BlockBudgetLedger::reserved(item.estimated_bytes); @@ -311,89 +311,57 @@ impl WorkQueue { marked } - /// Settle a body only if this height still owns an active request reservation. + /// Release the request reservation for a received body, only if this height + /// still owns an active one. Returns the reserved estimate so the caller + /// returns it to the `ByteBudget` once the body is accounted downstream. /// /// Returns `None` when a central watchdog or local timeout already released /// and returned the height. Late bodies from that superseded claim must not - /// resurrect a second charge. - pub(super) fn settle_active_reserved_height( - &self, - height: block::Height, - actual: u64, - ) -> Option { + /// release a second time. + pub(super) fn release_active_reserved_height(&self, height: block::Height) -> Option { let mut inner = self.lock(); - let (reserved_before, delta) = { + let released = { let item = inner.in_flight.get_mut(&height)?; if !item.budget.is_reserved() { return None; } - // Reserved(reserved) -> Held(actual): the reserved charge drops to 0. - (item.budget.reserved_charge(), item.budget.settle(actual)) + item.budget.release_reserved() }; - inner.reserved_bytes = inner.reserved_bytes.saturating_sub(reserved_before); - Some(delta) + inner.reserved_bytes = inner.reserved_bytes.saturating_sub(released); + Some(released) } - /// Mark a height as directly held after the caller admitted `actual` bytes. + /// Claim a received height into `in_flight` (moving it out of `pending` if + /// needed) and clear any request reservation it still owned, returning that + /// prior charge so the caller releases it from the `ByteBudget`. /// - /// Used for unmatched queued bodies, which did not have a prior request - /// estimate reservation. - pub(super) fn mark_held_direct(&self, height: block::Height, actual: u64) -> u64 { + /// Used for unmatched queued bodies, which consume no request reservation: + /// the retained body is bounded by the resident look-ahead gate instead. + pub(super) fn claim_received(&self, height: block::Height) -> u64 { let mut inner = self.lock(); if let Some(item) = inner.in_flight.get_mut(&height) { - // X -> Held(actual): any reserved charge this item still owned is gone. - let reserved_before = item.budget.reserved_charge(); - let previous_charge = item.budget.release(); - item.budget = BlockBudgetLedger::Held(actual); - inner.reserved_bytes = inner.reserved_bytes.saturating_sub(reserved_before); - return previous_charge; + let released = item.budget.release_reserved(); + inner.reserved_bytes = inner.reserved_bytes.saturating_sub(released); + return released; } if let Some(mut item) = inner.pending.remove(&height) { - let reserved_before = item.budget.reserved_charge(); - let previous_charge = item.budget.release(); - item.budget = BlockBudgetLedger::Held(actual); + let released = item.budget.release_reserved(); inner.in_flight.insert(height, item); - inner.reserved_bytes = inner.reserved_bytes.saturating_sub(reserved_before); - return previous_charge; + inner.reserved_bytes = inner.reserved_bytes.saturating_sub(released); + return released; } 0 } - /// Release *any* live charge (reserved estimate **or** held body bytes) for - /// `heights`, exactly once. This is the non-Held-aware release: it hands back - /// held-body bytes the Sequencer also owns, so production must never call it - /// (see [`release_reserved_heights`](Self::release_reserved_heights) and - /// [`release_reserved_and_return_items`](Self::release_reserved_and_return_items)). - /// Retained only to exercise the raw ledger arithmetic in unit tests; the - /// `#[cfg(test)]` gate is what structurally enforces "prod is Held-aware". - #[cfg(test)] - pub(super) fn release_heights(&self, heights: impl IntoIterator) -> u64 { - let mut released = 0u64; - let mut reserved_removed = 0u64; - let mut inner = self.lock(); - for height in heights { - if let Some(item) = inner.in_flight.get_mut(&height) { - reserved_removed = reserved_removed.saturating_add(item.budget.reserved_charge()); - released = released.saturating_add(item.budget.release()); - } else if let Some(item) = inner.pending.get_mut(&height) { - reserved_removed = reserved_removed.saturating_add(item.budget.reserved_charge()); - released = released.saturating_add(item.budget.release()); - } - } - inner.reserved_bytes = inner.reserved_bytes.saturating_sub(reserved_removed); - released - } - - /// Release only the still-reserved size-estimate charges for `heights`, + /// Release any still-active request-estimate reservations for `heights`, /// exactly once, leaving each height in place. /// - /// A height that already settled to `Held(actual)` is owned by the body - /// handoff / Sequencer path (it releases those actual bytes on commit), so it - /// is skipped here: never released and never double-counted. Mirrors + /// A height whose body already arrived is `Released` (the receipt path ended + /// its reservation), so it is skipped: never released twice and never + /// double-counted. Mirrors /// [`release_reserved_and_return_items`](Self::release_reserved_and_return_items) /// for callers dropping heights below the floor (GC / stale trim) rather than - /// returning them to `pending`. Use instead of [`release_heights`](Self::release_heights) - /// on any path a competing peer's late body may have converted to `Held`. + /// returning them to `pending`. pub(super) fn release_reserved_heights( &self, heights: impl IntoIterator, @@ -402,39 +370,36 @@ impl WorkQueue { let mut inner = self.lock(); for height in heights { if let Some(item) = inner.in_flight.get_mut(&height) { - if item.budget.is_reserved() { - released = released.saturating_add(item.budget.release_reserved()); - } + released = released.saturating_add(item.budget.release_reserved()); } else if let Some(item) = inner.pending.get_mut(&height) { - if item.budget.is_reserved() { - released = released.saturating_add(item.budget.release_reserved()); - } + released = released.saturating_add(item.budget.release_reserved()); } } inner.reserved_bytes = inner.reserved_bytes.saturating_sub(released); released } - /// Release and return `in_flight` heights to `pending`. + /// Release and return `in_flight` heights to `pending`, whether or not their + /// reservation already ended. Production paths use the reserved-only variant + /// below (a received height must stay claimed for the commit pipeline); this + /// remains to exercise the raw return arithmetic in unit tests. + #[cfg(test)] pub(super) fn release_and_return_items( &self, heights: impl IntoIterator, ) -> u64 { let mut moved = false; let mut released = 0u64; - let mut reserved_removed = 0u64; { let mut inner = self.lock(); for height in heights { if let Some(mut item) = inner.in_flight.remove(&height) { - reserved_removed = - reserved_removed.saturating_add(item.budget.reserved_charge()); - released = released.saturating_add(item.budget.release()); + released = released.saturating_add(item.budget.release_reserved()); inner.pending.insert(height, item); moved = true; } } - inner.reserved_bytes = inner.reserved_bytes.saturating_sub(reserved_removed); + inner.reserved_bytes = inner.reserved_bytes.saturating_sub(released); } if moved { self.available.notify_waiters(); @@ -444,9 +409,10 @@ impl WorkQueue { /// Release and return only still-reserved `in_flight` heights to `pending`. /// - /// A height that has already settled to `Held(actual)` is owned by the body - /// handoff / Sequencer path. A central watchdog may clear stale peer claims, - /// but it must not release or requeue those bytes. + /// A height whose reservation was already released at receipt (`Released`) + /// is owned by the body handoff / Sequencer path. A central watchdog may + /// clear stale peer claims, but it must not requeue a height whose body we + /// already hold. pub(super) fn release_reserved_and_return_items( &self, heights: impl IntoIterator, @@ -466,9 +432,7 @@ impl WorkQueue { .in_flight .remove(&height) .expect("reserved item exists because it was just checked"); - // Only reserved items reach here, so the released bytes are exactly - // the reserved charge leaving the queue. - released = released.saturating_add(item.budget.release()); + released = released.saturating_add(item.budget.release_reserved()); inner.pending.insert(height, item); moved = true; } @@ -484,9 +448,8 @@ impl WorkQueue { /// floor)` and drop every `pending`/`in_flight` entry `<= floor`. /// /// Returns request-estimate bytes that were still reserved for unreceived - /// heights. Held body bytes are cleared from the ledger here but are not - /// returned: the Sequencer releases those actual body bytes when it drops - /// reorder/applying state. + /// heights, so the caller returns them to the `ByteBudget`. Received bodies + /// carry no charge here; the Sequencer's buffers own them. pub(super) fn advance_floor(&self, floor: block::Height) -> u64 { let mut inner = self.lock(); inner.floor = inner.floor.max(floor); @@ -631,10 +594,6 @@ impl WorkQueue { self.lock().pending.keys().next().copied() } - pub(super) fn min_in_flight(&self) -> Option { - self.lock().in_flight.keys().next().copied() - } - pub(super) fn first_pending_in_range( &self, low: block::Height, diff --git a/zebra-network/src/zakura/testkit/blocksync_fuzz/invariants.rs b/zebra-network/src/zakura/testkit/blocksync_fuzz/invariants.rs index 797b8745740..af49aad9721 100644 --- a/zebra-network/src/zakura/testkit/blocksync_fuzz/invariants.rs +++ b/zebra-network/src/zakura/testkit/blocksync_fuzz/invariants.rs @@ -20,7 +20,8 @@ pub(crate) struct InvariantReport { pub(crate) state_samples: usize, /// Peak aggregate in-flight requests across all peers. pub(crate) max_outstanding: u64, - /// Peak reserved download bytes (memory pressure). + /// Peak reserved outstanding-request bytes (wire-limiter pressure; received + /// bodies release their reservation, so this excludes retained bodies). pub(crate) peak_budget_reserved: u64, /// Peak retained pipeline wire bytes (`sequencer_input + reorder + applying`), the /// wire-byte footprint of bodies actually held in memory. Multiplied by the @@ -271,17 +272,36 @@ pub(crate) fn assert_core( outstanding_bound, ); - // The global byte budget is never over-committed: peak reserved download bytes - // (in-flight + reorder + applying) must stay within the configured ceiling. Every - // per-peer routine reserves against the same CAS-guarded `ByteBudget`, so this must - // hold no matter how many peers race — the memory bound the spec requires. Vacuous - // only for scenarios that set an effectively unbounded budget (`u64::MAX`); the - // tight-ceiling scenarios make it bite. + // The global request budget is never over-committed beyond its bounded floor + // overdraft: peak reserved outstanding-request bytes stay within the configured + // ceiling plus at most one floor request's reservation. Every per-peer routine + // reserves against the same CAS-guarded `ByteBudget`, so this must hold no matter + // how many peers race. Vacuous only for scenarios that set an effectively + // unbounded budget (`u64::MAX`); the tight-ceiling scenarios make it bite. + let overdraft_slack = scenario.config.floor_request_byte_reservation(); assert!( - report.peak_budget_reserved <= scenario.config.max_inflight_block_bytes, - "peak reserved bytes {} exceeded the global in-flight byte budget {}", + report.peak_budget_reserved + <= scenario + .config + .max_inflight_block_bytes + .saturating_add(overdraft_slack), + "peak reserved bytes {} exceeded the in-flight request budget {} (+{} floor-overdraft slack)", report.peak_budget_reserved, scenario.config.max_inflight_block_bytes, + overdraft_slack, + ); + + // Quiescence: `budget_reserved` tracks outstanding requests only, so every + // reservation must have drained (receipt, timeout, watchdog, floor GC, or + // reset — each exactly once). The receipt release is ordered *after* the body + // is handed to the sequencer, so the harness's final snapshot can race one + // in-flight release; allow at most that one request's worth of remnant. + assert!( + report.final_budget_reserved <= overdraft_slack, + "reserved request bytes {} must drain to zero at quiescence (allowing one \ + in-flight receipt release, <= {})", + report.final_budget_reserved, + overdraft_slack, ); } diff --git a/zebra-network/src/zakura/testkit/blocksync_fuzz/tests.rs b/zebra-network/src/zakura/testkit/blocksync_fuzz/tests.rs index a2c5b3496dc..a9c405dbcd5 100644 --- a/zebra-network/src/zakura/testkit/blocksync_fuzz/tests.rs +++ b/zebra-network/src/zakura/testkit/blocksync_fuzz/tests.rs @@ -831,30 +831,31 @@ async fn fuzz_mixed_block_sizes() { } /// Commit stall: the mock commit pipeline drains slowly and in bursts while fast peers -/// keep serving. This is the `BLOCKSYNC_BYTE_CWND_PLAN` step-3 system check — the apply -/// backlog must be bounded **only** by the byte budget (download parks on the memory -/// ceiling, not on a commit-coupled throttle), and the verified tip must resume the -/// instant each stall clears so the run still reaches the target. +/// keep serving. The apply backlog is bounded by the **resident look-ahead gate** +/// (retention no longer charges the in-flight wire budget), and the verified tip must +/// resume the instant each stall clears so the run still reaches the target. /// /// `run_checked` already asserts the target is reached (vtip resumes; no commit-induced -/// wedge). On top of that we assert the peak reserved bytes stayed within the configured -/// ceiling (the queue did not grow toward the full chain) yet the ceiling was actually -/// exercised (the stall created real backpressure — the bound is not vacuous). +/// wedge). On top of that we assert real retention pressure built up (peak retained +/// resident at least half the budget) and that the peak stayed within the budget plus +/// the commit-window exemption. At this 400-block chain length the window slack alone +/// covers the whole chain, so that upper bound cannot bind here — the tight-bound check +/// is the 1_200-block `fuzz_commit_stall_resident_plateau` variant; this test pins the +/// no-stall, pressure, and quiescence behavior of the bursty stall shape. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn fuzz_commit_stall() { let blocks = 400; let body_bytes = 32 * 1024usize; - // A small resident download budget so the commit stall makes it bind: the chain is - // ~12.8 MB (400 × 32 KiB) against a 2 MiB ceiling, so the budget must recycle ~6× and - // download genuinely waits on commit. Kept below the request-count cap's byte - // equivalent (2 peers × 64 reqs × 32 KiB = 4 MiB) so the *byte budget*, not the - // request count, is the binding constraint. (The fuzzer reactor path does not call - // `validate()`, which would otherwise require a full checkpoint range; the synthetic - // bodies are tiny and the mock committer needs no checkpoint batch, so a sub-floor - // ceiling is the right knob to exercise byte-budget backpressure here.) + // The resident equivalent of the old 2 MiB wire ceiling: retention is now gated by + // `max_reorder_lookahead_bytes` (resident-denominated), so the chain (~12.8 MB wire, + // ~51 MB resident) must recycle through the gate while download waits on commit. + // (The fuzzer reactor path does not call the config clamps; the synthetic bodies are + // tiny and the mock committer needs no checkpoint batch, so a sub-range budget is the + // right knob to exercise retention backpressure here.) let byte_ceiling: u64 = 2 * 1024 * 1024; + let resident_budget = byte_ceiling * DESERIALIZED_MEM_FACTOR; let config = ZakuraBlockSyncConfig { - max_inflight_block_bytes: byte_ceiling, + max_reorder_lookahead_bytes: resident_budget, max_blocks_per_response: 1, ..fuzz_config() }; @@ -869,7 +870,7 @@ async fn fuzz_commit_stall() { ); scenario.target_block_bytes = Some(body_bytes); // Steady 1 ms/commit plus a 120 ms burst every 40 commits — a slow, sawtoothing drain - // that holds the byte budget full between bursts without ever permanently wedging. + // that holds retention at the gate between bursts without ever permanently wedging. scenario.commit = CommitProfile { per_commit_delay: Duration::from_millis(1), burst: Some(CommitBurstStall { @@ -880,33 +881,49 @@ async fn fuzz_commit_stall() { scenario.deadline = Duration::from_secs(60); let (_, report) = run_checked("fuzz_commit_stall", scenario, 64).await; - // The byte budget is the only bound on the apply backlog: peak reserved bytes stay - // within the ceiling plus a small floor-bypass / request-boundary margin — i.e. the - // queue did not grow toward the 12.8 MB chain despite the commit stall. - let margin = (body_bytes as u64).saturating_mul(16); - let bound = byte_ceiling.saturating_add(margin); + // Same bound math as `fuzz_commit_stall_resident_plateau`: budget + one commit + // window of exempt bodies + a small request-boundary margin. (At this 400-block + // chain length the window slack alone covers the chain, so the plateau test's + // 1_200-block variant is the tight-bound check; this one pins the no-stall and + // quiescence behavior of the stall shape.) + // `usize → u64` widenings are lossless on all supported (64-bit) targets. + let window_slack = (MIN_BS_CHECKPOINT_SUBMITTED_BLOCK_APPLIES as u64) + .saturating_mul(body_bytes as u64) + .saturating_mul(DESERIALIZED_MEM_FACTOR); + let margin = (body_bytes as u64) + .saturating_mul(16) + .saturating_mul(DESERIALIZED_MEM_FACTOR); + let peak_retained_resident = report + .peak_retained_pipeline_wire_bytes + .saturating_mul(DESERIALIZED_MEM_FACTOR); + let bound = resident_budget + .saturating_add(window_slack) + .saturating_add(margin); assert!( - report.peak_budget_reserved <= bound, - "reserved bytes {} must stay within the {} B memory ceiling (+{} B margin); the \ - apply backlog is bounded by the byte budget, not unbounded by the commit stall", - report.peak_budget_reserved, - byte_ceiling, + peak_retained_resident <= bound, + "peak retained resident cost {} must stay within the {} B budget \ + (+{} B commit-window slack, +{} B margin); the apply backlog is bounded by \ + the resident gate, not unbounded by the commit stall", + peak_retained_resident, + resident_budget, + window_slack, margin, ); - // Non-vacuous: the stall actually filled the budget (otherwise the bound proves - // nothing about backpressure). + // Non-vacuous: the stall actually built real retention pressure (otherwise the + // bound proves nothing about backpressure). assert!( - report.peak_budget_reserved >= byte_ceiling / 2, - "the commit stall should have created real byte-budget backpressure (reserved \ - {} of the {} B ceiling)", - report.peak_budget_reserved, - byte_ceiling, + peak_retained_resident >= resident_budget / 2, + "the commit stall should have created real retained-memory pressure \ + (retained {} of the {} B budget)", + peak_retained_resident, + resident_budget, ); tracing::info!( - peak_budget_reserved = report.peak_budget_reserved, + peak_retained_pipeline_wire_bytes = report.peak_retained_pipeline_wire_bytes, + peak_retained_resident, final_budget_reserved = report.final_budget_reserved, - byte_ceiling, - "commit_stall byte-budget backpressure observation", + resident_budget, + "commit_stall retention backpressure observation", ); } @@ -1174,19 +1191,20 @@ async fn fuzz_reorder() { run_checked("fuzz_reorder", scenario, 32).await; } -/// Many peers racing against a tight global byte budget: every per-peer routine reserves -/// against the one shared `ByteBudget`, so this stresses the concurrent reservation path -/// end-to-end. A steady slow commit keeps the shared budget full so it actually binds -/// while eight routines reserve concurrently. `assert_core` asserts -/// `peak_budget_reserved` never exceeds the configured ceiling (the global memory bound -/// under multi-peer contention — the spec's "concurrent reservations MUST NOT -/// over-commit"); here we additionally assert the ceiling was genuinely approached, so -/// that bound is not vacuous. +/// Many peers racing against a tight global request budget: every per-peer routine +/// reserves against the one shared `ByteBudget`, so this stresses the concurrent +/// reservation path end-to-end. Eight fast peers' opening windows (64 requests × 32 KiB +/// each) far oversubscribe the 3 MiB ceiling, so the shared budget genuinely binds +/// while the routines race. `assert_core` asserts `peak_budget_reserved` never exceeds +/// the configured ceiling plus the bounded floor overdraft (the spec's "concurrent +/// reservations MUST NOT over-commit"); here we additionally assert the ceiling was +/// genuinely approached, so that bound is not vacuous. #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn fuzz_multi_peer_tight_budget() { let blocks = 400; let body_bytes = 32 * 1024usize; - // ~12.8 MB chain against a 3 MiB ceiling ⇒ the shared budget recycles ~4× and binds. + // 8 peers × 64-request opening windows × 32 KiB ⇒ ~16 MiB of instantaneous request + // demand against a 3 MiB ceiling, so reservations contend on every fill pass. let byte_ceiling: u64 = 3 * 1024 * 1024; let config = ZakuraBlockSyncConfig { max_inflight_block_bytes: byte_ceiling, @@ -1198,8 +1216,7 @@ async fn fuzz_multi_peer_tight_budget() { .collect(); let mut scenario = Scenario::new(blocks, 0x57ea_000f, config, peers); scenario.target_block_bytes = Some(body_bytes); - // A steady slow commit so the budget stays full and binds; instant commit would keep - // it nearly empty and make the bound vacuous. + // A steady slow commit stretches the run so many fill passes contend on the budget. scenario.commit = CommitProfile { per_commit_delay: Duration::from_millis(1), burst: None, @@ -1208,8 +1225,7 @@ async fn fuzz_multi_peer_tight_budget() { let (_, report) = run_checked("fuzz_multi_peer_tight_budget", scenario, 128).await; // Non-vacuous: the tight ceiling was genuinely approached under 8-peer contention, so - // `assert_core`'s `peak_budget_reserved <= max_inflight_block_bytes` proves a real - // bound rather than an idle budget. + // `assert_core`'s over-commit bound proves a real limit rather than an idle budget. assert!( report.peak_budget_reserved >= byte_ceiling / 2, "the tight budget should bind under 8-peer contention (reserved {} of {} B)", diff --git a/zebra-network/src/zakura/transport/guard.rs b/zebra-network/src/zakura/transport/guard.rs index d7f6977eea6..4418a35a531 100644 --- a/zebra-network/src/zakura/transport/guard.rs +++ b/zebra-network/src/zakura/transport/guard.rs @@ -132,28 +132,23 @@ impl ByteBudget { .ok(); } - /// Settle an estimated reservation to the actual bytes now held. - /// - /// If `actual` is smaller, this releases slack. If it is larger, this charges - /// the overshoot so held bodies are never under-counted. - #[cfg(test)] - pub(crate) fn settle(&mut self, reserved: u64, actual: u64) { - if actual > reserved { - self.charge(actual - reserved); - } else { - self.release(reserved - actual); - } - } - /// Audit the shared counter against an externally-derived expected value. /// - /// The expected value can be a cross-task snapshot, so transient handoff - /// skew is recorded as a metric rather than emitted as a warning. + /// The expected value is a cross-task snapshot, and every paired call site + /// orders its two ledgers the same way: reserve paths charge this budget + /// before marking the source ledger, and release paths drain the source + /// ledger before releasing here. Healthy handoff skew (e.g. a routine + /// parked between receipt and the post-forward release) therefore only + /// ever leaves the budget *above* the expected value, so an excess is not + /// counted as drift — a *persistent* excess (a leaked release) is caught + /// by the quiescence drain checks instead. Only a shortfall, which no + /// healthy interleaving can produce (a double release or lost charge), is + /// recorded as drift. /// - /// Returns `true` when the budget matches. + /// Returns `true` when the budget covers the expected value. pub(crate) fn audit(&self, expected: u64, _context: &'static str) -> bool { let actual = self.reserved(); - let ok = actual == expected; + let ok = actual >= expected; if !ok { metrics::counter!("sync.block.budget.audit_drift").increment(1); } @@ -338,20 +333,17 @@ mod tests { } #[test] - fn byte_budget_settles_estimates_to_actuals() { + fn byte_budget_charge_overdrafts_past_the_max() { + // `charge` bypasses the admission gate (the block-sync floor overdraft): + // it can push `reserved` past the max, and later releases drain it back. let mut budget = ByteBudget::new(1_000); - assert!(budget.try_reserve(300)); - budget.settle(300, 200); - assert_eq!(budget.reserved(), 200); - - assert!(budget.try_reserve(300)); - budget.settle(300, 300); - assert_eq!(budget.reserved(), 500); - - assert!(budget.try_reserve(300)); - budget.settle(300, 450); - assert_eq!(budget.reserved(), 950); - assert_eq!(budget.available(), 50); + assert!(budget.try_reserve(900)); + budget.charge(300); + assert_eq!(budget.reserved(), 1_200); + assert_eq!(budget.available(), 0); + assert!(!budget.try_reserve(1)); + budget.release(1_200); + assert_eq!(budget.reserved(), 0); } // A `ByteBudget` is cloned and shared across the block-sync Sequencer and every