feat(network)!: regulate GetBlocks with paired streams

crates/zakura-consensus/Cargo.toml

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

crates/zakura-header-chain/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "zakura-header-chain"
-version = "2.0.1-rc0"
+version = "2.0.1-rc1"
 authors.workspace = true
 description = "Fork-aware header-chain domain types and transition engine for Zakura"
 license.workspace = true

crates/zakura-network/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "zakura-network"
-version = "7.1.0-rc0"
+version = "8.0.0"
 authors.workspace = true
 description = "Networking code for the Zakura node. Internal crate, published to support cargo install zakura"
 # # Legal
@@ -83,9 +83,9 @@ proptest = { workspace = true, optional = true }
 proptest-derive = { workspace = true, optional = true }
 
 zakura-chain = { path = "../zakura-chain", version = "7.0.0-rc0", features = ["async-error"] }
-zakura-header-chain = { path = "../zakura-header-chain", version = "2.0.1-rc0" }
+zakura-header-chain = { path = "../zakura-header-chain", version = "2.0.1-rc1" }
 zakura-jsonl-trace = { path = "../zakura-jsonl-trace", version = "1.2.0" }
-zakura-node-services = { path = "../zakura-node-services", version = "3.2.3-rc0" }
+zakura-node-services = { path = "../zakura-node-services", version = "3.2.3-rc1" }
 
 [dev-dependencies]
 proptest = { workspace = true }

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

@@ -103,6 +103,12 @@ 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;
 
+/// Encoded payload bytes in either terminal GetBlocks response message.
+pub const GET_BLOCKS_TERMINAL_PAYLOAD_BYTES: u64 = 9;
+
+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 +239,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 +324,32 @@ 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.
+///
+/// Each session has one response producer, held through state work and transport
+/// writes. Node capacity returns when the last query, result, or frame owner drops.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields, default)]
+pub struct GetBlocksRegulationConfig {
+    /// 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 {
+            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 +393,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 +500,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 {
@@ -349,10 +410,8 @@ pub enum BlockSyncMisbehavior {
     MalformedMessage,
     /// A peer sent blocks that were not requested.
     UnsolicitedBlock,
-    /// A peer requested more blocks than this node advertised it can serve.
-    GetBlocksTooLong,
-    /// A peer exceeded this node's inbound `GetBlocks` serving budget.
-    GetBlocksSpam,
+    /// A peer requested blocks before sending its required `Status`.
+    GetBlocksBeforeStatus,
     /// A peer supplied a body whose payload does not match its requested header.
     BodyPayloadMismatch(zakura_header_chain::BodyPayloadMismatch),
     /// A commitment-matching body deterministically failed consensus.
@@ -376,10 +435,9 @@ pub enum BlockSyncMisbehavior {
 /// Each per-peer pipe-routine ([`PeerRoutine`](super::peer_routine)) decodes its
 /// own frames and runs the download logic locally; it forwards only the concerns
 /// that need reactor-global state (serving, status advertisement, the producer,
-/// 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)]
+/// misbehavior aggregation) over this channel. Serving waits for channel capacity
+/// with stream reads paused; control notifications use bounded `try_send`.
+#[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 +454,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 of the admitted request.
+        request: super::serving_regulation::GetBlocksRequest,
+        /// 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).