This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 6ae792a37 refactor(configs): fork ShardingConfig for legacy and
server-ng (#3633)
6ae792a37 is described below
commit 6ae792a37d2d470fadba7f18b780442f582bd258
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Wed Jul 8 15:09:06 2026 +0200
refactor(configs): fork ShardingConfig for legacy and server-ng (#3633)
---
core/configs/src/lib.rs | 3 +-
core/configs/src/server_config/defaults.rs | 10 +-
core/configs/src/server_config/displays.rs | 3 +-
core/configs/src/server_config/sharding.rs | 155 +----------
core/configs/src/server_config/system.rs | 15 +-
core/configs/src/server_config/validators.rs | 288 ++++++--------------
core/configs/src/server_ng_config/defaults.rs | 4 +-
core/configs/src/server_ng_config/mod.rs | 1 +
core/configs/src/server_ng_config/server_ng.rs | 7 +-
core/configs/src/server_ng_config/sharding.rs | 357 +++++++++++++++++++++++++
core/configs_derive/src/config_env.rs | 11 +-
core/server-ng/config.toml | 14 +-
core/server-ng/src/bootstrap.rs | 8 +-
core/server-ng/src/dispatch.rs | 16 +-
core/server-ng/src/http.rs | 4 +-
core/server-ng/src/http/state.rs | 4 +-
core/server-ng/src/partition_reconciler.rs | 8 +-
core/server-ng/src/snapshot.rs | 14 +-
core/server/config.toml | 2 +-
19 files changed, 520 insertions(+), 404 deletions(-)
diff --git a/core/configs/src/lib.rs b/core/configs/src/lib.rs
index f3bd7685b..4d454f56f 100644
--- a/core/configs/src/lib.rs
+++ b/core/configs/src/lib.rs
@@ -30,5 +30,6 @@ pub use server_config::{
tcp, validators, websocket,
};
pub use server_ng_config::{
- COMPONENT_NG, message_bus, quic as ng_quic, server_ng, tcp as ng_tcp,
websocket as ng_websocket,
+ COMPONENT_NG, message_bus, 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_config/defaults.rs
b/core/configs/src/server_config/defaults.rs
index c29baeac4..155239cd3 100644
--- a/core/configs/src/server_config/defaults.rs
+++ b/core/configs/src/server_config/defaults.rs
@@ -26,7 +26,6 @@ use super::server::{
PersonalAccessTokenConfig, ServerConfig, TelemetryConfig,
TelemetryLogsConfig,
TelemetryTracesConfig,
};
-use super::sharding::ShardingConfig;
use super::system::{
BackupConfig, CompatibilityConfig, CompressionConfig, EncryptionConfig,
LoggingConfig,
MessageDeduplicationConfig, PartitionConfig, RecoveryConfig,
RuntimeConfig, SegmentConfig,
@@ -35,6 +34,7 @@ use super::system::{
use super::tcp::TcpSocketConfig;
use super::tcp::{TcpConfig, TcpTlsConfig};
use super::websocket::{WebSocketConfig, WebSocketTlsConfig};
+use configs::ConfigEnvMappings;
use iggy_common::IggyByteSize;
use iggy_common::IggyDuration;
use std::sync::Arc;
@@ -321,9 +321,9 @@ impl Default for PersonalAccessTokenCleanerConfig {
}
}
-impl Default for SystemConfig {
- fn default() -> SystemConfig {
- SystemConfig {
+impl<S: ConfigEnvMappings + Default> Default for SystemConfig<S> {
+ fn default() -> Self {
+ Self {
path: SERVER_CONFIG.system.path.parse().unwrap(),
backup: BackupConfig::default(),
runtime: RuntimeConfig::default(),
@@ -338,7 +338,7 @@ impl Default for SystemConfig {
message_deduplication: MessageDeduplicationConfig::default(),
recovery: RecoveryConfig::default(),
memory_pool: MemoryPoolConfig::default(),
- sharding: ShardingConfig::default(),
+ sharding: S::default(),
}
}
}
diff --git a/core/configs/src/server_config/displays.rs
b/core/configs/src/server_config/displays.rs
index 24b34bf81..f7693965d 100644
--- a/core/configs/src/server_config/displays.rs
+++ b/core/configs/src/server_config/displays.rs
@@ -30,6 +30,7 @@ use super::{
},
tcp::{TcpConfig, TcpSocketConfig, TcpTlsConfig},
};
+use configs::ConfigEnvMappings;
use std::fmt::{Display, Formatter};
impl Display for HttpConfig {
@@ -345,7 +346,7 @@ impl Display for TelemetryTracesConfig {
}
}
-impl Display for SystemConfig {
+impl<S: ConfigEnvMappings> Display for SystemConfig<S> {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
f,
diff --git a/core/configs/src/server_config/sharding.rs
b/core/configs/src/server_config/sharding.rs
index 2f23cfdb7..87a411e1f 100644
--- a/core/configs/src/server_config/sharding.rs
+++ b/core/configs/src/server_config/sharding.rs
@@ -15,11 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_common::IggyDuration;
use serde::{Deserialize, Serialize};
-use serde_with::{DisplayFromStr, serde_as};
-use std::time::Duration;
+use super::defaults::SERVER_CONFIG;
use configs::ConfigEnv;
// `CpuAllocation`/`NumaConfig` are pure config types and live in their own
@@ -28,96 +26,9 @@ use configs::ConfigEnv;
// `configs::sharding::*` path stable for existing callers.
pub use cpu_allocation::{CpuAllocation, NumaConfig};
-/// Default capacity of the per-shard inter-shard inbox channel. Sized
-/// comfortably above the consensus working set, which is roughly
-/// `PIPELINE_PREPARE_QUEUE_MAX (= 32) * replica_count * directions`
-/// frames in flight per shard, without allowing a runaway producer to
-/// eat unbounded memory. Tunable via `[system.sharding] inbox_capacity`
-/// in TOML.
-///
-/// The capacity must also absorb the worst-case cross-shard client
-/// Reply burst. Unlike consensus frames, client Replies have no VSR
-/// retransmit path: a Reply lost on full inbox is gone and the client
-/// times out. A reasonable lower bound is
-/// `max_inflight_client_requests / num_shards` (assuming requests are
-/// distributed evenly across owning shards) plus the consensus
-/// headroom above.
-///
-/// Consensus frames and client-reply forwards share this one channel,
-/// so the two headrooms are not independent: a consensus burst or
-/// retransmit storm can fill the inbox with consensus frames exactly
-/// when a client Reply needs the space. A single `inbox_capacity` knob
-/// cannot isolate the two frame classes - size it for the sum of both
-/// worst cases occurring together. Watch the drop-site `tracing` logs
-/// (and, once a per-shard exporter lands, the `frame_drops_total`
-/// `{variant="forward_client_send"}` counter) to detect when the bound
-/// is too low in production.
-pub const DEFAULT_INBOX_CAPACITY: usize = 1024;
-
-/// Maximum permitted per-shard inbox depth. The channel is allocated
-/// up-front per shard, so a runaway value here OOMs the process at boot.
-/// `1 << 20` (~1M frames) is several orders of magnitude above any
-/// realistic backpressure target and still fits comfortably in process
-/// address space.
-pub const INBOX_CAPACITY_MAX: usize = 1 << 20;
-
-/// Default bus shutdown drain timeout. Sized larger than typical TCP RTT
-/// times in-flight write-batch so writers receive their full last
-/// `write_vectored_all` budget before the connection registry kicks in.
-pub const DEFAULT_SHUTDOWN_DRAIN_TIMEOUT: Duration = Duration::from_secs(10);
-
-/// Default watchdog poll cadence for the cross-thread shutdown flag.
-/// 50ms keeps Ctrl-C latency operator-visible without measurable wakeup
-/// overhead.
-pub const DEFAULT_SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(50);
-
-/// Hard upper bound on `shutdown_drain_timeout`. A drain that never
-/// completes wedges process exit; capping at 10 minutes guarantees the
-/// watchdog eventually force-tears the bus even with a pathological
-/// config typo.
-pub const SHUTDOWN_DRAIN_TIMEOUT_MAX: Duration = Duration::from_secs(600);
-
-/// Hard upper bound on `shutdown_poll_interval`. A pollerinterval longer
-/// than the drain timeout makes the flag effectively unobservable; cap
-/// at 5s so Ctrl-C latency stays bounded regardless of config.
-pub const SHUTDOWN_POLL_INTERVAL_MAX: Duration = Duration::from_secs(5);
-
-/// Default safety-tick cadence for the partition reconciliation loop.
-/// The reconciler also wakes on every `LifecycleFrame::MetadataCommitTick`
-/// broadcast by shard 0; this fallback covers dropped wake-ups (the wake
-/// channel is intentionally capacity-1) and the initial post-bootstrap
-/// convergence window before shard 0's first tick. One second is
-/// invisible to operators yet keeps idle clusters from burning CPU
-/// re-reading the same target snapshot.
-pub const DEFAULT_RECONCILE_PERIODIC_INTERVAL: Duration =
Duration::from_secs(1);
-
-/// Hard upper bound on `reconcile_periodic_interval`. A tick longer
-/// than ~30s makes post-failure recovery latency operator-visible; the
-/// cap reins in pathological typos without disturbing reasonable
-/// production values.
-pub const RECONCILE_PERIODIC_INTERVAL_MAX: Duration = Duration::from_secs(30);
-
-const fn default_inbox_capacity() -> usize {
- DEFAULT_INBOX_CAPACITY
-}
-
-const fn default_pin_cores() -> bool {
- true
-}
-
-fn default_shutdown_drain_timeout() -> IggyDuration {
- IggyDuration::new(DEFAULT_SHUTDOWN_DRAIN_TIMEOUT)
-}
-
-fn default_shutdown_poll_interval() -> IggyDuration {
- IggyDuration::new(DEFAULT_SHUTDOWN_POLL_INTERVAL)
-}
-
-fn default_reconcile_periodic_interval() -> IggyDuration {
- IggyDuration::new(DEFAULT_RECONCILE_PERIODIC_INTERVAL)
-}
-
-#[serde_as]
+/// Sharding config for the legacy `core/server`. That server consumes only
+/// `cpu_allocation` and `pin_cores`; the bus / shutdown / reconcile knobs are
+/// server-ng concepts and live in [`crate::server_ng_config::sharding`].
#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct ShardingConfig {
#[serde(default)]
@@ -133,70 +44,14 @@ pub struct ShardingConfig {
/// threads freely within the allowed set. With a NUMA-aware allocation,
/// `false` drops both the CPU and memory-node bindings (and logs a
/// warning, since NUMA placement without pinning is meaningless).
- #[serde(default = "default_pin_cores")]
pub pin_cores: bool,
- /// Per-shard inter-shard inbox channel capacity. Bounded by design.
- /// Drops on full inbox of consensus frames are recovered by VSR
- /// retransmit. Drops of cross-shard client Reply frames are terminal:
- /// the client never receives the reply (no in-protocol retransmit).
- /// Both frame classes share this one channel, so a consensus burst
- /// can starve client-reply forwards: size against the worst-case sum
- /// of consensus working set + peak client-reply fan-out per shard
- /// occurring together; see `DEFAULT_INBOX_CAPACITY` for the
- /// rationale. Used by `core/server-ng`; the legacy server uses its
- /// own hard-coded inbox sizing.
- ///
- // TODO(hubcio): split into two priority lanes - one bounded queue for
- // consensus frames (drops recovered by VSR retransmit) and one for
- // client `Reply` frames (drops terminal, must be sized for worst-case
- // fan-out). Current single-channel design is the minimum-viable
- // wiring so `frame_drops_total{variant,reason}` surfaces under load
- // and yields real numbers to size the split against.
- #[serde(default = "default_inbox_capacity")]
- pub inbox_capacity: usize,
- /// Wall-clock budget for a single shard's bus drain on shutdown.
- /// Drives `IggyMessageBus::shutdown(..)` from the per-shard watchdog
- /// and the parallel-join survivor path. Sized larger than typical
- /// TCP RTT times in-flight write-batch so writers receive their full
- /// last `write_vectored_all` budget before the connection registry
- /// force-tears the bus. Slow-fsync hosts may need to extend this past
- /// the default; the cap is `SHUTDOWN_DRAIN_TIMEOUT_MAX` so a config
- /// typo cannot wedge process exit.
- #[serde(default = "default_shutdown_drain_timeout")]
- #[serde_as(as = "DisplayFromStr")]
- #[config_env(leaf)]
- pub shutdown_drain_timeout: IggyDuration,
- /// Poll cadence for the cross-thread shutdown flag and for the
- /// `await_metadata_bundle` / `broadcast_metadata_bundle` poll loops.
- /// Trades off Ctrl-C latency against idle wakeup cost; the default
- /// keeps shutdown observably prompt without measurable scheduler
- /// overhead. Capped at `SHUTDOWN_POLL_INTERVAL_MAX` so the flag
- /// remains effectively observable regardless of config.
- #[serde(default = "default_shutdown_poll_interval")]
- #[serde_as(as = "DisplayFromStr")]
- #[config_env(leaf)]
- pub shutdown_poll_interval: IggyDuration,
- /// Safety-tick cadence for the partition reconciliation loop; the
- /// reconciler also wakes immediately on every
- /// `LifecycleFrame::MetadataCommitTick` from shard 0. See
- /// [`DEFAULT_RECONCILE_PERIODIC_INTERVAL`] for the rationale; values
- /// above [`RECONCILE_PERIODIC_INTERVAL_MAX`] are rejected by the
- /// validator.
- #[serde(default = "default_reconcile_periodic_interval")]
- #[serde_as(as = "DisplayFromStr")]
- #[config_env(leaf)]
- pub reconcile_periodic_interval: IggyDuration,
}
impl Default for ShardingConfig {
fn default() -> Self {
Self {
cpu_allocation: CpuAllocation::default(),
- pin_cores: default_pin_cores(),
- inbox_capacity: DEFAULT_INBOX_CAPACITY,
- shutdown_drain_timeout: default_shutdown_drain_timeout(),
- shutdown_poll_interval: default_shutdown_poll_interval(),
- reconcile_periodic_interval: default_reconcile_periodic_interval(),
+ pin_cores: SERVER_CONFIG.system.sharding.pin_cores,
}
}
}
diff --git a/core/configs/src/server_config/system.rs
b/core/configs/src/server_config/system.rs
index b1d124c20..60437a94f 100644
--- a/core/configs/src/server_config/system.rs
+++ b/core/configs/src/server_config/system.rs
@@ -18,7 +18,7 @@
use super::cache_indexes::CacheIndexesConfig;
use super::server::MemoryPoolConfig;
use super::sharding::ShardingConfig;
-use configs::ConfigEnv;
+use configs::{ConfigEnv, ConfigEnvMappings};
use iggy_common::IggyByteSize;
use iggy_common::IggyError;
use iggy_common::IggyExpiry;
@@ -33,8 +33,13 @@ use server_common::log::LoggingSettings;
pub const INDEX_EXTENSION: &str = "index";
pub const LOG_EXTENSION: &str = "log";
+// Generic over the sharding config so the legacy server and `server-ng` each
+// bind their own `ShardingConfig` (different knob sets, different default
+// source) while sharing this whole struct and its path helpers. The default
+// type param keeps bare `SystemConfig` meaning the legacy variant, so existing
+// callers compile unchanged.
#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
-pub struct SystemConfig {
+pub struct SystemConfig<S: ConfigEnvMappings = ShardingConfig> {
pub path: String,
pub backup: BackupConfig,
pub state: StateConfig,
@@ -49,7 +54,7 @@ pub struct SystemConfig {
pub message_deduplication: MessageDeduplicationConfig,
pub recovery: RecoveryConfig,
pub memory_pool: MemoryPoolConfig,
- pub sharding: ShardingConfig,
+ pub sharding: S,
}
#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
@@ -188,7 +193,7 @@ pub struct StateConfig {
pub retry_delay: IggyDuration,
}
-impl SystemConfig {
+impl<S: ConfigEnvMappings> SystemConfig<S> {
pub fn get_system_path(&self) -> String {
self.path.to_string()
}
@@ -352,7 +357,7 @@ impl SystemConfig {
}
}
-impl SystemPaths for SystemConfig {
+impl<S: ConfigEnvMappings> SystemPaths for SystemConfig<S> {
fn get_system_path(&self) -> String {
SystemConfig::get_system_path(self)
}
diff --git a/core/configs/src/server_config/validators.rs
b/core/configs/src/server_config/validators.rs
index ca8410e28..d35763bc6 100644
--- a/core/configs/src/server_config/validators.rs
+++ b/core/configs/src/server_config/validators.rs
@@ -21,10 +21,7 @@ use super::server::{
DataMaintenanceConfig, MessageSaverConfig, MessagesMaintenanceConfig,
TelemetryConfig,
};
use super::server::{MemoryPoolConfig, PersonalAccessTokenConfig, ServerConfig};
-use super::sharding::{
- CpuAllocation, INBOX_CAPACITY_MAX, RECONCILE_PERIODIC_INTERVAL_MAX,
SHUTDOWN_DRAIN_TIMEOUT_MAX,
- SHUTDOWN_POLL_INTERVAL_MAX, ShardingConfig,
-};
+use super::sharding::{CpuAllocation, ShardingConfig};
use super::system::SegmentConfig;
use super::system::{CompressionConfig, LoggingConfig, PartitionConfig};
use crate::ConfigurationError;
@@ -379,139 +376,70 @@ impl Validatable<ConfigurationError> for
MemoryPoolConfig {
}
}
-impl Validatable<ConfigurationError> for ShardingConfig {
- fn validate(&self) -> Result<(), ConfigurationError> {
- if self.inbox_capacity == 0 {
- eprintln!(
- "Invalid sharding configuration: inbox_capacity must be > 0
(crossfire silently \
- rounds 0 to 1, masking config errors)"
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
- if self.inbox_capacity > INBOX_CAPACITY_MAX {
- eprintln!(
- "Invalid sharding configuration: inbox_capacity {} exceeds the
{} cap (each \
- shard preallocates a channel of this size; oversizing here
OOMs the process at \
- boot)",
- self.inbox_capacity, INBOX_CAPACITY_MAX
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
-
- let drain = self.shutdown_drain_timeout.get_duration();
- if drain.is_zero() {
- eprintln!(
- "Invalid sharding configuration: shutdown_drain_timeout must
be > 0 (a zero \
- budget force-tears the bus mid-WAL-fsync on every shutdown)"
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
- if drain > SHUTDOWN_DRAIN_TIMEOUT_MAX {
- eprintln!(
- "Invalid sharding configuration: shutdown_drain_timeout {:?}
exceeds the {:?} \
- cap (an unbounded drain wedges process exit on bus stall)",
- drain, SHUTDOWN_DRAIN_TIMEOUT_MAX
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
-
- let poll = self.shutdown_poll_interval.get_duration();
- if poll.is_zero() {
- eprintln!(
- "Invalid sharding configuration: shutdown_poll_interval must
be > 0 (a zero \
- cadence busy-loops every shard's watchdog and
metadata-handoff poller)"
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
- if poll > SHUTDOWN_POLL_INTERVAL_MAX {
- eprintln!(
- "Invalid sharding configuration: shutdown_poll_interval {:?}
exceeds the {:?} \
- cap (a coarse cadence stalls Ctrl-C handling and metadata
handoff abort)",
- poll, SHUTDOWN_POLL_INTERVAL_MAX
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
- if poll > drain {
- eprintln!(
- "Invalid sharding configuration: shutdown_poll_interval {:?}
must be <= \
- shutdown_drain_timeout {:?} (a poll cadence coarser than the
drain budget makes \
- the shutdown flag effectively unobservable)",
- poll, drain
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
-
- let reconcile = self.reconcile_periodic_interval.get_duration();
- if reconcile.is_zero() {
- eprintln!(
- "Invalid sharding configuration: reconcile_periodic_interval
resolves to zero. \
- Note that \"0\", \"none\", \"unlimited\", and \"disabled\"
all parse to zero. The \
- periodic reconcile tick is a safety net for dropped
commit-wakes and cannot be \
- turned off; set a positive duration (default \"1s\", max
{RECONCILE_PERIODIC_INTERVAL_MAX:?})."
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
- if reconcile > RECONCILE_PERIODIC_INTERVAL_MAX {
- eprintln!(
- "Invalid sharding configuration: reconcile_periodic_interval
{:?} exceeds the \
- {:?} cap (a long tick makes post-failure convergence latency
operator-visible)",
- reconcile, RECONCILE_PERIODIC_INTERVAL_MAX
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
+/// Validate a [`CpuAllocation`] against the machine's available parallelism
+/// and, when pinning, the process affinity mask. Shared by the legacy and
+/// server-ng sharding configs, which both carry these two knobs.
+pub(crate) fn validate_cpu_allocation(
+ cpu_allocation: &CpuAllocation,
+ pin_cores: bool,
+) -> Result<(), ConfigurationError> {
+ let available_cpus = available_parallelism()
+ .map_err(|_| {
+ eprintln!("Failed to detect available CPU cores");
+ ConfigurationError::InvalidConfigurationValue
+ })?
+ .get();
+
+ match cpu_allocation {
+ CpuAllocation::All => Ok(()),
+ CpuAllocation::Count(count) => {
+ if *count == 0 {
+ eprintln!("Invalid sharding configuration: cpu_allocation
count cannot be 0");
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if *count > available_cpus {
+ eprintln!(
+ "Invalid sharding configuration: cpu_allocation count
{count} exceeds available CPU cores {available_cpus}"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ Ok(())
}
-
- let available_cpus = available_parallelism()
- .map_err(|_| {
- eprintln!("Failed to detect available CPU cores");
- ConfigurationError::InvalidConfigurationValue
- })?
- .get();
-
- match &self.cpu_allocation {
- CpuAllocation::All => Ok(()),
- CpuAllocation::Count(count) => {
- if *count == 0 {
- eprintln!("Invalid sharding configuration: cpu_allocation
count cannot be 0");
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
- if *count > available_cpus {
- eprintln!(
- "Invalid sharding configuration: cpu_allocation count
{count} exceeds available CPU cores {available_cpus}"
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
- Ok(())
+ CpuAllocation::Range(start, end) => {
+ if start >= end {
+ eprintln!(
+ "Invalid sharding configuration: cpu_allocation range
{start}..{end} is invalid (start must be less than end)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
}
- CpuAllocation::Range(start, end) => {
- if start >= end {
- eprintln!(
- "Invalid sharding configuration: cpu_allocation range
{start}..{end} is invalid (start must be less than end)"
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
- if *end - *start > available_cpus {
- eprintln!(
- "Invalid sharding configuration: cpu_allocation range
{start}..{end} yields {} shards, exceeding available CPU cores
{available_cpus}",
- *end - *start
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
- if !self.pin_cores {
- return Ok(());
- }
- let allowed = allowed_cpus();
- if let Some(cpu) = (*start..*end).find(|cpu|
!allowed.contains(cpu)) {
- eprintln!(
- "Invalid sharding configuration: cpu_allocation range
{start}..{end} includes CPU {cpu}, which is outside the set of cores allowed
for this process (affinity/cpuset mask)"
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
- }
- Ok(())
+ if *end - *start > available_cpus {
+ eprintln!(
+ "Invalid sharding configuration: cpu_allocation range
{start}..{end} yields {} shards, exceeding available CPU cores
{available_cpus}",
+ *end - *start
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if !pin_cores {
+ return Ok(());
+ }
+ let allowed = allowed_cpus();
+ if let Some(cpu) = (*start..*end).find(|cpu|
!allowed.contains(cpu)) {
+ eprintln!(
+ "Invalid sharding configuration: cpu_allocation range
{start}..{end} includes CPU {cpu}, which is outside the set of cores allowed
for this process (affinity/cpuset mask)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
}
- // NUMA topology validation requires hwlocality (runtime dep).
- // Full NUMA validation happens in shard_allocator at startup.
- CpuAllocation::NumaAware(_) => Ok(()),
+ Ok(())
}
+ // NUMA topology validation requires hwlocality (runtime dep).
+ // Full NUMA validation happens in shard_allocator at startup.
+ CpuAllocation::NumaAware(_) => Ok(()),
+ }
+}
+
+impl Validatable<ConfigurationError> for ShardingConfig {
+ fn validate(&self) -> Result<(), ConfigurationError> {
+ validate_cpu_allocation(&self.cpu_allocation, self.pin_cores)
}
}
@@ -958,69 +886,6 @@ mod cluster_shards_count_determinism_tests {
}
}
-#[cfg(test)]
-mod sharding_shutdown_knob_tests {
- use super::*;
- use crate::server_config::sharding::{SHUTDOWN_DRAIN_TIMEOUT_MAX,
SHUTDOWN_POLL_INTERVAL_MAX};
- use iggy_common::IggyDuration;
- use std::time::Duration;
-
- #[test]
- fn defaults_validate() {
- assert!(ShardingConfig::default().validate().is_ok());
- }
-
- #[test]
- fn zero_drain_is_rejected() {
- let cfg = ShardingConfig {
- shutdown_drain_timeout: IggyDuration::new(Duration::ZERO),
- ..ShardingConfig::default()
- };
- assert!(cfg.validate().is_err());
- }
-
- #[test]
- fn over_cap_drain_is_rejected() {
- let cfg = ShardingConfig {
- shutdown_drain_timeout: IggyDuration::new(
- SHUTDOWN_DRAIN_TIMEOUT_MAX + Duration::from_secs(1),
- ),
- ..ShardingConfig::default()
- };
- assert!(cfg.validate().is_err());
- }
-
- #[test]
- fn zero_poll_is_rejected() {
- let cfg = ShardingConfig {
- shutdown_poll_interval: IggyDuration::new(Duration::ZERO),
- ..ShardingConfig::default()
- };
- assert!(cfg.validate().is_err());
- }
-
- #[test]
- fn over_cap_poll_is_rejected() {
- let cfg = ShardingConfig {
- shutdown_poll_interval: IggyDuration::new(
- SHUTDOWN_POLL_INTERVAL_MAX + Duration::from_secs(1),
- ),
- ..ShardingConfig::default()
- };
- assert!(cfg.validate().is_err());
- }
-
- #[test]
- fn poll_greater_than_drain_is_rejected() {
- let cfg = ShardingConfig {
- shutdown_drain_timeout:
IggyDuration::new(Duration::from_millis(20)),
- shutdown_poll_interval:
IggyDuration::new(Duration::from_millis(50)),
- ..ShardingConfig::default()
- };
- assert!(cfg.validate().is_err());
- }
-}
-
#[cfg(test)]
mod sharding_cpu_range_tests {
use super::*;
@@ -1029,7 +894,7 @@ mod sharding_cpu_range_tests {
fn inverted_range_is_rejected() {
let cfg = ShardingConfig {
cpu_allocation: CpuAllocation::Range(2, 2),
- ..ShardingConfig::default()
+ pin_cores: true,
};
assert!(cfg.validate().is_err());
}
@@ -1039,7 +904,7 @@ mod sharding_cpu_range_tests {
let first = allowed_cpus()[0];
let cfg = ShardingConfig {
cpu_allocation: CpuAllocation::Range(first, first + 1),
- ..ShardingConfig::default()
+ pin_cores: true,
};
assert!(cfg.validate().is_ok());
}
@@ -1049,7 +914,7 @@ mod sharding_cpu_range_tests {
let past_last = allowed_cpus().last().copied().unwrap() + 1;
let cfg = ShardingConfig {
cpu_allocation: CpuAllocation::Range(past_last, past_last + 1),
- ..ShardingConfig::default()
+ pin_cores: true,
};
assert!(cfg.validate().is_err());
}
@@ -1063,7 +928,7 @@ mod sharding_cpu_range_tests {
let available = available_parallelism().unwrap().get();
let cfg = ShardingConfig {
cpu_allocation: CpuAllocation::Range(first, first + available + 1),
- ..ShardingConfig::default()
+ pin_cores: true,
};
assert!(cfg.validate().is_err());
}
@@ -1075,7 +940,6 @@ mod sharding_cpu_range_tests {
let cfg = ShardingConfig {
cpu_allocation: CpuAllocation::Range(1 << 20, (1 << 20) + 1),
pin_cores: false,
- ..ShardingConfig::default()
};
assert!(cfg.validate().is_ok());
@@ -1083,8 +947,28 @@ mod sharding_cpu_range_tests {
let cfg = ShardingConfig {
cpu_allocation: CpuAllocation::Range(0, available + 1),
pin_cores: false,
- ..ShardingConfig::default()
};
assert!(cfg.validate().is_err());
}
}
+
+#[cfg(test)]
+mod sharding_embedded_default_tests {
+ use super::*;
+ use figment::Figment;
+ use figment::providers::{Format, Toml};
+
+ // Guards the single source of truth: the legacy sharding defaults resolve
+ // from the embedded legacy TOML, not hard-coded Rust values.
+ #[test]
+ fn legacy_embedded_toml_resolves_sharding_defaults() {
+ let toml_str = include_str!("../../../server/config.toml");
+ let config: ServerConfig = Figment::new()
+ .merge(Toml::string(toml_str))
+ .extract()
+ .expect("embedded legacy TOML deserializes");
+ config.validate().expect("embedded legacy config validates");
+
+ assert!(config.system.sharding.pin_cores);
+ }
+}
diff --git a/core/configs/src/server_ng_config/defaults.rs
b/core/configs/src/server_ng_config/defaults.rs
index ff4ba2f85..c7e724934 100644
--- a/core/configs/src/server_ng_config/defaults.rs
+++ b/core/configs/src/server_ng_config/defaults.rs
@@ -29,6 +29,7 @@
use super::message_bus::MessageBusConfig;
use super::quic::{QuicCertificateConfig, QuicConfig, QuicSocketConfig};
+use super::server_ng::NgSystemConfig;
use super::server_ng::{ExtraConfig, ServerNgConfig};
use super::tcp::{TcpConfig, TcpSocketConfig, TcpTlsConfig};
use super::websocket::{WebSocketConfig, WebSocketTlsConfig};
@@ -38,7 +39,6 @@ use crate::server_config::server::{
ConsumerGroupConfig, DataMaintenanceConfig, HeartbeatConfig,
MessageSaverConfig,
PersonalAccessTokenConfig, TelemetryConfig,
};
-use crate::server_config::system::SystemConfig;
use std::sync::Arc;
static_toml::static_toml! {
@@ -55,7 +55,7 @@ impl Default for ServerNgConfig {
heartbeat: HeartbeatConfig::default(),
message_saver: MessageSaverConfig::default(),
personal_access_token: PersonalAccessTokenConfig::default(),
- system: Arc::new(SystemConfig::default()),
+ system: Arc::new(NgSystemConfig::default()),
quic: QuicConfig::default(),
tcp: TcpConfig::default(),
websocket: WebSocketConfig::default(),
diff --git a/core/configs/src/server_ng_config/mod.rs
b/core/configs/src/server_ng_config/mod.rs
index a6e353682..e3196e521 100644
--- a/core/configs/src/server_ng_config/mod.rs
+++ b/core/configs/src/server_ng_config/mod.rs
@@ -32,6 +32,7 @@ pub mod displays;
pub mod message_bus;
pub mod quic;
pub mod server_ng;
+pub mod sharding;
pub mod tcp;
pub mod validators;
pub mod websocket;
diff --git a/core/configs/src/server_ng_config/server_ng.rs
b/core/configs/src/server_ng_config/server_ng.rs
index 56bd0a60d..0323f6df6 100644
--- a/core/configs/src/server_ng_config/server_ng.rs
+++ b/core/configs/src/server_ng_config/server_ng.rs
@@ -41,6 +41,11 @@ use std::sync::Arc;
const DEFAULT_CONFIG_PATH: &str = "core/server-ng/config.toml";
+/// The `server-ng` flavour of [`SystemConfig`], bound to this crate's own
+/// [`super::sharding::ShardingConfig`]. `core/server-ng` names this alias
+/// wherever it refers to the system config.
+pub type NgSystemConfig = SystemConfig<super::sharding::ShardingConfig>;
+
/// Top-level on-disk config schema for the `server-ng` binary.
///
/// Mirrors the legacy [`crate::server::ServerConfig`] section surface
@@ -64,7 +69,7 @@ pub struct ServerNgConfig {
pub message_saver: MessageSaverConfig,
pub personal_access_token: PersonalAccessTokenConfig,
pub heartbeat: HeartbeatConfig,
- pub system: Arc<SystemConfig>,
+ pub system: Arc<NgSystemConfig>,
pub quic: QuicConfig,
pub tcp: TcpConfig,
pub http: HttpConfig,
diff --git a/core/configs/src/server_ng_config/sharding.rs
b/core/configs/src/server_ng_config/sharding.rs
new file mode 100644
index 000000000..1f2e3f419
--- /dev/null
+++ b/core/configs/src/server_ng_config/sharding.rs
@@ -0,0 +1,357 @@
+// 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.
+
+//! Sharding config for `server-ng`. Forked from the legacy
+//! [`crate::server_config::sharding`] because the two servers own different
+//! knob sets and different default sources: this type carries the full
+//! thread-per-core + bus surface and reads its defaults from the server-ng
+//! TOML, while the legacy type keeps only `cpu_allocation` + `pin_cores`.
+
+use iggy_common::IggyDuration;
+use iggy_common::Validatable;
+use serde::{Deserialize, Serialize};
+use serde_with::{DisplayFromStr, serde_as};
+use std::time::Duration;
+
+use super::defaults::SERVER_NG_CONFIG;
+use crate::ConfigurationError;
+use crate::server_config::validators::validate_cpu_allocation;
+use configs::ConfigEnv;
+
+// Re-exported so the `configs::ng_sharding::*` path mirrors the legacy
+// `configs::sharding::*` surface for callers.
+pub use cpu_allocation::{CpuAllocation, NumaConfig};
+
+/// Maximum permitted per-shard inbox depth. The channel is allocated
+/// up-front per shard, so a runaway value here OOMs the process at boot.
+/// `1 << 20` (~1M frames) is several orders of magnitude above any
+/// realistic backpressure target and still fits comfortably in process
+/// address space.
+pub const INBOX_CAPACITY_MAX: usize = 1 << 20;
+
+/// Hard upper bound on `shutdown_drain_timeout`. A drain that never
+/// completes wedges process exit; capping at 10 minutes guarantees the
+/// watchdog eventually force-tears the bus even with a pathological
+/// config typo.
+pub const SHUTDOWN_DRAIN_TIMEOUT_MAX: Duration = Duration::from_secs(600);
+
+/// Hard upper bound on `shutdown_poll_interval`. A poll interval longer
+/// than the drain timeout makes the flag effectively unobservable; cap
+/// at 5s so Ctrl-C latency stays bounded regardless of config.
+pub const SHUTDOWN_POLL_INTERVAL_MAX: Duration = Duration::from_secs(5);
+
+/// Hard upper bound on `reconcile_periodic_interval`. A tick longer
+/// than ~30s makes post-failure recovery latency operator-visible; the
+/// cap reins in pathological typos without disturbing reasonable
+/// production values.
+pub const RECONCILE_PERIODIC_INTERVAL_MAX: Duration = Duration::from_secs(30);
+
+// Every omitted field falls back to the frozen `Default`, so a partial
+// `[system.sharding]` table resolves each key independently instead of
+// failing on the first missing one (parity with the legacy type).
+#[serde_as]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
+#[serde(default)]
+pub struct ShardingConfig {
+ #[serde(default)]
+ #[config_env(leaf)]
+ pub cpu_allocation: CpuAllocation,
+ /// Whether shard threads are pinned to dedicated CPU cores
+ /// (`sched_setaffinity`). Pinning maximizes cache locality when this
+ /// server owns its cores (dedicated host, `numa:` allocations). Set to
+ /// `false` when the server shares cores with other workloads — e.g. a
+ /// multi-tenant host slicing CPU via cgroup quotas — where every process
+ /// pinning to the same low-numbered cores would pile onto one core while
+ /// the rest sit idle; unpinned shards let the kernel scheduler place
+ /// threads freely within the allowed set. With a NUMA-aware allocation,
+ /// `false` drops both the CPU and memory-node bindings (and logs a
+ /// warning, since NUMA placement without pinning is meaningless).
+ pub pin_cores: bool,
+ /// Per-shard inter-shard inbox channel capacity. Bounded by design.
+ /// Drops on full inbox of consensus frames are recovered by VSR
+ /// retransmit. Drops of cross-shard client Reply frames are terminal:
+ /// the client never receives the reply (no in-protocol retransmit).
+ /// Both frame classes share this one channel, so a consensus burst
+ /// can starve client-reply forwards: size against the worst-case sum
+ /// of consensus working set + peak client-reply fan-out per shard
+ /// occurring together.
+ ///
+ // TODO(hubcio): split into two priority lanes - one bounded queue for
+ // consensus frames (drops recovered by VSR retransmit) and one for
+ // client `Reply` frames (drops terminal, must be sized for worst-case
+ // fan-out). Current single-channel design is the minimum-viable
+ // wiring so `frame_drops_total{variant,reason}` surfaces under load
+ // and yields real numbers to size the split against.
+ pub inbox_capacity: usize,
+ /// Wall-clock budget for a single shard's bus drain on shutdown.
+ /// Drives `IggyMessageBus::shutdown(..)` from the per-shard watchdog
+ /// and the parallel-join survivor path. Sized larger than typical
+ /// TCP RTT times in-flight write-batch so writers receive their full
+ /// last `write_vectored_all` budget before the connection registry
+ /// force-tears the bus. Slow-fsync hosts may need to extend this past
+ /// the default; the cap is `SHUTDOWN_DRAIN_TIMEOUT_MAX` so a config
+ /// typo cannot wedge process exit.
+ #[serde_as(as = "DisplayFromStr")]
+ #[config_env(leaf)]
+ pub shutdown_drain_timeout: IggyDuration,
+ /// Poll cadence for the cross-thread shutdown flag and for the
+ /// `await_metadata_bundle` / `broadcast_metadata_bundle` poll loops.
+ /// Trades off Ctrl-C latency against idle wakeup cost; the default
+ /// keeps shutdown observably prompt without measurable scheduler
+ /// overhead. Capped at `SHUTDOWN_POLL_INTERVAL_MAX` so the flag
+ /// remains effectively observable regardless of config.
+ #[serde_as(as = "DisplayFromStr")]
+ #[config_env(leaf)]
+ pub shutdown_poll_interval: IggyDuration,
+ /// Safety-tick cadence for the partition reconciliation loop; the
+ /// reconciler also wakes immediately on every
+ /// `LifecycleFrame::MetadataCommitTick` from shard 0. This periodic
+ /// fallback covers dropped wake-ups (the wake channel is capacity-1)
+ /// and the initial post-bootstrap convergence window. Values above
+ /// [`RECONCILE_PERIODIC_INTERVAL_MAX`] are rejected by the validator.
+ #[serde_as(as = "DisplayFromStr")]
+ #[config_env(leaf)]
+ pub reconcile_periodic_interval: IggyDuration,
+}
+
+impl Default for ShardingConfig {
+ fn default() -> Self {
+ Self {
+ cpu_allocation: CpuAllocation::default(),
+ pin_cores: SERVER_NG_CONFIG.system.sharding.pin_cores,
+ inbox_capacity: SERVER_NG_CONFIG.system.sharding.inbox_capacity as
usize,
+ shutdown_drain_timeout: SERVER_NG_CONFIG
+ .system
+ .sharding
+ .shutdown_drain_timeout
+ .parse()
+ .unwrap(),
+ shutdown_poll_interval: SERVER_NG_CONFIG
+ .system
+ .sharding
+ .shutdown_poll_interval
+ .parse()
+ .unwrap(),
+ reconcile_periodic_interval: SERVER_NG_CONFIG
+ .system
+ .sharding
+ .reconcile_periodic_interval
+ .parse()
+ .unwrap(),
+ }
+ }
+}
+
+impl Validatable<ConfigurationError> for ShardingConfig {
+ fn validate(&self) -> Result<(), ConfigurationError> {
+ if self.inbox_capacity == 0 {
+ eprintln!(
+ "Invalid sharding configuration: inbox_capacity must be > 0
(crossfire silently \
+ rounds 0 to 1, masking config errors)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if self.inbox_capacity > INBOX_CAPACITY_MAX {
+ eprintln!(
+ "Invalid sharding configuration: inbox_capacity {} exceeds the
{} cap (each \
+ shard preallocates a channel of this size; oversizing here
OOMs the process at \
+ boot)",
+ self.inbox_capacity, INBOX_CAPACITY_MAX
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ let drain = self.shutdown_drain_timeout.get_duration();
+ if drain.is_zero() {
+ eprintln!(
+ "Invalid sharding configuration: shutdown_drain_timeout must
be > 0 (a zero \
+ budget force-tears the bus mid-WAL-fsync on every shutdown)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if drain > SHUTDOWN_DRAIN_TIMEOUT_MAX {
+ eprintln!(
+ "Invalid sharding configuration: shutdown_drain_timeout {:?}
exceeds the {:?} \
+ cap (an unbounded drain wedges process exit on bus stall)",
+ drain, SHUTDOWN_DRAIN_TIMEOUT_MAX
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ let poll = self.shutdown_poll_interval.get_duration();
+ if poll.is_zero() {
+ eprintln!(
+ "Invalid sharding configuration: shutdown_poll_interval must
be > 0 (a zero \
+ cadence busy-loops every shard's watchdog and
metadata-handoff poller)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if poll > SHUTDOWN_POLL_INTERVAL_MAX {
+ eprintln!(
+ "Invalid sharding configuration: shutdown_poll_interval {:?}
exceeds the {:?} \
+ cap (a coarse cadence stalls Ctrl-C handling and metadata
handoff abort)",
+ poll, SHUTDOWN_POLL_INTERVAL_MAX
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if poll > drain {
+ eprintln!(
+ "Invalid sharding configuration: shutdown_poll_interval {:?}
must be <= \
+ shutdown_drain_timeout {:?} (a poll cadence coarser than the
drain budget makes \
+ the shutdown flag effectively unobservable)",
+ poll, drain
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ let reconcile = self.reconcile_periodic_interval.get_duration();
+ if reconcile.is_zero() {
+ eprintln!(
+ "Invalid sharding configuration: reconcile_periodic_interval
resolves to zero. \
+ Note that \"0\", \"none\", \"unlimited\", and \"disabled\"
all parse to zero. The \
+ periodic reconcile tick is a safety net for dropped
commit-wakes and cannot be \
+ turned off; set a positive duration (default \"1s\", max
{RECONCILE_PERIODIC_INTERVAL_MAX:?})."
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if reconcile > RECONCILE_PERIODIC_INTERVAL_MAX {
+ eprintln!(
+ "Invalid sharding configuration: reconcile_periodic_interval
{:?} exceeds the \
+ {:?} cap (a long tick makes post-failure convergence latency
operator-visible)",
+ reconcile, RECONCILE_PERIODIC_INTERVAL_MAX
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ validate_cpu_allocation(&self.cpu_allocation, self.pin_cores)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::server_ng_config::server_ng::ServerNgConfig;
+ use figment::Figment;
+ use figment::providers::{Format, Toml};
+
+ #[test]
+ fn defaults_validate() {
+ assert!(ShardingConfig::default().validate().is_ok());
+ }
+
+ #[test]
+ fn zero_drain_is_rejected() {
+ let cfg = ShardingConfig {
+ shutdown_drain_timeout: IggyDuration::new(Duration::ZERO),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn over_cap_drain_is_rejected() {
+ let cfg = ShardingConfig {
+ shutdown_drain_timeout: IggyDuration::new(
+ SHUTDOWN_DRAIN_TIMEOUT_MAX + Duration::from_secs(1),
+ ),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn zero_poll_is_rejected() {
+ let cfg = ShardingConfig {
+ shutdown_poll_interval: IggyDuration::new(Duration::ZERO),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn over_cap_poll_is_rejected() {
+ let cfg = ShardingConfig {
+ shutdown_poll_interval: IggyDuration::new(
+ SHUTDOWN_POLL_INTERVAL_MAX + Duration::from_secs(1),
+ ),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn poll_greater_than_drain_is_rejected() {
+ let cfg = ShardingConfig {
+ shutdown_drain_timeout:
IggyDuration::new(Duration::from_millis(20)),
+ shutdown_poll_interval:
IggyDuration::new(Duration::from_millis(50)),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ // Guards the single source of truth: the server-ng sharding defaults
+ // resolve from the embedded server-ng TOML, not hard-coded Rust values.
+ #[test]
+ fn ng_embedded_toml_resolves_sharding_defaults() {
+ let toml_str = include_str!("../../../server-ng/config.toml");
+ let config: ServerNgConfig = Figment::new()
+ .merge(Toml::string(toml_str))
+ .extract()
+ .expect("embedded server-ng TOML deserializes");
+ config
+ .validate()
+ .expect("embedded server-ng config validates");
+
+ let sharding = &config.system.sharding;
+ assert!(sharding.pin_cores);
+ assert_eq!(sharding.inbox_capacity, 1024);
+ assert_eq!(sharding.shutdown_drain_timeout, "10 s".parse().unwrap());
+ assert_eq!(sharding.shutdown_poll_interval, "50 ms".parse().unwrap());
+ assert_eq!(sharding.reconcile_periodic_interval, "1
s".parse().unwrap());
+ }
+
+ // Extract straight from a raw table (no embedded base layer) so the
+ // struct-level `#[serde(default)]` is what fills the gaps, not the
+ // provider's embedded-TOML fallback.
+ #[test]
+ fn partial_table_fills_missing_fields_with_frozen_defaults() {
+ let sharding: ShardingConfig = Figment::new()
+ .merge(Toml::string("pin_cores = false"))
+ .extract()
+ .expect("partial sharding table deserializes");
+
+ assert!(!sharding.pin_cores);
+ assert_eq!(sharding.inbox_capacity, 1024);
+ assert_eq!(sharding.shutdown_drain_timeout, "10 s".parse().unwrap());
+ assert_eq!(sharding.shutdown_poll_interval, "50 ms".parse().unwrap());
+ assert_eq!(sharding.reconcile_periodic_interval, "1
s".parse().unwrap());
+ }
+
+ #[test]
+ fn empty_table_yields_all_frozen_defaults() {
+ let sharding: ShardingConfig = Figment::new()
+ .merge(Toml::string(""))
+ .extract()
+ .expect("empty sharding table deserializes");
+
+ assert!(sharding.pin_cores);
+ assert_eq!(sharding.inbox_capacity, 1024);
+ assert_eq!(sharding.shutdown_drain_timeout, "10 s".parse().unwrap());
+ assert_eq!(sharding.shutdown_poll_interval, "50 ms".parse().unwrap());
+ assert_eq!(sharding.reconcile_periodic_interval, "1
s".parse().unwrap());
+ }
+}
diff --git a/core/configs_derive/src/config_env.rs
b/core/configs_derive/src/config_env.rs
index 7556cbb13..33f15e9f6 100644
--- a/core/configs_derive/src/config_env.rs
+++ b/core/configs_derive/src/config_env.rs
@@ -247,7 +247,7 @@ fn generate_struct_impl(
let const_definitions = generate_const_definitions(&mappings, prefix_str);
let mapping_entries = generate_mapping_entries(&mappings);
- let builder_methods = generate_builder_methods(&mappings, struct_name,
prefix_str);
+ let builder_methods = generate_builder_methods(&mappings, prefix_str);
let builder_name = format_ident!("{}EnvBuilder", struct_name);
let mappings_count = mappings.len();
@@ -572,16 +572,11 @@ fn generate_mapping_entries(mappings: &[EnvMapping]) ->
Vec<TokenStream2> {
.collect()
}
-fn generate_builder_methods(
- mappings: &[EnvMapping],
- struct_name: &Ident,
- prefix: &str,
-) -> Vec<TokenStream2> {
+fn generate_builder_methods(mappings: &[EnvMapping], prefix: &str) ->
Vec<TokenStream2> {
mappings
.iter()
.map(|m| {
let method_name = &m.builder_method_name;
- let const_name = &m.const_name;
let env_var_name = format!("{}{}", prefix, m.env_suffix);
let doc_secret = if m.is_secret {
" (secret - will be masked in logs)"
@@ -593,7 +588,7 @@ fn generate_builder_methods(
quote! {
#[doc = #doc]
pub fn #method_name(mut self, value: impl std::fmt::Display)
-> Self {
- self.envs.insert(#struct_name::#const_name.into(),
value.to_string());
+ self.envs.insert(#env_var_name.into(), value.to_string());
self
}
}
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index 70af981f5..92638aa18 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -637,7 +637,14 @@ cpu_allocation = 1
# (e.g. a multi-tenant host slicing CPU via cgroup quotas): unpinned shards
# let the kernel scheduler place threads freely instead of piling every
# process onto the same low-numbered cores.
-# pin_cores = true
+pin_cores = true
+
+# Per-shard inter-shard inbox capacity. Bounded by design: consensus-frame
+# drops recover via VSR retransmit, but cross-shard client-reply drops are
+# terminal. Size for the worst-case sum of both: the consensus working set
+# (~ PIPELINE_PREPARE_QUEUE_MAX (32) * replica_count * directions) plus peak
+# client-reply fan-out per shard.
+inbox_capacity = 1024
# Wall-clock budget for a single shard's bus drain on shutdown. Drives
# the per-shard watchdog and the parallel-join survivor path; sized
@@ -653,6 +660,11 @@ shutdown_drain_timeout = "10 s"
# scheduler overhead. Must be less than or equal to shutdown_drain_timeout.
shutdown_poll_interval = "50 ms"
+# Safety-tick cadence for the partition reconciliation loop. The reconciler
+# also wakes on every metadata commit from shard 0, so this only covers
+# dropped wake-ups and the initial post-bootstrap convergence window.
+reconcile_periodic_interval = "1 s"
+
[websocket]
enabled = true
address = "127.0.0.1:8092"
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 1c3a6799a..8d079bcd8 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -30,10 +30,10 @@ use crate::partition_helpers::{
use crate::segment_recovery::{RecoveredSegment, load_persisted_segments};
use crate::server_error::{ServerNgError, ShardJoinFailure,
ShardJoinFailureKind};
use crate::session_manager::SessionManager;
-use configs::server_ng::ServerNgConfig;
-use configs::sharding::{
+use configs::ng_sharding::{
INBOX_CAPACITY_MAX, SHUTDOWN_DRAIN_TIMEOUT_MAX, SHUTDOWN_POLL_INTERVAL_MAX,
};
+use configs::server_ng::ServerNgConfig;
use consensus::{LocalPipeline, MetadataHandle, PartitionsHandle, Sequencer,
VsrConsensus};
// `try_send` / `try_recv` resolve through these traits on `MAsyncTx` /
// `MAsyncRx`; the metadata-handoff loops below depend on the
@@ -462,7 +462,7 @@ pub async fn load_config(logging: &mut Logging) ->
Result<ServerNgConfig, Server
/// invariant is held by the type system, not by hoping the operator
/// never configures 65535 cores worth of shards.
fn resolve_shard_assignments(
- sharding: &configs::sharding::ShardingConfig,
+ sharding: &configs::ng_sharding::ShardingConfig,
) -> Result<(Vec<ShardInfo>, u16), ServerNgError> {
let allocator = ShardAllocator::new(&sharding.cpu_allocation,
sharding.pin_cores)
.map_err(ServerNgError::ShardAllocator)?;
@@ -486,7 +486,7 @@ fn resolve_shard_assignments(
/// usage) cannot OOM at boot or wedge process exit with an out-of-range
/// value.
fn validate_sharding_runtime_knobs(
- sharding: &configs::sharding::ShardingConfig,
+ sharding: &configs::ng_sharding::ShardingConfig,
) -> Result<(), ServerNgError> {
let inbox_capacity = sharding.inbox_capacity;
if inbox_capacity == 0 || inbox_capacity > INBOX_CAPACITY_MAX {
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index 5d510860a..6838a742a 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -51,7 +51,7 @@ use crate::snapshot;
use crate::users::maybe_rewrite_user_password_request;
use crate::wire::{request_body, usize_to_u32};
use bytes::Bytes;
-use configs::system::SystemConfig;
+use configs::server_ng::NgSystemConfig;
use consensus::{
Consensus, EvictionContext, MetadataHandle, PartitionsHandle,
build_eviction_message,
build_incompatible_protocol_eviction_message,
@@ -112,7 +112,7 @@ pub(crate) type ActiveClientRequests =
Rc<RefCell<HashSet<u128>>>;
pub(crate) fn make_client_request_handler(
shard: &Rc<ServerNgShard>,
sessions: &Rc<RefCell<SessionManager>>,
- system_config: Arc<SystemConfig>,
+ system_config: Arc<NgSystemConfig>,
) -> RequestHandler {
let shard = Rc::clone(shard);
let sessions = Rc::clone(sessions);
@@ -404,7 +404,7 @@ pub(crate) fn make_deferred_client_request_handler(
bus: &Rc<IggyMessageBus>,
shard_handle: &ServerNgShardHandle,
sessions: &Rc<RefCell<SessionManager>>,
- system_config: Arc<SystemConfig>,
+ system_config: Arc<NgSystemConfig>,
) -> RequestHandler {
let shard_handle = Rc::clone(shard_handle);
let sessions = Rc::clone(sessions);
@@ -536,7 +536,7 @@ pub(crate) fn make_metadata_submit_handler(
fn enqueue_client_request(
shard: Rc<ServerNgShard>,
sessions: Rc<RefCell<SessionManager>>,
- system_config: Arc<SystemConfig>,
+ system_config: Arc<NgSystemConfig>,
queues: ClientRequestQueues,
active: ActiveClientRequests,
client_id: u128,
@@ -561,7 +561,7 @@ fn enqueue_client_request(
async fn drain_client_requests(
shard: Rc<ServerNgShard>,
sessions: Rc<RefCell<SessionManager>>,
- system_config: Arc<SystemConfig>,
+ system_config: Arc<NgSystemConfig>,
queues: ClientRequestQueues,
active: ActiveClientRequests,
client_id: u128,
@@ -598,7 +598,7 @@ fn pop_next_client_request(
async fn handle_client_request(
shard: &Rc<ServerNgShard>,
sessions: &Rc<RefCell<SessionManager>>,
- system_config: &Arc<SystemConfig>,
+ system_config: &Arc<NgSystemConfig>,
transport_client_id: u128,
message: Message<iggy_binary_protocol::GenericHeader>,
) {
@@ -1035,7 +1035,7 @@ pub(crate) async fn dispatch_partition_request(
async fn handle_non_replicated_request(
shard: &Rc<ServerNgShard>,
sessions: &Rc<RefCell<SessionManager>>,
- system_config: &Arc<SystemConfig>,
+ system_config: &Arc<NgSystemConfig>,
transport_client_id: u128,
request: Message<RequestHeader>,
) {
@@ -1232,7 +1232,7 @@ async fn handle_default_non_replicated(
#[allow(clippy::future_not_send)]
async fn handle_get_snapshot(
shard: &Rc<ServerNgShard>,
- system_config: &Arc<SystemConfig>,
+ system_config: &Arc<NgSystemConfig>,
transport_client_id: u128,
request: &Message<RequestHeader>,
user_id: Option<u32>,
diff --git a/core/server-ng/src/http.rs b/core/server-ng/src/http.rs
index 38c402940..4163d83d8 100644
--- a/core/server-ng/src/http.rs
+++ b/core/server-ng/src/http.rs
@@ -46,7 +46,7 @@ use axum::middleware::{Next, from_fn};
use axum::routing::{delete, get, post, put};
use configs::cluster::{ClusterConfig, TransportPorts};
use configs::http::HttpConfig;
-use configs::system::SystemConfig;
+use configs::server_ng::NgSystemConfig;
use message_bus::client_listener;
use send_wrapper::SendWrapper;
use tracing::{error, info};
@@ -83,7 +83,7 @@ pub async fn start(
addr: SocketAddr,
http_config: &HttpConfig,
cluster: &ClusterConfig,
- system_config: Arc<SystemConfig>,
+ system_config: Arc<NgSystemConfig>,
self_ports: TransportPorts,
) -> Result<(), ServerNgError> {
let jwt = JwtManager::build(&http_config.jwt)?;
diff --git a/core/server-ng/src/http/state.rs b/core/server-ng/src/http/state.rs
index 9ddfb811f..55df0d8e8 100644
--- a/core/server-ng/src/http/state.rs
+++ b/core/server-ng/src/http/state.rs
@@ -26,7 +26,7 @@ use std::sync::Arc;
use axum::http::{HeaderName, HeaderValue};
use axum::response::Response;
-use configs::system::SystemConfig;
+use configs::server_ng::NgSystemConfig;
use consensus::{MetadataHandle, VsrConsensus};
use iggy_common::{ClusterMetadata, IggyTimestamp};
use message_bus::InstanceToken;
@@ -67,7 +67,7 @@ pub(in crate::http) struct HttpInner {
/// Read-only server config for the snapshot collector (log directory +
/// runtime config paths); the shard does not expose config on the read
/// path.
- pub(in crate::http) system_config: Arc<SystemConfig>,
+ pub(in crate::http) system_config: Arc<NgSystemConfig>,
/// Per-credential VSR sessions keyed by JWT `jti` / PAT hash. `RefCell` is
/// sound here - shard 0 is single-threaded and the `SendWrapper` state
/// bridge tolerates the `!Sync` interior - but the guard must never be
held
diff --git a/core/server-ng/src/partition_reconciler.rs
b/core/server-ng/src/partition_reconciler.rs
index ca0da51cd..37d3f5d3b 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -822,7 +822,7 @@ pub fn install_tick_handler(shard: &Rc<ServerNgShard>,
wake_tx: WakeTx) {
#[cfg(test)]
mod tests {
use super::{FailureCause, FailureRecord, ReconcilerCtx, reconcile_once};
- use configs::server_ng::ServerNgConfig;
+ use configs::server_ng::{NgSystemConfig, ServerNgConfig};
use consensus::{MetadataHandle, PartitionsHandle};
use iggy_binary_protocol::codec::WireEncode;
use iggy_binary_protocol::primitives::identifier::WireName;
@@ -1024,13 +1024,13 @@ mod tests {
fn test_config(tmp: &TempDir) -> ServerNgConfig {
let mut cfg = ServerNgConfig::default();
- // `SystemConfig` is not `Clone`, so `Arc::make_mut` is out; build a
+ // `NgSystemConfig` is not `Clone`, so `Arc::make_mut` is out; build a
// fresh value via struct-update syntax and swap the Arc wholesale.
// Only `path` differs from the default; every other field uses the
// runtime's defaults.
- let system = configs::system::SystemConfig {
+ let system = NgSystemConfig {
path: tmp.path().to_string_lossy().into_owned(),
- ..configs::system::SystemConfig::default()
+ ..NgSystemConfig::default()
};
cfg.system = Arc::new(system);
cfg
diff --git a/core/server-ng/src/snapshot.rs b/core/server-ng/src/snapshot.rs
index e67447da7..8761265ae 100644
--- a/core/server-ng/src/snapshot.rs
+++ b/core/server-ng/src/snapshot.rs
@@ -29,7 +29,7 @@ use std::time::Instant;
use async_zip::base::write::ZipFileWriter;
use async_zip::{Compression, ZipEntryBuilder};
-use configs::system::SystemConfig;
+use configs::server_ng::NgSystemConfig;
use futures::channel::oneshot;
use iggy_common::{IggyDuration, IggyError, SnapshotCompression,
SystemSnapshotType};
use tracing::{error, info, warn};
@@ -55,7 +55,7 @@ static SNAPSHOT_IN_PROGRESS: AtomicBool =
AtomicBool::new(false);
/// [`SNAPSHOT_IN_PROGRESS`]); a concurrent request busy-rejects with
/// [`IggyError::SnapshotFileCompletionFailed`].
pub async fn collect(
- system_config: Arc<SystemConfig>,
+ system_config: Arc<NgSystemConfig>,
compression: SnapshotCompression,
snapshot_types: Vec<SystemSnapshotType>,
) -> Result<Vec<u8>, IggyError> {
@@ -128,7 +128,7 @@ impl Drop for SnapshotInProgressGuard {
}
fn collect_blocking(
- system_config: &SystemConfig,
+ system_config: &NgSystemConfig,
compression: SnapshotCompression,
snapshot_types: &[SystemSnapshotType],
) -> Result<Vec<u8>, IggyError> {
@@ -156,7 +156,7 @@ fn collect_blocking(
fn capture(
snapshot_type: &SystemSnapshotType,
- system_config: &SystemConfig,
+ system_config: &NgSystemConfig,
) -> io::Result<Vec<u8>> {
match snapshot_type {
SystemSnapshotType::FilesystemOverview => {
@@ -195,7 +195,7 @@ fn process_list() -> io::Result<Vec<u8>> {
Ok(content)
}
-fn server_logs(system_config: &SystemConfig) -> io::Result<Vec<u8>> {
+fn server_logs(system_config: &NgSystemConfig) -> io::Result<Vec<u8>> {
// Mirror the logger's path derivation (server_common
`Logging::late_init`):
// it canonicalizes the configured subdirectory before joining the system
// path, so a relative `logging.path` that already exists resolves against
the
@@ -224,7 +224,7 @@ fn server_logs(system_config: &SystemConfig) ->
io::Result<Vec<u8>> {
Ok(content)
}
-fn server_config(system_config: &SystemConfig) -> io::Result<Vec<u8>> {
+fn server_config(system_config: &NgSystemConfig) -> io::Result<Vec<u8>> {
let config_path =
PathBuf::from(system_config.get_runtime_path()).join("current_config.toml");
std::fs::read(config_path)
}
@@ -277,7 +277,7 @@ mod tests {
// second collector thread) rather than piling up threads.
let held = SnapshotInProgressGuard::acquire().expect("flag starts
free");
let result = futures::executor::block_on(collect(
- Arc::new(SystemConfig::default()),
+ Arc::new(NgSystemConfig::default()),
SnapshotCompression::Stored,
vec![SystemSnapshotType::Test],
));
diff --git a/core/server/config.toml b/core/server/config.toml
index b31bf8a1c..2add7ca24 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -599,7 +599,7 @@ cpu_allocation = "numa:auto"
# (e.g. a multi-tenant host slicing CPU via cgroup quotas): unpinned shards
# let the kernel scheduler place threads freely instead of piling every
# process onto the same low-numbered cores.
-# pin_cores = true
+pin_cores = true
[websocket]
enabled = true
