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 d8d19fef4578d8df3c4b96be3ad71dde62dd90a3 Author: Hubert Gruszecki <[email protected]> AuthorDate: Fri Jul 24 14:16:20 2026 +0200 feat(server-ng): add partition.prepare_queue_depth for pipeline sizing Partition-plane consensus pipelines were constructed with the hardcoded 32/64 prepare/request depths while the metadata plane already sizes its pipeline from metadata.prepare_queue_depth. The partition knob must be separate from metadata's: partition pipelines exist per partition and their request queue pins full inbound produce batches, so depth multiplies memory by partition count, while metadata is a single small-op instance. Add an ng-only [partition] section with prepare_queue_depth (default 32, behavior unchanged) flowing to both construction paths as with_capacities(depth, depth * 2). The ceiling is 256, far below metadata's 4096, as a typo guard against per-partition memory blowup. The embedded default, the configs-crate const and consensus::PIPELINE_PREPARE_QUEUE_MAX are locked together by a static assert plus a lockstep test. --- core/configs/src/lib.rs | 4 +- core/configs/src/server_ng_config/defaults.rs | 13 +++ core/configs/src/server_ng_config/mod.rs | 1 + core/configs/src/server_ng_config/partition.rs | 117 ++++++++++++++++++++++++ core/configs/src/server_ng_config/server_ng.rs | 2 + core/configs/src/server_ng_config/validators.rs | 3 + core/server-ng/config.toml | 10 ++ core/server-ng/src/bootstrap.rs | 24 ++++- core/server-ng/src/partition_helpers.rs | 5 +- 9 files changed, 175 insertions(+), 4 deletions(-) diff --git a/core/configs/src/lib.rs b/core/configs/src/lib.rs index aca084501..8af9f174d 100644 --- a/core/configs/src/lib.rs +++ b/core/configs/src/lib.rs @@ -31,6 +31,6 @@ pub use server_config::{ }; pub use server_ng_config::{ 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, + metadata as ng_metadata, partition as ng_partition, 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 d464f2190..a48e9f8fd 100644 --- a/core/configs/src/server_ng_config/defaults.rs +++ b/core/configs/src/server_ng_config/defaults.rs @@ -33,6 +33,7 @@ use super::cluster::{ use super::http_admission::HttpAdmissionConfig; use super::message_bus::MessageBusConfig; use super::metadata::MetadataConfig; +use super::partition::PartitionConfig; use super::quic::{QuicCertificateConfig, QuicConfig, QuicSocketConfig}; use super::server_ng::NgSystemConfig; use super::server_ng::{ExtraConfig, ServerNgConfig}; @@ -68,6 +69,7 @@ impl Default for ServerNgConfig { telemetry: TelemetryConfig::default(), cluster: ClusterConfig::default(), metadata: MetadataConfig::default(), + partition: PartitionConfig::default(), message_bus: MessageBusConfig::default(), } } @@ -142,6 +144,17 @@ impl Default for MetadataConfig { } } +impl Default for PartitionConfig { + fn default() -> PartitionConfig { + // Read from the embedded TOML so the Default impl and the on-disk + // schema cannot drift (same pattern as MetadataConfig above). + let partition = &SERVER_NG_CONFIG.partition; + PartitionConfig { + prepare_queue_depth: partition.prepare_queue_depth as usize, + } + } +} + impl Default for HttpAdmissionConfig { fn default() -> HttpAdmissionConfig { // Read from the embedded TOML so the Default impl and the on-disk diff --git a/core/configs/src/server_ng_config/mod.rs b/core/configs/src/server_ng_config/mod.rs index 88980c252..dc2a736b2 100644 --- a/core/configs/src/server_ng_config/mod.rs +++ b/core/configs/src/server_ng_config/mod.rs @@ -33,6 +33,7 @@ pub mod displays; pub mod http_admission; pub mod message_bus; pub mod metadata; +pub mod partition; pub mod quic; pub mod server_ng; pub mod sharding; diff --git a/core/configs/src/server_ng_config/partition.rs b/core/configs/src/server_ng_config/partition.rs new file mode 100644 index 000000000..8bf4f7cb6 --- /dev/null +++ b/core/configs/src/server_ng_config/partition.rs @@ -0,0 +1,117 @@ +// 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 the per-partition consensus plane. +//! +//! One capacity knob previously hardcoded in the runtime crates: +//! +//! - `prepare_queue_depth` -> `consensus::PIPELINE_PREPARE_QUEUE_MAX` +//! (the pipeline's in-flight prepare bound; submits beyond it bounce +//! with the transient prepare-queue-full path the SDK retries) +//! +//! Distinct from `[metadata]` (a single, shard-0-global VSR plane) because +//! partition pipelines exist PER PARTITION. The default mirrors the runtime +//! constant so a default deployment is byte-identical; the ceiling is far +//! below metadata's because the request queue (`depth * 2` slots) pins full +//! inbound produce batches, so pinned memory scales with the partition count +//! (see [`MAX_PARTITION_PREPARE_QUEUE_DEPTH`]). +//! +//! The default is a duplicated literal rather than an import so +//! `core/configs` does not grow a build-time edge onto `core/consensus` +//! (mirroring [`super::metadata`]). `core/server-ng`'s bootstrap pins the +//! literal against the runtime constant with a static assert. + +use super::COMPONENT_NG; +use crate::ConfigurationError; +use configs::ConfigEnv; +use iggy_common::Validatable; +use serde::{Deserialize, Serialize}; + +/// Mirrors `consensus::PIPELINE_PREPARE_QUEUE_MAX`. +pub const DEFAULT_PARTITION_PREPARE_QUEUE_DEPTH: usize = 32; + +/// Upper bound on `prepare_queue_depth`. Unlike the single metadata pipeline, +/// a pipeline exists per partition, and each queued request pins a full +/// inbound produce batch (a 4 KiB floor up to megabytes). Worst-case pinned +/// memory therefore scales as `depth * 2 * partition_count * batch_size`, so +/// this ceiling sits far below metadata's 4096: it is a typo guard, not a +/// sizing endorsement. +pub const MAX_PARTITION_PREPARE_QUEUE_DEPTH: usize = 256; + +/// Capacity tunables for the per-partition consensus plane. +#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] +pub struct PartitionConfig { + /// Depth of a partition's prepare queue: how many uncommitted produce / + /// consumer-offset ops may be in flight at once for that partition. + /// Submits beyond it are rejected with the transient prepare-queue-full + /// path the SDK retries. Applies to every partition; raising it multiplies + /// pinned request-buffer memory by the partition count. + pub prepare_queue_depth: usize, +} + +impl Validatable<ConfigurationError> for PartitionConfig { + fn validate(&self) -> Result<(), ConfigurationError> { + if self.prepare_queue_depth == 0 { + eprintln!("{COMPONENT_NG} partition.prepare_queue_depth must be > 0"); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.prepare_queue_depth > MAX_PARTITION_PREPARE_QUEUE_DEPTH { + eprintln!( + "{COMPONENT_NG} partition.prepare_queue_depth ({}) exceeds the maximum ({MAX_PARTITION_PREPARE_QUEUE_DEPTH})", + self.prepare_queue_depth + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_impl_validates() { + // `Default` reads the shipped config.toml; the pristine deployment + // must validate. + assert!(PartitionConfig::default().validate().is_ok()); + } + + #[test] + fn rejects_zero() { + let config = PartitionConfig { + prepare_queue_depth: 0, + }; + assert!(config.validate().is_err()); + } + + #[test] + fn rejects_above_ceiling() { + let config = PartitionConfig { + prepare_queue_depth: MAX_PARTITION_PREPARE_QUEUE_DEPTH + 1, + }; + assert!(config.validate().is_err()); + } + + #[test] + fn accepts_at_ceiling() { + let config = PartitionConfig { + prepare_queue_depth: MAX_PARTITION_PREPARE_QUEUE_DEPTH, + }; + assert!(config.validate().is_ok()); + } +} diff --git a/core/configs/src/server_ng_config/server_ng.rs b/core/configs/src/server_ng_config/server_ng.rs index daa7b5036..77be95249 100644 --- a/core/configs/src/server_ng_config/server_ng.rs +++ b/core/configs/src/server_ng_config/server_ng.rs @@ -20,6 +20,7 @@ use super::cluster::ClusterConfig; use super::http_admission::HttpAdmissionConfig; use super::message_bus::MessageBusConfig; use super::metadata::MetadataConfig; +use super::partition::PartitionConfig; use super::quic::QuicConfig; use super::tcp::TcpConfig; use super::websocket::WebSocketConfig; @@ -80,6 +81,7 @@ pub struct ServerNgConfig { pub telemetry: TelemetryConfig, pub cluster: ClusterConfig, pub metadata: MetadataConfig, + pub partition: PartitionConfig, pub message_bus: MessageBusConfig, } diff --git a/core/configs/src/server_ng_config/validators.rs b/core/configs/src/server_ng_config/validators.rs index 9063bf747..8ff975ae7 100644 --- a/core/configs/src/server_ng_config/validators.rs +++ b/core/configs/src/server_ng_config/validators.rs @@ -83,6 +83,9 @@ impl Validatable<ConfigurationError> for ServerNgConfig { self.metadata.validate().error(|e: &ConfigurationError| { format!("{COMPONENT_NG} (error: {e}) - failed to validate metadata config") })?; + self.partition.validate().error(|e: &ConfigurationError| { + format!("{COMPONENT_NG} (error: {e}) - failed to validate partition config") + })?; self.http_admission .validate() .error(|e: &ConfigurationError| { diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index 7dfb2b8b8..fe342c55b 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -749,6 +749,16 @@ prepare_queue_depth = 32 # rewrites per checkpoint. journal_slots = 1024 +# Per-partition consensus plane tunables. Unlike [metadata] (one shard-0 +# plane), a pipeline exists per partition, so raising this multiplies pinned +# request-buffer memory by the partition count. Keep it modest. +[partition] +# Depth of a partition's prepare queue: how many uncommitted produce / +# consumer-offset ops may be in flight at once for that partition. Submits +# beyond it are rejected with the transient prepare-queue-full path and +# retried by the SDK. Must be > 0 and <= 256. +prepare_queue_depth = 32 + # Message bus configuration. # Tunables for the inter-shard / inter-replica internal bus that ships # consensus traffic between replicas and SDK-client traffic between diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index 58021ac40..d86235e4c 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -1693,6 +1693,10 @@ const _: () = assert!( configs::ng_metadata::DEFAULT_METADATA_JOURNAL_SLOTS == journal::prepare_journal::DEFAULT_SLOT_COUNT ); +const _: () = assert!( + configs::ng_partition::DEFAULT_PARTITION_PREPARE_QUEUE_DEPTH + == consensus::PIPELINE_PREPARE_QUEUE_MAX +); /// Convert a consensus-timer interval to whole ticks, floored at one tick so a /// sub-tick value still fires and saturated on overflow. fn duration_to_ticks(interval: Duration) -> u64 { @@ -1893,13 +1897,16 @@ async fn load_partition( let stream_id = namespace.stream_id(); let topic_id = namespace.topic_id(); let partition_id = namespace.partition_id(); + // Request queue holds 2x the prepare depth (buffered requests drain as + // prepares commit); depth is the per-partition `[partition]` knob. + let prepare_queue_depth = config.partition.prepare_queue_depth; let consensus = VsrConsensus::new( cluster_id, self_replica_id, replica_count, namespace.inner(), bus, - LocalPipeline::new(), + LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2), ); consensus.set_normal_heartbeat_ticks(cluster_heartbeat_ticks(config)); consensus.set_commit_message_ticks(commit_broadcast_ticks(config)); @@ -3214,6 +3221,21 @@ mod tests { ); } + #[test] + fn default_partition_prepare_queue_depth_matches_consensus_constant() { + // The config default lives in core/server-ng/config.toml and flows + // through PartitionConfig::default(); keep the embedded value in + // lockstep with the pipeline depth LocalPipeline::new() (the simulator + // and tests) runs on, so a default deployment is byte-identical. + let config_default = configs::ng_partition::PartitionConfig::default().prepare_queue_depth; + assert_eq!( + config_default, + consensus::PIPELINE_PREPARE_QUEUE_MAX, + "[partition] prepare_queue_depth default drifted from \ + consensus::PIPELINE_PREPARE_QUEUE_MAX" + ); + } + #[test] fn shutdown_on_drop_armed_flips_flag() { let flag = Arc::new(AtomicBool::new(false)); diff --git a/core/server-ng/src/partition_helpers.rs b/core/server-ng/src/partition_helpers.rs index aad42b9d4..8bccd5521 100644 --- a/core/server-ng/src/partition_helpers.rs +++ b/core/server-ng/src/partition_helpers.rs @@ -487,13 +487,16 @@ pub async fn build_partition_fresh( source })?; + // Request queue holds 2x the prepare depth (buffered requests drain as + // prepares commit); depth is the per-partition `[partition]` knob. + let prepare_queue_depth = config.partition.prepare_queue_depth; let consensus = VsrConsensus::new( cluster_id, self_replica_id, replica_count, namespace.inner(), bus, - LocalPipeline::new(), + LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2), ); consensus.set_normal_heartbeat_ticks(crate::bootstrap::cluster_heartbeat_ticks(config)); consensus.set_commit_message_ticks(crate::bootstrap::commit_broadcast_ticks(config));
