feat(network)!: regulate GetBlocks with paired streams

README.md

@@ -115,7 +115,7 @@ cargo install --locked zakura
 Alternatively, you can install it from GitHub:
 
 ```sh
-cargo install --git https://github.com/zakura-core/zakura --tag v1.3.2 zakura
+cargo install --git https://github.com/zakura-core/zakura --tag v1.3.3 zakura
 ```
 
 You can start Zakura by running

crates/zakura-consensus/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "zakura-consensus"
-version = "7.0.1"
+version = "7.0.2"
 authors.workspace = true
 description = "Implementation of Zcash consensus checks for the Zakura node. Internal crate, published to support cargo install zakura"
 license.workspace = true
@@ -68,7 +68,7 @@ tower-fallback = { package = "zakura-tower-fallback", path = "../tower-fallback/
 tower-batch-control = { package = "zakura-tower-batch-control", path = "../tower-batch-control/", version = "1.3.0" }
 
 zakura-script = { path = "../zakura-script", version = "3.2.1" }
-zakura-state = { path = "../zakura-state", version = "7.0.0" }
+zakura-state = { path = "../zakura-state", version = "8.0.0" }
 zakura-node-services = { path = "../zakura-node-services", version = "3.2.1" }
 zakura-chain = { path = "../zakura-chain", version = "6.0.0" }
 zakura-header-chain = { path = "../zakura-header-chain", version = "2.0.0" }
@@ -93,7 +93,7 @@ toml = { workspace = true }
 
 tokio = { workspace = true, features = ["full", "tracing", "test-util"] }
 
-zakura-state = { path = "../zakura-state", version = "7.0.0", features = ["proptest-impl"] }
+zakura-state = { path = "../zakura-state", version = "8.0.0", features = ["proptest-impl"] }
 zakura-chain = { path = "../zakura-chain", version = "6.0.0", features = ["proptest-impl"] }
 zakura-test = { path = "../zakura-test/", version = "2.1.0" }
 

crates/zakura-network/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "zakura-network"
-version = "7.0.1"
+version = "8.0.0"
 authors.workspace = true
 description = "Networking code for the Zakura node. Internal crate, published to support cargo install zakura"
 # # Legal

crates/zakura-network/src/zakura/block_sync/config.rs

@@ -103,6 +103,30 @@ pub const DEFAULT_BS_SIZE_DEVIATION_TOLERANCE: u32 = 200;
 /// only controls how many bounded body frames a server sends before `BlocksDone`.
 pub const MAX_BS_RESPONSE_BYTES: u32 = DEFAULT_BS_MAX_RESPONSE_BYTES;
 
+/// Fixed byte-equivalent work charged for each admitted `GetBlocks` request.
+///
+/// This follows the initial peer-message regulation design. The response-byte
+/// portion is refunded when unused, but this request-processing cost is not.
+pub const GET_BLOCKS_REQUEST_OVERHEAD_BYTES: u64 = 64 * 1024;
+/// Encoded payload bytes in either terminal GetBlocks response message.
+pub const GET_BLOCKS_TERMINAL_PAYLOAD_BYTES: u64 = 9;
+
+const MIB: u64 = 1024 * 1024;
+const DEFAULT_GET_BLOCKS_PEER_RATE_BYTES_PER_SECOND: u64 = 128 * MIB;
+// These wire caps are `u32`, so widening them to `u64` is lossless.
+const DEFAULT_GET_BLOCKS_PEER_RATE_CAPACITY_BYTES: u64 = MAX_BS_RESPONSE_BYTES as u64
+    + MAX_BS_BLOCKS_PER_REQUEST as u64
+    + GET_BLOCKS_TERMINAL_PAYLOAD_BYTES
+    + GET_BLOCKS_REQUEST_OVERHEAD_BYTES;
+const DEFAULT_GET_BLOCKS_PEER_OUTSTANDING_BYTES: u64 = 64 * MIB;
+const DEFAULT_GET_BLOCKS_NODE_RATE_BYTES_PER_SECOND: u64 = 256 * MIB;
+const DEFAULT_GET_BLOCKS_NODE_RATE_CAPACITY_BYTES: u64 = 128 * MIB;
+const DEFAULT_GET_BLOCKS_NODE_OUTSTANDING_BYTES: u64 = 256 * MIB;
+const DEFAULT_GET_BLOCKS_PEER_PENDING_REQUESTS: usize = 64;
+const DEFAULT_GET_BLOCKS_NODE_PENDING_REQUESTS: usize = 1024;
+const DEFAULT_GET_BLOCKS_NODE_ACTIVE_REQUESTS: usize = 64;
+const DEFAULT_GET_BLOCKS_QUERY_TIMEOUT: Duration = Duration::from_secs(8);
+
 /// Default steady-state cwnd gain, percent of the bandwidth-delay product. 300% ramps a
 /// proven peer up as `1 → 3 → 9 …`; the reliability discount and delay-gradient ceiling
 /// pull it back if the extra concurrency costs drops or standing queue.
@@ -233,7 +257,9 @@ pub struct ZakuraBlockSyncConfig {
     /// Initial per-peer BBR cwnd (cold-start point), in blocks; converges to the
     /// BDP-derived target once the first delivery is measured.
     pub initial_inflight_requests: u32,
-    /// Maximum total response bytes this node advertises per `GetBlocks` response.
+    /// Maximum serialized block bytes this node advertises per `GetBlocks` response.
+    /// Must be at least [`block::MAX_BLOCK_BYTES`] so any single valid block fits.
+    /// Message discriminators and the response terminator are reserved separately.
     pub max_response_bytes: u32,
     /// Maximum estimated bytes reserved for outstanding block-body requests: a
     /// DoS/pacing bound on in-flight wire data, released at receipt. Received
@@ -316,6 +342,61 @@ pub struct ZakuraBlockSyncConfig {
     pub floor_bypass_slots: u32,
     /// Block-sync peer caps and queue limits owned by this service.
     pub peer_limits: ServicePeerLimits,
+    /// Resource policy for serving inbound `GetBlocks` requests.
+    pub get_blocks_regulation: GetBlocksRegulationConfig,
+}
+
+/// Node and peer bounds applied before GetBlocks state work starts.
+///
+/// Rate budgets bound bursts and sustained work. Outstanding-byte budgets do
+/// not refill; capacity returns when the last ledger, query, or result owner
+/// drops, or when its queued frame leaves the application-owned transport path.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct GetBlocksRegulationConfig {
+    /// Byte-equivalent work retained for each admitted request regardless of response size.
+    pub request_overhead_bytes: u64,
+    /// Sustained byte-equivalent serving allowance for one authenticated peer.
+    pub peer_rate_bytes_per_second: u64,
+    /// Burst allowance retained across reconnects for one authenticated peer.
+    pub peer_rate_capacity_bytes: u64,
+    /// Reserved and queued response bytes allowed for one live peer session.
+    pub peer_outstanding_bytes: u64,
+    /// Sustained aggregate byte-equivalent serving allowance for GetBlocks.
+    pub node_rate_bytes_per_second: u64,
+    /// Aggregate GetBlocks burst allowance for this node.
+    pub node_rate_capacity_bytes: u64,
+    /// Admitted GetBlocks response bytes not yet handed off by the transport.
+    pub node_outstanding_bytes: u64,
+    /// Decoded GetBlocks requests retained while one peer waits for admission.
+    pub peer_pending_requests: usize,
+    /// Fully retained GetBlocks queue entries across all peers. A full queue
+    /// additionally permits one blocked decoded input per live session.
+    pub node_pending_requests: usize,
+    /// State queries and responses that may remain active across all peers.
+    pub node_active_requests: usize,
+    /// Response deadline for a state query. Timed-out reads keep their resource
+    /// charges until the underlying state work finishes.
+    #[serde(with = "humantime_serde")]
+    pub query_timeout: Duration,
+}
+
+impl Default for GetBlocksRegulationConfig {
+    fn default() -> Self {
+        Self {
+            request_overhead_bytes: GET_BLOCKS_REQUEST_OVERHEAD_BYTES,
+            peer_rate_bytes_per_second: DEFAULT_GET_BLOCKS_PEER_RATE_BYTES_PER_SECOND,
+            peer_rate_capacity_bytes: DEFAULT_GET_BLOCKS_PEER_RATE_CAPACITY_BYTES,
+            peer_outstanding_bytes: DEFAULT_GET_BLOCKS_PEER_OUTSTANDING_BYTES,
+            node_rate_bytes_per_second: DEFAULT_GET_BLOCKS_NODE_RATE_BYTES_PER_SECOND,
+            node_rate_capacity_bytes: DEFAULT_GET_BLOCKS_NODE_RATE_CAPACITY_BYTES,
+            node_outstanding_bytes: DEFAULT_GET_BLOCKS_NODE_OUTSTANDING_BYTES,
+            peer_pending_requests: DEFAULT_GET_BLOCKS_PEER_PENDING_REQUESTS,
+            node_pending_requests: DEFAULT_GET_BLOCKS_NODE_PENDING_REQUESTS,
+            node_active_requests: DEFAULT_GET_BLOCKS_NODE_ACTIVE_REQUESTS,
+            query_timeout: DEFAULT_GET_BLOCKS_QUERY_TIMEOUT,
+        }
+    }
 }
 
 fn deserialize_ignored_replace_legacy_syncer<'de, D>(deserializer: D) -> Result<bool, D::Error>
@@ -359,6 +440,7 @@ impl Default for ZakuraBlockSyncConfig {
             bbr_cwnd_unit: CwndUnit::Bytes,
             floor_bypass_slots: DEFAULT_BS_FLOOR_BYPASS_SLOTS,
             peer_limits: ServicePeerLimits::default(),
+            get_blocks_regulation: GetBlocksRegulationConfig::default(),
         }
     }
 }
