This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/iggy-server-hardening in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b3ea6f84c0100030c11041da03600e9bf7f1e2f3 Author: Hubert Gruszecki <[email protected]> AuthorDate: Fri Jul 24 12:44:39 2026 +0200 feat(server-ng): make HTTP write admission caps configurable The HTTP write admission caps (128 in-flight writes global on the shard-0 budget, 32 per session) were module-private consts, untunable without a rebuild, yet they are the operator's lever bounding buffered write memory (cap x http.max_request_size). Surface them as a new [http_admission] section following the [metadata] precedent: the ng http section reuses the shared legacy HttpConfig struct, so ng-only knobs cannot live there without leaking into the legacy config. Defaults stay 128/32 in the embedded config.toml, pinned by a lockstep test; validation rejects zero, a per-session cap above the global cap, and values past a 4096 ceiling. Admission logic takes the caps as parameters, behavior at defaults unchanged. --- core/configs/src/lib.rs | 5 +- core/configs/src/server_ng_config/defaults.rs | 16 ++ .../configs/src/server_ng_config/http_admission.rs | 184 +++++++++++++++++++++ core/configs/src/server_ng_config/mod.rs | 1 + core/configs/src/server_ng_config/server_ng.rs | 2 + core/configs/src/server_ng_config/validators.rs | 5 + core/server-ng/config.toml | 14 ++ core/server-ng/src/bootstrap.rs | 1 + core/server-ng/src/http.rs | 4 + core/server-ng/src/http/admission.rs | 92 ++++++----- core/server-ng/src/http/error.rs | 22 +-- core/server-ng/src/http/session.rs | 2 +- core/server-ng/src/http/state.rs | 12 +- core/server-ng/src/http/submit.rs | 14 +- 14 files changed, 311 insertions(+), 63 deletions(-) diff --git a/core/configs/src/lib.rs b/core/configs/src/lib.rs index 5c82dbd3a..aca084501 100644 --- a/core/configs/src/lib.rs +++ b/core/configs/src/lib.rs @@ -30,6 +30,7 @@ pub use server_config::{ tcp, validators, websocket, }; pub use server_ng_config::{ - COMPONENT_NG, cluster as ng_cluster, message_bus, metadata as ng_metadata, quic as ng_quic, - server_ng, sharding as ng_sharding, tcp as ng_tcp, websocket as ng_websocket, + COMPONENT_NG, cluster as ng_cluster, http_admission as ng_http_admission, message_bus, + metadata as ng_metadata, quic as ng_quic, server_ng, sharding as ng_sharding, tcp as ng_tcp, + websocket as ng_websocket, }; diff --git a/core/configs/src/server_ng_config/defaults.rs b/core/configs/src/server_ng_config/defaults.rs index 40e47d68b..d464f2190 100644 --- a/core/configs/src/server_ng_config/defaults.rs +++ b/core/configs/src/server_ng_config/defaults.rs @@ -30,6 +30,7 @@ use super::cluster::{ ClusterAuthConfig, ClusterConfig, ClusterNodeConfig, ClusterTlsConfig, TransportPorts, }; +use super::http_admission::HttpAdmissionConfig; use super::message_bus::MessageBusConfig; use super::metadata::MetadataConfig; use super::quic::{QuicCertificateConfig, QuicConfig, QuicSocketConfig}; @@ -63,6 +64,7 @@ impl Default for ServerNgConfig { tcp: TcpConfig::default(), websocket: WebSocketConfig::default(), http: HttpConfig::default(), + http_admission: HttpAdmissionConfig::default(), telemetry: TelemetryConfig::default(), cluster: ClusterConfig::default(), metadata: MetadataConfig::default(), @@ -140,6 +142,20 @@ impl Default for MetadataConfig { } } +impl Default for HttpAdmissionConfig { + fn default() -> HttpAdmissionConfig { + // Read from the embedded TOML so the Default impl and the on-disk + // schema cannot drift; the lockstep test pins these to the canonical + // caps. + let http_admission = &SERVER_NG_CONFIG.http_admission; + HttpAdmissionConfig { + max_in_flight_writes: http_admission.max_in_flight_writes as u32, + max_in_flight_writes_per_session: http_admission.max_in_flight_writes_per_session + as u32, + } + } +} + impl Default for QuicConfig { fn default() -> QuicConfig { QuicConfig { diff --git a/core/configs/src/server_ng_config/http_admission.rs b/core/configs/src/server_ng_config/http_admission.rs new file mode 100644 index 000000000..5e99de1d5 --- /dev/null +++ b/core/configs/src/server_ng_config/http_admission.rs @@ -0,0 +1,184 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! On-disk schema for shard-0 HTTP write admission. +//! +//! Two in-flight caps previously hardcoded in the server-ng HTTP write +//! path, bounding awaited partition writes (produce / consumer-offset) +//! that each park a handler and pin a buffered body while its +//! decode/encode/HS256 CPU runs on the single core that also pumps +//! consensus: +//! +//! - `max_in_flight_writes` - shard-0-global budget across all sessions. +//! - `max_in_flight_writes_per_session` - one credential's slice of that +//! budget, so a session that outruns its own commits saturates itself +//! (429) before it can starve the shared budget. +//! +//! The defaults here are the canonical caps; `core/server-ng`'s HTTP +//! layer reads the configured values at boot, and the lockstep test +//! below pins the shipped `config.toml` defaults to these constants so a +//! default deployment keeps the historical behavior. + +use super::COMPONENT_NG; +use crate::ConfigurationError; +use configs::ConfigEnv; +use iggy_common::Validatable; +use serde::{Deserialize, Serialize}; + +/// Canonical shard-0-global in-flight write budget. +pub const DEFAULT_MAX_IN_FLIGHT_WRITES: u32 = 128; + +/// Canonical per-session in-flight write cap. +pub const DEFAULT_MAX_IN_FLIGHT_WRITES_PER_SESSION: u32 = 32; + +/// Upper bound on `max_in_flight_writes`. Every awaited write pins a +/// buffered request body (budget x `http.max_request_size`) on the single +/// shard-0 core; four thousand in-flight writes is far past any sane +/// deployment and a likely unit typo. +pub const MAX_IN_FLIGHT_WRITES_CEILING: u32 = 4096; + +/// Admission caps for shard-0 HTTP awaited partition writes. +#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] +pub struct HttpAdmissionConfig { + /// Shard-0-global budget for concurrently awaited partition writes + /// across every session. Bounds the worst-case buffered bytes + /// (budget x `http.max_request_size`) and how far admitted HTTP work + /// can delay the consensus pump (budget x per-request CPU). Refusals + /// past it are the server-busy 503. + pub max_in_flight_writes: u32, + + /// Per-session cap on concurrently awaited partition writes, bounding + /// how much of `max_in_flight_writes` one credential may occupy. + /// Refusals past it are the too-many-in-flight 429, a session's own + /// backpressure signal before it can starve the shared budget. Must + /// not exceed `max_in_flight_writes`. + pub max_in_flight_writes_per_session: u32, +} + +impl Validatable<ConfigurationError> for HttpAdmissionConfig { + fn validate(&self) -> Result<(), ConfigurationError> { + if self.max_in_flight_writes == 0 { + eprintln!("{COMPONENT_NG} http_admission.max_in_flight_writes must be > 0"); + return Err(ConfigurationError::InvalidConfigurationValue); + } + // One ceiling suffices: per_session may not exceed the global budget + // (checked below), so this bounds it transitively. + if self.max_in_flight_writes > MAX_IN_FLIGHT_WRITES_CEILING { + eprintln!( + "{COMPONENT_NG} http_admission.max_in_flight_writes ({}) exceeds the maximum ({MAX_IN_FLIGHT_WRITES_CEILING})", + self.max_in_flight_writes + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.max_in_flight_writes_per_session == 0 { + eprintln!("{COMPONENT_NG} http_admission.max_in_flight_writes_per_session must be > 0"); + return Err(ConfigurationError::InvalidConfigurationValue); + } + // A per-session cap above the global budget can never bind: the + // session would always hit the shared 503 first, so the 429 signal + // is dead. Reject the nonsensical ordering (equal is fine - a + // single session may then fill the whole budget). + if self.max_in_flight_writes_per_session > self.max_in_flight_writes { + eprintln!( + "{COMPONENT_NG} http_admission.max_in_flight_writes_per_session ({}) must not exceed http_admission.max_in_flight_writes ({})", + self.max_in_flight_writes_per_session, self.max_in_flight_writes + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn cfg(global: u32, per_session: u32) -> HttpAdmissionConfig { + HttpAdmissionConfig { + max_in_flight_writes: global, + max_in_flight_writes_per_session: per_session, + } + } + + #[test] + fn embedded_defaults_match_canonical_caps() { + // Behavior lockstep: `Default` reads the shipped config.toml, which must + // carry the caps the HTTP admission path historically hardcoded so a + // default deployment is unaffected by surfacing them as config. + let config = HttpAdmissionConfig::default(); + assert_eq!(config.max_in_flight_writes, DEFAULT_MAX_IN_FLIGHT_WRITES); + assert_eq!( + config.max_in_flight_writes_per_session, + DEFAULT_MAX_IN_FLIGHT_WRITES_PER_SESSION + ); + } + + #[test] + fn default_impl_validates() { + assert!(HttpAdmissionConfig::default().validate().is_ok()); + } + + #[test] + fn rejects_zero_global() { + assert!( + cfg(0, DEFAULT_MAX_IN_FLIGHT_WRITES_PER_SESSION) + .validate() + .is_err() + ); + } + + #[test] + fn rejects_zero_per_session() { + assert!(cfg(DEFAULT_MAX_IN_FLIGHT_WRITES, 0).validate().is_err()); + } + + #[test] + fn rejects_per_session_above_global() { + assert!(cfg(32, 33).validate().is_err()); + } + + #[test] + fn accepts_per_session_equal_global() { + // A single session may fill the whole budget; equal is the boundary. + assert!(cfg(32, 32).validate().is_ok()); + } + + #[test] + fn rejects_above_ceiling() { + assert!( + cfg( + MAX_IN_FLIGHT_WRITES_CEILING + 1, + DEFAULT_MAX_IN_FLIGHT_WRITES_PER_SESSION + ) + .validate() + .is_err() + ); + } + + #[test] + fn accepts_at_ceiling() { + // Inclusive ceiling; per-session rides under it via per_session <= global. + assert!( + cfg( + MAX_IN_FLIGHT_WRITES_CEILING, + DEFAULT_MAX_IN_FLIGHT_WRITES_PER_SESSION + ) + .validate() + .is_ok() + ); + } +} diff --git a/core/configs/src/server_ng_config/mod.rs b/core/configs/src/server_ng_config/mod.rs index 159d0c2f5..88980c252 100644 --- a/core/configs/src/server_ng_config/mod.rs +++ b/core/configs/src/server_ng_config/mod.rs @@ -30,6 +30,7 @@ pub mod cluster; pub mod defaults; pub mod displays; +pub mod http_admission; pub mod message_bus; pub mod metadata; pub mod quic; diff --git a/core/configs/src/server_ng_config/server_ng.rs b/core/configs/src/server_ng_config/server_ng.rs index 3cd8f5376..daa7b5036 100644 --- a/core/configs/src/server_ng_config/server_ng.rs +++ b/core/configs/src/server_ng_config/server_ng.rs @@ -17,6 +17,7 @@ use super::COMPONENT_NG; use super::cluster::ClusterConfig; +use super::http_admission::HttpAdmissionConfig; use super::message_bus::MessageBusConfig; use super::metadata::MetadataConfig; use super::quic::QuicConfig; @@ -74,6 +75,7 @@ pub struct ServerNgConfig { pub quic: QuicConfig, pub tcp: TcpConfig, pub http: HttpConfig, + pub http_admission: HttpAdmissionConfig, pub websocket: WebSocketConfig, pub telemetry: TelemetryConfig, pub cluster: ClusterConfig, diff --git a/core/configs/src/server_ng_config/validators.rs b/core/configs/src/server_ng_config/validators.rs index 1c9745b8c..9063bf747 100644 --- a/core/configs/src/server_ng_config/validators.rs +++ b/core/configs/src/server_ng_config/validators.rs @@ -83,6 +83,11 @@ impl Validatable<ConfigurationError> for ServerNgConfig { self.metadata.validate().error(|e: &ConfigurationError| { format!("{COMPONENT_NG} (error: {e}) - failed to validate metadata config") })?; + self.http_admission + .validate() + .error(|e: &ConfigurationError| { + format!("{COMPONENT_NG} (error: {e}) - failed to validate http admission config") + })?; self.system .logging .validate() diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index 08fea06d4..7dfb2b8b8 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -173,6 +173,20 @@ cert_file = "core/certs/iggy_cert.pem" # Path to the TLS key file. key_file = "core/certs/iggy_key.pem" +# Admission caps for shard-0 HTTP awaited partition writes (produce and +# consumer-offset writes that park a handler and pin a buffered body until the +# write commits). These bound how much buffered memory and consensus-pump delay +# concurrent HTTP writes can impose on shard 0. +[http_admission] +# Shard-0-global budget for concurrently awaited partition writes across every +# session. Requests past it are refused with a server-busy 503. Must be > 0. +max_in_flight_writes = 128 + +# Per-session cap on concurrently awaited partition writes, so one credential +# that outruns its own commits gets a too-many-in-flight 429 before it can +# starve the shared budget. Must be > 0 and must not exceed max_in_flight_writes. +max_in_flight_writes_per_session = 32 + # TCP server configuration. [tcp] # Determines if the TCP server is active. diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index 3212c944d..a626b6a3c 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -2251,6 +2251,7 @@ async fn start_tcp_runtime( shard, http_addr, &config.http, + &config.http_admission, &config.cluster, Arc::clone(&config.system), self_ports, diff --git a/core/server-ng/src/http.rs b/core/server-ng/src/http.rs index 37a5535b4..f884006dc 100644 --- a/core/server-ng/src/http.rs +++ b/core/server-ng/src/http.rs @@ -52,6 +52,7 @@ use axum::middleware::{Next, from_fn}; use axum::routing::{delete, get, post, put}; use configs::http::{HttpConfig, HttpCorsConfig}; use configs::ng_cluster::{ClusterConfig, TransportPorts}; +use configs::ng_http_admission::HttpAdmissionConfig; use configs::server_ng::NgSystemConfig; use iggy_common::IggyError; use message_bus::client_listener; @@ -93,6 +94,7 @@ pub async fn start( shard: &Rc<ServerNgShard>, addr: SocketAddr, http_config: &HttpConfig, + admission: &HttpAdmissionConfig, cluster: &ClusterConfig, system_config: Arc<NgSystemConfig>, self_ports: TransportPorts, @@ -129,6 +131,8 @@ pub async fn start( // never consulted here. metadata_view: Arc::new(AtomicU64::new(crate::cluster_meta::METADATA_VIEW_UNKNOWN)), }, + max_in_flight_writes: admission.max_in_flight_writes, + max_in_flight_writes_per_session: admission.max_in_flight_writes_per_session, in_flight_writes: Cell::new(0), })); // Saturating: a configured limit past the pointer width (32-bit target, diff --git a/core/server-ng/src/http/admission.rs b/core/server-ng/src/http/admission.rs index e88f79a00..afb614bed 100644 --- a/core/server-ng/src/http/admission.rs +++ b/core/server-ng/src/http/admission.rs @@ -23,27 +23,6 @@ use std::cell::Cell; use crate::http::error::PartitionWriteError; -/// Per-session cap on concurrently awaited partition writes (produce / -/// consumer-offset). Bounds how much of [`MAX_IN_FLIGHT_WRITES_GLOBAL`] one -/// credential can occupy, so a session that outruns its own commits saturates -/// itself (429, its own backpressure signal) before it can starve every other -/// session out of the shared budget. -const MAX_IN_FLIGHT_WRITES_PER_SESSION: u32 = 32; - -/// Shard-0-global budget for concurrently awaited partition writes, across all -/// sessions. Every admitted write parks a handler for up to -/// [`PARTITION_WRITE_REPLY_TIMEOUT`] while pinning its buffered body, and its -/// decode/encode/HS256 CPU runs on the same single-threaded core that pumps -/// consensus. The budget therefore bounds both starvation terms: budget x -/// `max_request_size` bounds the worst-case buffered bytes, and budget x -/// per-request CPU bounds how far admitted HTTP work can delay the consensus -/// pump. `?ack=none` produces are admitted through the same caps: they install -/// no reply slot and never await a commit, but they still park inside dispatch -/// for the routable-wait budget while pinning their buffered body, so leaving -/// them uncapped would bypass both terms. A session that saturates its own cap -/// reads its own 429 before it can spill onto the shared budget. -const MAX_IN_FLIGHT_WRITES_GLOBAL: u32 = 128; - /// In-flight admission token for one awaited partition write. One guard owns /// both releases (session + global) so success, every error return, the reply /// timeout, and handler cancellation (the client hanging up mid-await drops @@ -64,14 +43,26 @@ impl Drop for InFlightWriteGuard<'_> { /// shard-0 global budget, incrementing both counters only when both pass. The /// session cap is checked first so a session that saturates itself reads as /// its own 429 rather than as server-wide pressure. +/// +/// Both caps come from `[http_admission]`. `max_global` bounds both starvation +/// terms every admitted write imposes on shard 0 - budget x `max_request_size` +/// worst-case buffered bytes, budget x per-request decode/encode/HS256 CPU on +/// the core that also pumps consensus. `max_per_session` bounds one credential's +/// slice of that budget, so a session that outruns its own commits reads its own +/// 429 before it can spill onto the shared budget. `?ack=none` produces are +/// admitted through the same caps: they install no reply slot, but still park +/// inside dispatch while pinning their buffered body, so leaving them uncapped +/// would bypass both terms. pub(in crate::http) fn admit_partition_write<'a>( session_in_flight: &'a Cell<u32>, global_in_flight: &'a Cell<u32>, + max_per_session: u32, + max_global: u32, ) -> Result<InFlightWriteGuard<'a>, PartitionWriteError> { - if session_in_flight.get() >= MAX_IN_FLIGHT_WRITES_PER_SESSION { + if session_in_flight.get() >= max_per_session { return Err(PartitionWriteError::TooManyInFlight); } - if global_in_flight.get() >= MAX_IN_FLIGHT_WRITES_GLOBAL { + if global_in_flight.get() >= max_global { return Err(PartitionWriteError::ServerBusy); } session_in_flight.set(session_in_flight.get() + 1); @@ -86,11 +77,23 @@ pub(in crate::http) fn admit_partition_write<'a>( mod tests { use super::*; + const SESSION_CAP: u32 = 32; + const GLOBAL_CAP: u32 = 128; + + /// Admit at the fixture caps, so each test reads the counting behavior + /// without restating the caps on every call. + fn admit<'a>( + session_in_flight: &'a Cell<u32>, + global_in_flight: &'a Cell<u32>, + ) -> Result<InFlightWriteGuard<'a>, PartitionWriteError> { + admit_partition_write(session_in_flight, global_in_flight, SESSION_CAP, GLOBAL_CAP) + } + #[test] fn in_flight_write_guard_decrements_both_counters_on_drop() { let session = Cell::new(0); let global = Cell::new(0); - let guard = admit_partition_write(&session, &global).expect("below both caps"); + let guard = admit(&session, &global).expect("below both caps"); assert_eq!(session.get(), 1); assert_eq!(global.get(), 1); drop(guard); @@ -100,27 +103,27 @@ mod tests { #[test] fn admission_at_session_cap_rejects_with_too_many_in_flight() { - let session = Cell::new(MAX_IN_FLIGHT_WRITES_PER_SESSION); + let session = Cell::new(SESSION_CAP); let global = Cell::new(0); assert!(matches!( - admit_partition_write(&session, &global), + admit(&session, &global), Err(PartitionWriteError::TooManyInFlight) )); // A refusal must not leak a partial increment on either counter. - assert_eq!(session.get(), MAX_IN_FLIGHT_WRITES_PER_SESSION); + assert_eq!(session.get(), SESSION_CAP); assert_eq!(global.get(), 0); } #[test] fn admission_at_global_budget_rejects_with_server_busy() { let session = Cell::new(0); - let global = Cell::new(MAX_IN_FLIGHT_WRITES_GLOBAL); + let global = Cell::new(GLOBAL_CAP); assert!(matches!( - admit_partition_write(&session, &global), + admit(&session, &global), Err(PartitionWriteError::ServerBusy) )); assert_eq!(session.get(), 0); - assert_eq!(global.get(), MAX_IN_FLIGHT_WRITES_GLOBAL); + assert_eq!(global.get(), GLOBAL_CAP); } #[test] @@ -128,21 +131,21 @@ mod tests { let session = Cell::new(0); let global = Cell::new(0); let mut guards = Vec::new(); - for _ in 0..MAX_IN_FLIGHT_WRITES_PER_SESSION { - guards.push(admit_partition_write(&session, &global).expect("below both caps")); + for _ in 0..SESSION_CAP { + guards.push(admit(&session, &global).expect("below both caps")); } assert!(matches!( - admit_partition_write(&session, &global), + admit(&session, &global), Err(PartitionWriteError::TooManyInFlight) )); let released = 3; - guards.truncate((MAX_IN_FLIGHT_WRITES_PER_SESSION - released) as usize); - assert_eq!(session.get(), MAX_IN_FLIGHT_WRITES_PER_SESSION - released); + guards.truncate((SESSION_CAP - released) as usize); + assert_eq!(session.get(), SESSION_CAP - released); for _ in 0..released { - guards.push(admit_partition_write(&session, &global).expect("released slots")); + guards.push(admit(&session, &global).expect("released slots")); } assert!(matches!( - admit_partition_write(&session, &global), + admit(&session, &global), Err(PartitionWriteError::TooManyInFlight) )); drop(guards); @@ -153,13 +156,12 @@ mod tests { #[test] fn global_budget_spans_sessions_and_reopens_after_release() { let global = Cell::new(0); - let session_count = - MAX_IN_FLIGHT_WRITES_GLOBAL.div_ceil(MAX_IN_FLIGHT_WRITES_PER_SESSION) as usize; + let session_count = GLOBAL_CAP.div_ceil(SESSION_CAP) as usize; let sessions: Vec<Cell<u32>> = (0..session_count).map(|_| Cell::new(0)).collect(); let mut guards = Vec::new(); 'fill: for session in &sessions { - for _ in 0..MAX_IN_FLIGHT_WRITES_PER_SESSION { - match admit_partition_write(session, &global) { + for _ in 0..SESSION_CAP { + match admit(session, &global) { Ok(guard) => guards.push(guard), Err(PartitionWriteError::ServerBusy) => break 'fill, Err(other) => { @@ -168,17 +170,17 @@ mod tests { } } } - assert_eq!(global.get(), MAX_IN_FLIGHT_WRITES_GLOBAL); + assert_eq!(global.get(), GLOBAL_CAP); // A fresh session is refused on the shared budget, not its own cap. let fresh = Cell::new(0); assert!(matches!( - admit_partition_write(&fresh, &global), + admit(&fresh, &global), Err(PartitionWriteError::ServerBusy) )); drop(guards.pop()); - let readmitted = admit_partition_write(&fresh, &global).expect("budget slot released"); + let readmitted = admit(&fresh, &global).expect("budget slot released"); assert_eq!(fresh.get(), 1); - assert_eq!(global.get(), MAX_IN_FLIGHT_WRITES_GLOBAL); + assert_eq!(global.get(), GLOBAL_CAP); drop(readmitted); } } diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs index 63fbe935a..59fa369ec 100644 --- a/core/server-ng/src/http/error.rs +++ b/core/server-ng/src/http/error.rs @@ -221,14 +221,14 @@ pub(in crate::http) enum PartitionWriteError { /// The in-process reply slot could not be installed. Transient server /// condition -> the shared 503, retryable. Unavailable, - /// This session is already at [`MAX_IN_FLIGHT_WRITES_PER_SESSION`] + /// This session is already at its `http_admission.max_in_flight_writes_per_session` /// awaited writes. 429: the caller's own concurrency is the problem, so /// it must drain its outstanding writes before submitting more. TooManyInFlight, - /// Shard 0 is already at [`MAX_IN_FLIGHT_WRITES_GLOBAL`] awaited writes - /// across all sessions. 503 with its own code (distinct from the shared - /// consensus-unavailable body) so an operator can tell admission shedding - /// from a consensus outage. + /// Shard 0 is already at its `http_admission.max_in_flight_writes` awaited + /// writes across all sessions. 503 with its own code (distinct from the + /// shared consensus-unavailable body) so an operator can tell admission + /// shedding from a consensus outage. ServerBusy, /// No committed reply within [`PARTITION_WRITE_REPLY_TIMEOUT`], or the /// session's reply target was torn down mid-wait. 504: the commit may @@ -323,10 +323,10 @@ fn service_unavailable() -> Response { ) } -/// 429 for a session at [`MAX_IN_FLIGHT_WRITES_PER_SESSION`] awaited partition -/// writes. Shaped like every other HTTP error (`ErrorResponse`) so clients -/// parse one error schema; the remedy is the caller's own: let outstanding -/// writes finish, then retry. +/// 429 for a session at its `http_admission.max_in_flight_writes_per_session` +/// awaited partition writes. Shaped like every other HTTP error +/// (`ErrorResponse`) so clients parse one error schema; the remedy is the +/// caller's own: let outstanding writes finish, then retry. fn too_many_in_flight_response() -> Response { with_retry_after(error_response( StatusCode::TOO_MANY_REQUESTS, @@ -335,8 +335,8 @@ fn too_many_in_flight_response() -> Response { )) } -/// 503 for shard 0 at [`MAX_IN_FLIGHT_WRITES_GLOBAL`] awaited partition writes -/// across all sessions. A distinct `server_busy` code (unlike the shared +/// 503 for shard 0 at its `http_admission.max_in_flight_writes` awaited partition +/// writes across all sessions. A distinct `server_busy` code (unlike the shared /// consensus-unavailable 503) so admission shedding is tellable from a /// consensus outage; retry with backoff. fn server_busy_response() -> Response { diff --git a/core/server-ng/src/http/session.rs b/core/server-ng/src/http/session.rs index a7f697993..14ac9347b 100644 --- a/core/server-ng/src/http/session.rs +++ b/core/server-ng/src/http/session.rs @@ -88,7 +88,7 @@ pub(in crate::http) struct HttpSession { /// token. pub(in crate::http) registry_token: Cell<Option<InstanceToken>>, /// Awaited partition writes currently in flight on this session, gated by - /// [`MAX_IN_FLIGHT_WRITES_PER_SESSION`]. Only [`InFlightWriteGuard`] + /// `HttpInner::max_in_flight_writes_per_session`. Only [`InFlightWriteGuard`] /// touches it, so every admission is paired with exactly one release. pub(in crate::http) in_flight_writes: Cell<u32>, } diff --git a/core/server-ng/src/http/state.rs b/core/server-ng/src/http/state.rs index 777518946..f3f5bfaaa 100644 --- a/core/server-ng/src/http/state.rs +++ b/core/server-ng/src/http/state.rs @@ -78,9 +78,17 @@ pub(in crate::http) struct HttpInner { /// requests for one credential from each running its own `Register`. pub(in crate::http) registrations: RegistrationBarrier, pub(in crate::http) roster: ClusterRoster, + /// Shard-0-global cap on concurrently awaited partition writes, from + /// `[http_admission] max_in_flight_writes`. Passed to `admit_partition_write` + /// alongside the per-session cap. + pub(in crate::http) max_in_flight_writes: u32, + /// Per-session cap on concurrently awaited partition writes, from + /// `[http_admission] max_in_flight_writes_per_session`. Uniform across + /// sessions, so it lives here rather than on each `HttpSession`. + pub(in crate::http) max_in_flight_writes_per_session: u32, /// Awaited partition writes currently in flight across all sessions, gated - /// by [`MAX_IN_FLIGHT_WRITES_GLOBAL`]. Only [`InFlightWriteGuard`] touches - /// it, so every admission is paired with exactly one release. + /// by `max_in_flight_writes`. Only [`InFlightWriteGuard`] touches it, so + /// every admission is paired with exactly one release. pub(in crate::http) in_flight_writes: Cell<u32>, } diff --git a/core/server-ng/src/http/submit.rs b/core/server-ng/src/http/submit.rs index f343f546f..020d1a29c 100644 --- a/core/server-ng/src/http/submit.rs +++ b/core/server-ng/src/http/submit.rs @@ -308,7 +308,12 @@ pub(in crate::http) async fn partition_write_replicated( // actually unbounded - the slot install, the dispatch, and the parked // reply await that pins this request's buffers for up to the reply // timeout. Held across every exit below; released by `Drop`. - let _in_flight = admit_partition_write(&session.in_flight_writes, &state.in_flight_writes)?; + let _in_flight = admit_partition_write( + &session.in_flight_writes, + &state.in_flight_writes, + state.max_in_flight_writes_per_session, + state.max_in_flight_writes, + )?; ensure_in_process_reply_target(state, session); let request_id = session.next_data_request_id(); let message = build_request_message( @@ -371,7 +376,12 @@ pub(in crate::http) async fn produce_unacked( session: &HttpSession, body: &[u8], ) -> Result<(), PartitionWriteError> { - let _in_flight = admit_partition_write(&session.in_flight_writes, &state.in_flight_writes)?; + let _in_flight = admit_partition_write( + &session.in_flight_writes, + &state.in_flight_writes, + state.max_in_flight_writes_per_session, + state.max_in_flight_writes, + )?; let request_id = session.next_data_request_id(); let message = build_request_message( Operation::SendMessages,