@@ -465,6 +547,7 @@ impl ZakuraBlockSyncConfig {
         if self.bbr_probe_rtt_interval <= self.bbr_probe_rtt_duration {
             return Err("bbr_probe_rtt_interval must exceed bbr_probe_rtt_duration");
         }
+        super::serving_regulation::validate_config(self)?;
         Ok(())
     }
 

crates/zakura-network/src/zakura/block_sync/events.rs

@@ -1,7 +1,7 @@
 #[cfg(any(test, feature = "proptest-impl"))]
 use super::state::BlockSyncFrontiers;
 use super::{request::*, *};
-use std::num::NonZeroU64;
+use std::{fmt, num::NonZeroU64};
 
 /// Committed header metadata used by block sync to schedule and validate a body.
 #[derive(Copy, Clone, Debug, Eq, PartialEq)]
@@ -80,6 +80,8 @@ pub enum BlockSyncEvent {
     },
     /// Node wiring finished or abandoned a `Block` response to an inbound `GetBlocks`.
     BlockRangeResponseFinished {
+        /// Exact inbound request being completed.
+        request_id: BlockRangeRequestId,
         /// Peer whose served-response slot can be released.
         peer: ZakuraPeerId,
         /// First requested height.
@@ -91,6 +93,10 @@ pub enum BlockSyncEvent {
     },
     /// State returned committed bodies requested by a peer and the reactor should send them.
     BlockRangeResponseReady {
+        /// Keep the returned blocks charged until the reactor consumes or drops them.
+        lease: BlockRangeQueryLease,
+        /// Exact inbound request being completed.
+        request_id: BlockRangeRequestId,
         /// Peer whose inbound request is being served.
         peer: ZakuraPeerId,
         /// First requested height.
@@ -102,6 +108,24 @@ pub enum BlockSyncEvent {
     },
 }
 
+/// Session lifecycle facts sent from [`BlockSyncService`] to the reactor.
+///
+/// This internal channel carries the session generation needed to reject stale
+/// connect and disconnect events without exposing transport bookkeeping in the
+/// public driver API.
+#[derive(Clone, Debug)]
+pub(super) enum BlockSyncPeerLifecycleEvent {
+    /// A newly admitted stream session is ready for reactor bookkeeping.
+    Connected(BlockSyncPeerSession),
+    /// One exact stream session has ended.
+    Disconnected {
+        /// Peer whose stream session ended.
+        peer: ZakuraPeerId,
+        /// Generation assigned when the session was admitted.
+        session_id: u64,
+    },
+}
+
 /// Result of applying a block-sync body through the verifier driver.
 #[derive(Copy, Clone, Debug, Eq, PartialEq)]
 pub enum BlockApplyResult {
@@ -253,6 +277,34 @@ impl BlockApplyOutcome {
 /// ignore those stale completions instead of releasing a newer in-flight body.
 pub type BlockApplyToken = u64;
 
+/// Monotonic identity assigned to each inbound `GetBlocks` request.
+///
+/// The state driver echoes this identity so a delayed completion cannot settle
+/// a newer request or send blocks through a replacement peer session.
+#[derive(Copy, Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct BlockRangeRequestId(NonZeroU64);
+
+impl BlockRangeRequestId {
+    /// Construct an ID, returning `None` for the reserved zero value.
+    pub const fn new(value: u64) -> Option<Self> {
+        match NonZeroU64::new(value) {
+            Some(value) => Some(Self(value)),
+            None => None,
+        }
+    }
+
+    /// Return the nonzero integer carried by this request identity.
+    pub const fn get(self) -> u64 {
+        self.0.get()
+    }
+}
+
+impl fmt::Display for BlockRangeRequestId {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(formatter)
+    }
+}
+
 /// Actions emitted by the future block-sync reactor for the service seam.
 #[derive(Clone, Debug)]
 pub enum BlockSyncAction {
@@ -271,12 +323,21 @@ pub enum BlockSyncAction {
     },
     /// Ask node wiring to read committed bodies for an inbound `GetBlocks`.
     QueryBlocksByHeightRange {
+        /// Resource ownership to retain through the actual state read and response.
+        lease: BlockRangeQueryLease,
+        /// Exact inbound request identity to echo in the response event.
+        request_id: BlockRangeRequestId,
         /// Peer that requested the range.
         peer: ZakuraPeerId,
         /// First height.
         start: block::Height,
         /// Maximum count.
         count: u32,
+        /// Maximum total encoded block-body bytes the state result may contain.
+        max_response_bytes: u32,
+        /// Response deadline. The driver retains capacity until the underlying
+        /// state future completes, including after this timeout.
+        timeout: Duration,
     },
     /// Parent-first body ready for B3's verifier/commit driver.
     SubmitBlock {
@@ -351,7 +412,7 @@ pub enum BlockSyncMisbehavior {
     UnsolicitedBlock,
     /// A peer requested more blocks than this node advertised it can serve.
     GetBlocksTooLong,
-    /// A peer exceeded this node's inbound `GetBlocks` serving budget.
+    /// A peer requested blocks before sending its required `Status`.
     GetBlocksSpam,
     /// A peer supplied a body whose payload does not match its requested header.
     BodyPayloadMismatch(zakura_header_chain::BodyPayloadMismatch),
@@ -379,7 +440,7 @@ pub enum BlockSyncMisbehavior {
 /// misbehavior aggregation) over this channel. The sender is `try_send`/bounded
 /// so a busy reactor never backpressures a routine's decode loop into stalling
 /// its transport (the only blocking routine send is the Sequencer `AcceptBody`).
-#[derive(Clone, Debug)]
+#[derive(Debug)]
 pub(super) enum RoutineToReactor {
     /// A routine received a `Status` and updated its own servable/caps + the
     /// registry. The reactor advertises our `Status` reply and republishes the
@@ -396,10 +457,10 @@ pub(super) enum RoutineToReactor {
     ServeGetBlocks {
         /// Peer that requested the range.
         peer: ZakuraPeerId,
-        /// First requested height.
-        start_height: block::Height,
-        /// Requested block count.
-        count: u32,
+        /// Decoded fields and their bounded pending-input ownership.
+        request: super::serving_regulation::PendingGetBlocksRequest,
+        /// Provisional resource ownership from the originating peer session.
+        attempt: super::serving_regulation::AdmissionAttempt,
     },
     /// A routine drained its pending work; the producer should re-query (it
     /// self-gates on low-water, so the ping is idempotent/cheap).

crates/zakura-network/src/zakura/block_sync/mod.rs

@@ -46,6 +46,7 @@ mod request;
 mod sequencer;
 mod sequencer_task;
 mod service;
+mod serving_regulation;
 mod state;
 #[cfg(test)]
 mod tests;
@@ -64,11 +65,14 @@ pub use bench::{
 };
 #[cfg(test)]
 pub(crate) use config::MIN_BS_CHECKPOINT_SUBMITTED_BLOCK_APPLIES;
-pub use config::{BlockSyncStatus, CwndUnit, ZakuraBlockSyncConfig, MAX_BS_RESPONSE_BYTES};
+pub use config::{
+    BlockSyncStatus, CwndUnit, GetBlocksRegulationConfig, ZakuraBlockSyncConfig,
+    MAX_BS_RESPONSE_BYTES,
+};
 pub use error::BlockSyncWireError;
 pub use events::{
-    BlockApplyOutcome, BlockApplyResult, BlockApplyToken, BlockSyncAction, BlockSyncBlockMeta,
-    BlockSyncEvent, BlockSyncMisbehavior,
+    BlockApplyOutcome, BlockApplyResult, BlockApplyToken, BlockRangeRequestId, BlockSyncAction,
+    BlockSyncBlockMeta, BlockSyncEvent, BlockSyncMisbehavior,
 };
 pub use reactor::spawn_block_sync_reactor;
 pub use request::BlockSizeEstimate;
@@ -145,3 +149,7 @@ pub(crate) fn test_block_apply_outcome(result: BlockApplyResult) -> BlockApplyOu
         }
     }
 }
+
+#[cfg(any(test, feature = "zakura-testkit"))]
+pub(crate) use serving_regulation::query_lease_for_test;
+pub use serving_regulation::BlockRangeQueryLease;