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 931bde2aa766f48ebb2d8069151e4898e3c4d8fd Author: Hubert Gruszecki <[email protected]> AuthorDate: Fri Jul 24 17:59:38 2026 +0200 feat(server-ng): make VSR repair pacing and eviction ring configurable Repair behavior was pinned at compile time: the evicted ring kept 4096 entries / 16 MiB per partition, retries paced at 100 ticks, and chunks capped at 128 frames. Operators could not trade repair memory for catch-up speed, and nothing stopped a config that sets message_bus.peer_queue_capacity at or below the chunk size, where a repair round overruns the per-peer queue and silently drops its own tail, wedging repair into slow retries. Surface [partition] evicted_ring_capacity / evicted_ring_bytes_max and [cluster] repair_retry_interval / repair_chunk_max, applied at boot through setters so the simulator and tests keep compile-time defaults. Ring caps apply on both fresh and recovered partitions; pacing applies once per shard. A new top-level check requires the chunk to stay strictly below the peer queue, which also floors peer_queue_capacity for the first time. The ring-overflow rejoin scenario now pins its premise against the default capacity so raising the default fails loud instead of going false-green. --- core/configs/src/server_ng_config/cluster.rs | 107 +++++++++++++++++++++ core/configs/src/server_ng_config/defaults.rs | 8 ++ core/configs/src/server_ng_config/partition.rs | 105 +++++++++++++++++++- core/configs/src/server_ng_config/validators.rs | 37 +++++++ .../scenarios/reconnect_after_restart_scenario.rs | 9 ++ core/partitions/src/journal.rs | 25 ++++- core/partitions/src/lib.rs | 1 + core/server-ng/config.toml | 25 +++++ core/server-ng/src/bootstrap.rs | 90 +++++++++++++++++ core/server-ng/src/partition_helpers.rs | 8 ++ core/shard/src/lib.rs | 59 +++++++++--- 11 files changed, 456 insertions(+), 18 deletions(-) diff --git a/core/configs/src/server_ng_config/cluster.rs b/core/configs/src/server_ng_config/cluster.rs index 4fdb87b91..131318400 100644 --- a/core/configs/src/server_ng_config/cluster.rs +++ b/core/configs/src/server_ng_config/cluster.rs @@ -61,6 +61,17 @@ pub const DEFAULT_VIEW_PROBE_ATTEMPTS_MAX: u32 = 5; /// typo guard, not a sizing endorsement. const MAX_VIEW_PROBE_ATTEMPTS: u32 = 100; +/// Default per-round repair-serving chunk. Duplicated here rather than imported +/// so `core/configs` keeps off a build-time edge onto `core/shard` (mirroring +/// [`DEFAULT_VIEW_PROBE_ATTEMPTS_MAX`]); `core/server-ng`'s bootstrap +/// static-asserts it equal to `shard::REPAIR_CHUNK_MAX`. +pub const DEFAULT_REPAIR_CHUNK_MAX: usize = 128; + +/// Upper bound on `repair_chunk_max`. A chunk rides the per-peer bus queue, so +/// the load-bearing rule is `repair_chunk_max < message_bus.peer_queue_capacity` +/// (enforced at the top level); this standalone ceiling is a typo guard. +const MAX_REPAIR_CHUNK_MAX: usize = 1024; + /// Length floor for the replica-auth PSK, in raw bytes. The 32-byte MAC key /// is KDF-derived from these bytes at use-site, so any encoding clearing this /// length is accepted. @@ -128,6 +139,22 @@ fn default_view_probe_attempts_max() -> u32 { SERVER_NG_CONFIG.cluster.view_probe_attempts_max as u32 } +/// serde fallback for configs written before the field existed; the value +/// itself lives in `core/server-ng/config.toml` like every other default. +fn default_repair_retry_interval() -> IggyDuration { + SERVER_NG_CONFIG + .cluster + .repair_retry_interval + .parse() + .unwrap() +} + +/// serde fallback for configs written before the field existed; the value +/// itself lives in `core/server-ng/config.toml` like every other default. +fn default_repair_chunk_max() -> usize { + SERVER_NG_CONFIG.cluster.repair_chunk_max as usize +} + #[serde_as] #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] #[serde(deny_unknown_fields)] @@ -199,6 +226,25 @@ pub struct ClusterConfig { /// `MAX_VIEW_PROBE_ATTEMPTS`. #[serde(default = "default_view_probe_attempts_max")] pub view_probe_attempts_max: u32, + /// How long a stalled journal-repair stream waits before re-requesting its + /// remaining window from the serving peer. Repair frames are + /// fire-and-forget over the lossy bus, so a session with no retry wedges + /// forever on a single dropped frame. Paces both the metadata and + /// partition repair loops. Sizes the retry threshold in consensus ticks. + /// Zero (and the `0` / `disabled` / `unlimited` sentinels, which all parse + /// to zero) is rejected at boot. + #[serde(default = "default_repair_retry_interval")] + #[serde_as(as = "DisplayFromStr")] + #[config_env(leaf)] + pub repair_retry_interval: IggyDuration, + /// Prepares a peer serves per repair round before the requester walks to + /// the next chunk. Each frame rides the per-peer message-bus queue, so this + /// must stay below `message_bus.peer_queue_capacity` or a full round + /// overruns the queue and silently drops frames (enforced at the top + /// level). Applies to both the metadata and partition repair planes. Must + /// be > 0 and <= `MAX_REPAIR_CHUNK_MAX`. + #[serde(default = "default_repair_chunk_max")] + pub repair_chunk_max: usize, /// Full roster of cluster members. Intended to be byte-identical across /// every node so operators ship one config. The running node's identity /// is supplied out-of-band via the `--replica-id` CLI flag, which @@ -432,6 +478,31 @@ impl Validatable<ConfigurationError> for ClusterConfig { return Err(ConfigurationError::InvalidConfigurationValue); } + // The repair retry interval sizes a tick threshold that has to advance; + // `0` / `disabled` / `unlimited` all collapse to zero and would wedge + // every stalled repair stream - reject them. The chunk is a per-round + // count with a standalone typo ceiling; the load-bearing rule against + // message_bus.peer_queue_capacity is enforced at the top level. + if self.repair_retry_interval.get_duration().is_zero() { + eprintln!( + "Invalid cluster configuration: cluster.repair_retry_interval must be nonzero \ + (it paces stalled-repair retries)" + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.repair_chunk_max == 0 { + eprintln!("Invalid cluster configuration: cluster.repair_chunk_max must be > 0"); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.repair_chunk_max > MAX_REPAIR_CHUNK_MAX { + eprintln!( + "Invalid cluster configuration: cluster.repair_chunk_max ({}) exceeds the maximum \ + ({MAX_REPAIR_CHUNK_MAX})", + self.repair_chunk_max + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.nodes.is_empty() { eprintln!( "Invalid cluster configuration: cluster.nodes must contain at least one entry when cluster is enabled" @@ -596,6 +667,8 @@ mod tests { request_start_view_retransmit_interval: default_request_start_view_retransmit_interval( ), view_probe_attempts_max: default_view_probe_attempts_max(), + repair_retry_interval: default_repair_retry_interval(), + repair_chunk_max: default_repair_chunk_max(), nodes: Vec::new(), auth: ClusterAuthConfig { enabled: true, @@ -640,6 +713,8 @@ mod cluster_validate_tests { request_start_view_retransmit_interval: default_request_start_view_retransmit_interval( ), view_probe_attempts_max: default_view_probe_attempts_max(), + repair_retry_interval: default_repair_retry_interval(), + repair_chunk_max: default_repair_chunk_max(), nodes, auth: ClusterAuthConfig::default(), tls: ClusterTlsConfig::default(), @@ -755,6 +830,38 @@ mod cluster_validate_tests { assert!(c.validate().is_ok()); } + #[test] + fn validate_rejects_zero_repair_retry_interval() { + // `0` / `disabled` / `unlimited` all collapse to zero and would wedge + // stalled repair streams. + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.repair_retry_interval = IggyDuration::new(Duration::ZERO); + assert!(c.validate().is_err()); + } + + #[test] + fn validate_rejects_zero_repair_chunk_max() { + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.repair_chunk_max = 0; + assert!(c.validate().is_err()); + } + + #[test] + fn validate_rejects_repair_chunk_max_above_ceiling() { + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.repair_chunk_max = MAX_REPAIR_CHUNK_MAX + 1; + assert!(c.validate().is_err()); + } + + #[test] + fn validate_accepts_repair_chunk_max_at_ceiling() { + // Section-level validate only; the cross-section rule against + // message_bus.peer_queue_capacity lives in the top-level validate. + let mut c = cfg(vec![node("n1", 0), node("n2", 1)]); + c.repair_chunk_max = MAX_REPAIR_CHUNK_MAX; + assert!(c.validate().is_ok()); + } + #[test] fn validate_rejects_empty_nodes() { let c = cfg(vec![]); diff --git a/core/configs/src/server_ng_config/defaults.rs b/core/configs/src/server_ng_config/defaults.rs index 4c3768c24..560602751 100644 --- a/core/configs/src/server_ng_config/defaults.rs +++ b/core/configs/src/server_ng_config/defaults.rs @@ -107,6 +107,12 @@ impl Default for ClusterConfig { .parse() .unwrap(), view_probe_attempts_max: SERVER_NG_CONFIG.cluster.view_probe_attempts_max as u32, + repair_retry_interval: SERVER_NG_CONFIG + .cluster + .repair_retry_interval + .parse() + .unwrap(), + repair_chunk_max: SERVER_NG_CONFIG.cluster.repair_chunk_max as usize, nodes: SERVER_NG_CONFIG .cluster .nodes @@ -168,6 +174,8 @@ impl Default for PartitionConfig { let partition = &SERVER_NG_CONFIG.partition; PartitionConfig { prepare_queue_depth: partition.prepare_queue_depth as usize, + evicted_ring_capacity: partition.evicted_ring_capacity as usize, + evicted_ring_bytes_max: partition.evicted_ring_bytes_max.parse().unwrap(), } } } diff --git a/core/configs/src/server_ng_config/partition.rs b/core/configs/src/server_ng_config/partition.rs index 8bf4f7cb6..580668e4b 100644 --- a/core/configs/src/server_ng_config/partition.rs +++ b/core/configs/src/server_ng_config/partition.rs @@ -17,11 +17,14 @@ //! On-disk schema for the per-partition consensus plane. //! -//! One capacity knob previously hardcoded in the runtime crates: +//! Capacity knobs 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) +//! - `evicted_ring_capacity` -> `partitions::EVICTED_RING_CAPACITY` and +//! `evicted_ring_bytes_max` -> `partitions::EVICTED_RING_BYTES_MAX` +//! (the per-partition journal-repair retention ring's dual ceilings) //! //! Distinct from `[metadata]` (a single, shard-0-global VSR plane) because //! partition pipelines exist PER PARTITION. The default mirrors the runtime @@ -38,7 +41,7 @@ use super::COMPONENT_NG; use crate::ConfigurationError; use configs::ConfigEnv; -use iggy_common::Validatable; +use iggy_common::{IggyByteSize, Validatable}; use serde::{Deserialize, Serialize}; /// Mirrors `consensus::PIPELINE_PREPARE_QUEUE_MAX`. @@ -52,6 +55,22 @@ pub const DEFAULT_PARTITION_PREPARE_QUEUE_DEPTH: usize = 32; /// sizing endorsement. pub const MAX_PARTITION_PREPARE_QUEUE_DEPTH: usize = 256; +/// Mirrors `partitions::EVICTED_RING_CAPACITY`. +pub const DEFAULT_EVICTED_RING_CAPACITY: usize = 4096; + +/// Upper bound on `evicted_ring_capacity`. The ring exists per multi-replica +/// partition and each retained entry pins a full committed batch, so +/// worst-case pinned memory scales with the partition count; a typo guard, +/// not a sizing endorsement. +pub const MAX_EVICTED_RING_CAPACITY: usize = 65536; + +/// Mirrors `partitions::EVICTED_RING_BYTES_MAX`. +pub const DEFAULT_EVICTED_RING_BYTES_MAX: u64 = 16 * 1024 * 1024; + +/// Upper bound on `evicted_ring_bytes_max`, per partition. Whichever ring cap +/// trips first evicts; this byte ceiling is the second typo guard. +pub const MAX_EVICTED_RING_BYTES: u64 = 256 * 1024 * 1024; + /// Capacity tunables for the per-partition consensus plane. #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] pub struct PartitionConfig { @@ -61,6 +80,21 @@ pub struct PartitionConfig { /// path the SDK retries. Applies to every partition; raising it multiplies /// pinned request-buffer memory by the partition count. pub prepare_queue_depth: usize, + + /// Entries the evicted ring retains per multi-replica partition for + /// journal repair after a peer rejoins. Larger widens the window a + /// restarting peer can be served from the ring before falling back to + /// bulk sync, at the cost of pinned memory per partition. Must be > 0 and + /// <= [`MAX_EVICTED_RING_CAPACITY`]. Single-replica partitions retain + /// nothing regardless. + pub evicted_ring_capacity: usize, + + /// Byte ceiling for the evicted ring per partition; whichever ring cap + /// (this or [`Self::evicted_ring_capacity`]) trips first evicts. Bounds + /// the ring memory a burst of large batches can pin. Must be > 0 and <= + /// [`MAX_EVICTED_RING_BYTES`]. + #[config_env(leaf)] + pub evicted_ring_bytes_max: IggyByteSize, } impl Validatable<ConfigurationError> for PartitionConfig { @@ -76,6 +110,28 @@ impl Validatable<ConfigurationError> for PartitionConfig { ); return Err(ConfigurationError::InvalidConfigurationValue); } + if self.evicted_ring_capacity == 0 { + eprintln!("{COMPONENT_NG} partition.evicted_ring_capacity must be > 0"); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if self.evicted_ring_capacity > MAX_EVICTED_RING_CAPACITY { + eprintln!( + "{COMPONENT_NG} partition.evicted_ring_capacity ({}) exceeds the maximum ({MAX_EVICTED_RING_CAPACITY})", + self.evicted_ring_capacity + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + let ring_bytes = self.evicted_ring_bytes_max.as_bytes_u64(); + if ring_bytes == 0 { + eprintln!("{COMPONENT_NG} partition.evicted_ring_bytes_max must be > 0"); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if ring_bytes > MAX_EVICTED_RING_BYTES { + eprintln!( + "{COMPONENT_NG} partition.evicted_ring_bytes_max ({ring_bytes} bytes) exceeds the maximum ({MAX_EVICTED_RING_BYTES} bytes)" + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } Ok(()) } } @@ -92,26 +148,65 @@ mod tests { } #[test] - fn rejects_zero() { + fn rejects_zero_prepare_queue_depth() { let config = PartitionConfig { prepare_queue_depth: 0, + ..PartitionConfig::default() }; assert!(config.validate().is_err()); } #[test] - fn rejects_above_ceiling() { + fn rejects_prepare_queue_depth_above_ceiling() { let config = PartitionConfig { prepare_queue_depth: MAX_PARTITION_PREPARE_QUEUE_DEPTH + 1, + ..PartitionConfig::default() }; assert!(config.validate().is_err()); } #[test] - fn accepts_at_ceiling() { + fn accepts_prepare_queue_depth_at_ceiling() { let config = PartitionConfig { prepare_queue_depth: MAX_PARTITION_PREPARE_QUEUE_DEPTH, + ..PartitionConfig::default() }; assert!(config.validate().is_ok()); } + + #[test] + fn rejects_zero_evicted_ring_capacity() { + let config = PartitionConfig { + evicted_ring_capacity: 0, + ..PartitionConfig::default() + }; + assert!(config.validate().is_err()); + } + + #[test] + fn rejects_evicted_ring_capacity_above_ceiling() { + let config = PartitionConfig { + evicted_ring_capacity: MAX_EVICTED_RING_CAPACITY + 1, + ..PartitionConfig::default() + }; + assert!(config.validate().is_err()); + } + + #[test] + fn rejects_zero_evicted_ring_bytes_max() { + let config = PartitionConfig { + evicted_ring_bytes_max: IggyByteSize::from(0_u64), + ..PartitionConfig::default() + }; + assert!(config.validate().is_err()); + } + + #[test] + fn rejects_evicted_ring_bytes_max_above_ceiling() { + let config = PartitionConfig { + evicted_ring_bytes_max: IggyByteSize::from(MAX_EVICTED_RING_BYTES + 1), + ..PartitionConfig::default() + }; + assert!(config.validate().is_err()); + } } diff --git a/core/configs/src/server_ng_config/validators.rs b/core/configs/src/server_ng_config/validators.rs index 8ff975ae7..cbebdcb36 100644 --- a/core/configs/src/server_ng_config/validators.rs +++ b/core/configs/src/server_ng_config/validators.rs @@ -160,6 +160,20 @@ impl Validatable<ConfigurationError> for ServerNgConfig { format!("{COMPONENT_NG} (error: {e}) - failed to validate message_bus config") })?; + // Repair frames ride the bounded per-peer message-bus queue. A repair + // round of cluster.repair_chunk_max frames that meets or overruns + // message_bus.peer_queue_capacity drops its own tail silently, wedging + // the repair loop into slow retries. Keep the chunk strictly below the + // queue; this also floors peer_queue_capacity, which is otherwise only + // checked for > 0. + if self.cluster.repair_chunk_max >= self.message_bus.peer_queue_capacity { + eprintln!( + "{COMPONENT_NG} cluster.repair_chunk_max ({}) must be < message_bus.peer_queue_capacity ({}): repair frames ride the per-peer bus queue, so a chunk that fills or overruns it drops frames and wedges repair", + self.cluster.repair_chunk_max, self.message_bus.peer_queue_capacity + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + self.quic.validate().error(|e: &ConfigurationError| { format!("{COMPONENT_NG} (error: {e}) - failed to validate quic config") })?; @@ -339,6 +353,29 @@ mod tests { assert!(config.validate().is_err()); } + #[test] + fn given_peer_queue_capacity_not_above_repair_chunk_max_when_validating_should_reject() { + // The default repair_chunk_max (128) must stay strictly below + // peer_queue_capacity; shrinking the queue to the chunk size is the + // silent wedged-repair footgun this cross-section guard closes. + let config = config_with_override("[message_bus]\npeer_queue_capacity = 128\n"); + assert!(config.validate().is_err()); + } + + #[test] + fn given_repair_chunk_max_at_peer_queue_capacity_when_validating_should_reject() { + let config = config_with_override("[cluster]\nrepair_chunk_max = 256\n"); + assert!(config.validate().is_err()); + } + + #[test] + fn given_repair_chunk_max_below_peer_queue_capacity_when_validating_should_pass() { + let config = config_with_override("[cluster]\nrepair_chunk_max = 255\n"); + config + .validate() + .expect("a chunk below the peer queue capacity must validate"); + } + /// The warn-helper baseline is [`ServerNgConfig::default`], but the reused /// legacy sections source that default from the legacy server config.toml, /// not this NG file. Pin the knobs the helper compares so any drift between diff --git a/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs b/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs index e994e5d44..bc2da26a7 100644 --- a/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs +++ b/core/integration/tests/server/scenarios/reconnect_after_restart_scenario.rs @@ -621,6 +621,15 @@ async fn poll_from_zero_until( pub async fn run_ring_overflow_rejoin(harness: &mut TestHarness) { const RING_OVERFLOW_OPS: u32 = 4300; const POST_RESTART_OPS: u32 = 50; + // The point of this scenario is overflowing the peers' evicted ring so + // RangeEvicted and the commit floor engage. The harness runs the server on + // the shipped default ring capacity; if that default ever reaches this op + // count the overflow stops happening and the test silently passes without + // covering the floor path. Fail loud instead. + const _: () = assert!( + RING_OVERFLOW_OPS as usize > configs::ng_partition::DEFAULT_EVICTED_RING_CAPACITY, + "RING_OVERFLOW_OPS must exceed the default evicted ring capacity, or this scenario no longer exercises RangeEvicted", + ); let setup_client = harness .root_client() diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index feb78c24b..fb24564f4 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -177,6 +177,14 @@ where evicted_ring: UnsafeCell<VecDeque<(u64, JournalBuffer)>>, /// Running byte total of the buffers held by `evicted_ring`. evicted_ring_bytes: Cell<u64>, + /// Entry-count ceiling for `evicted_ring`. Defaults to + /// [`EVICTED_RING_CAPACITY`]; server-ng overrides it from config at + /// partition build. + evicted_ring_capacity: Cell<usize>, + /// Byte ceiling for `evicted_ring`. Defaults to + /// [`EVICTED_RING_BYTES_MAX`]; server-ng overrides it from config at + /// partition build. + evicted_ring_bytes_max: Cell<u64>, /// Single-replica groups have nobody to repair; retaining evicted /// entries for them is pure memory waste. repair_retention: Cell<bool>, @@ -207,6 +215,8 @@ where }), evicted_ring: UnsafeCell::new(VecDeque::new()), evicted_ring_bytes: Cell::new(0), + evicted_ring_capacity: Cell::new(EVICTED_RING_CAPACITY), + evicted_ring_bytes_max: Cell::new(EVICTED_RING_BYTES_MAX), repair_retention: Cell::new(true), } } @@ -286,6 +296,15 @@ impl PartitionJournal<PartitionJournalMemStorage> { } } + /// Override the evicted-ring ceilings from configuration. Called once at + /// partition build, before any eviction, so the caps govern the first + /// flush onward. Leaves `repair_retention` untouched: the single-replica + /// disable path stands on its own. + pub fn set_ring_caps(&self, capacity: usize, bytes_max: u64) { + self.evicted_ring_capacity.set(capacity); + self.evicted_ring_bytes_max.set(bytes_max); + } + /// Resident (un-evicted) entry count; diagnostics only. pub fn resident_count(&self) -> usize { let op_to_storage_offset = unsafe { &*self.op_to_storage_offset.get() }; @@ -466,8 +485,8 @@ impl PartitionJournal<PartitionJournalMemStorage> { }; ring_bytes += entry.len() as u64; ring.push_back((op, entry)); - while ring.len() > EVICTED_RING_CAPACITY - || (ring_bytes > EVICTED_RING_BYTES_MAX && ring.len() > 1) + while ring.len() > self.evicted_ring_capacity.get() + || (ring_bytes > self.evicted_ring_bytes_max.get() && ring.len() > 1) { if let Some((_, dropped)) = ring.pop_front() { ring_bytes -= dropped.len() as u64; @@ -597,6 +616,8 @@ where inner: UnsafeCell::new(JournalInner { storage }), evicted_ring: UnsafeCell::new(VecDeque::new()), evicted_ring_bytes: Cell::new(0), + evicted_ring_capacity: Cell::new(EVICTED_RING_CAPACITY), + evicted_ring_bytes_max: Cell::new(EVICTED_RING_BYTES_MAX), repair_retention: Cell::new(true), } } diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs index 0abd137d1..33e33d02e 100644 --- a/core/partitions/src/lib.rs +++ b/core/partitions/src/lib.rs @@ -37,6 +37,7 @@ pub use iggy_index_reader::IggyIndexReader; pub use iggy_index_writer::IggyIndexWriter; pub use iggy_partition::IggyPartition; pub use iggy_partitions::IggyPartitions; +pub use journal::{EVICTED_RING_BYTES_MAX, EVICTED_RING_CAPACITY}; pub use messages_writer::MessagesWriter; pub use offset_storage::delete_persisted_offset; pub use poll_plan::{AutoCommitApplied, PollPlan}; diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml index eb5510271..fe259284d 100644 --- a/core/server-ng/config.toml +++ b/core/server-ng/config.toml @@ -634,6 +634,19 @@ request_start_view_retransmit_interval = "1s" # Must be between 1 and 100. view_probe_attempts_max = 5 +# How long a stalled journal-repair stream waits before re-requesting its +# remaining window from the serving peer (duration). Repair frames are +# fire-and-forget over the lossy bus, so a session with no retry wedges forever +# on a single dropped frame. Paces both the metadata and partition repair loops; +# must be nonzero. +repair_retry_interval = "1s" + +# Prepares a peer serves per repair round before the requester walks to the next +# chunk (integer). Each frame rides the per-peer message-bus queue, so this must +# stay strictly below message_bus.peer_queue_capacity or a full round overruns +# the queue and drops frames. Must be > 0 and <= 1024. +repair_chunk_max = 128 + # Replica-to-replica authentication (PSK + BLAKE3 keyed-MAC handshake). [cluster.auth] # When true, every replica peer must complete the authenticated handshake or be @@ -789,6 +802,18 @@ clients_table_max = 8192 # retried by the SDK. Must be > 0 and <= 256. prepare_queue_depth = 32 +# Entries the evicted ring retains per multi-replica partition for journal +# repair after a peer rejoins. Larger widens the window a restarting peer can be +# served from the ring before falling back to bulk sync, at the cost of pinned +# memory per partition. Must be > 0 and <= 65536. Single-replica partitions +# retain nothing regardless. +evicted_ring_capacity = 4096 + +# Byte ceiling for the evicted ring per partition; whichever ring cap (this or +# evicted_ring_capacity) trips first evicts. Bounds the ring memory a burst of +# large batches can pin. Must be > 0 and <= "256 MiB". +evicted_ring_bytes_max = "16 MiB" + # 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 a78cec889..c7b59df25 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -1688,6 +1688,10 @@ async fn build_shard_for_thread( .map_err(ServerNgError::ShardConstruction)?; let shard = Rc::new(built.shard); + // Repair pacing is shared by both planes' repair loops, so it is a + // per-shard tunable set once here rather than per consensus group. + shard.set_repair_retry_ticks(repair_retry_ticks(config)); + shard.set_repair_chunk_max(config.cluster.repair_chunk_max as u64); *shard_handle.borrow_mut() = Some(Rc::downgrade(&shard)); Ok((shard, sessions)) } @@ -1713,6 +1717,14 @@ const _: () = assert!( ); const _: () = assert!(configs::ng_cluster::DEFAULT_VIEW_PROBE_ATTEMPTS_MAX == consensus::PROBE_ATTEMPTS_MAX); +const _: () = assert!( + configs::ng_partition::DEFAULT_EVICTED_RING_CAPACITY == partitions::EVICTED_RING_CAPACITY +); +const _: () = assert!( + configs::ng_partition::DEFAULT_EVICTED_RING_BYTES_MAX == partitions::EVICTED_RING_BYTES_MAX +); +const _: () = + assert!(configs::ng_cluster::DEFAULT_REPAIR_CHUNK_MAX as u64 == shard::REPAIR_CHUNK_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 { @@ -1808,6 +1820,17 @@ pub(crate) fn request_start_view_ticks(config: &ServerNgConfig) -> u64 { ) } +/// `[cluster] repair_retry_interval` in consensus ticks: how long a stalled +/// journal-repair stream waits before re-requesting its window. Both planes' +/// repair loops share it, so it is applied once per shard (not per consensus +/// group). Clamped to `u32`, the width of the session idle-tick counter. +pub(crate) fn repair_retry_ticks(config: &ServerNgConfig) -> u32 { + u32::try_from(duration_to_ticks( + config.cluster.repair_retry_interval.get_duration(), + )) + .unwrap_or(u32::MAX) +} + #[allow(clippy::too_many_arguments)] fn restore_metadata_consensus( journal: &PrepareJournal, @@ -2018,6 +2041,13 @@ async fn load_partition( })?; let mut partition = IggyPartition::new(stats.clone(), consensus); + // Recovered partitions honor the same config-surfaced ring ceilings as the + // fresh-create path (build_partition_fresh). Retention is already off for + // single-replica groups, so this only sizes the multi-replica ring. + partition.log.journal().inner.set_ring_caps( + config.partition.evicted_ring_capacity, + config.partition.evicted_ring_bytes_max.as_bytes_u64(), + ); partition.set_partition_dir(config.system.get_partition_path( stream_id, topic_id, @@ -3415,6 +3445,66 @@ mod tests { ); } + #[test] + fn default_repair_retry_interval_matches_partitions_constant() { + // The config default lives in core/server-ng/config.toml (a string, so + // no static assert can pin it); keep it in lockstep with the built-in + // the simulator and un-configured replicas run on. + let config_default = configs::ng_cluster::ClusterConfig::default() + .repair_retry_interval + .get_duration() + .as_millis(); + let built_in = + u128::from(partitions::REPAIR_RETRY_TICKS) * shard::CONSENSUS_TICK_INTERVAL.as_millis(); + assert_eq!( + config_default, built_in, + "[cluster] repair_retry_interval default drifted from \ + partitions::REPAIR_RETRY_TICKS" + ); + } + + #[test] + fn default_repair_chunk_max_matches_shard_constant() { + // Belt and suspenders with the static assert above: that pins the + // duplicated configs-crate literal, this pins the shipped config.toml + // value the simulator and un-configured replicas run on. + let config_default = configs::ng_cluster::ClusterConfig::default().repair_chunk_max; + assert_eq!( + config_default as u64, + shard::REPAIR_CHUNK_MAX, + "[cluster] repair_chunk_max default drifted from shard::REPAIR_CHUNK_MAX" + ); + } + + #[test] + fn default_evicted_ring_capacity_matches_partitions_constant() { + // Belt and suspenders with the static assert above; this pins the + // shipped config.toml value. + let config_default = + configs::ng_partition::PartitionConfig::default().evicted_ring_capacity; + assert_eq!( + config_default, + partitions::EVICTED_RING_CAPACITY, + "[partition] evicted_ring_capacity default drifted from \ + partitions::EVICTED_RING_CAPACITY" + ); + } + + #[test] + fn default_evicted_ring_bytes_max_matches_partitions_constant() { + // Belt and suspenders with the static assert above; this pins the + // shipped config.toml value. + let config_default = configs::ng_partition::PartitionConfig::default() + .evicted_ring_bytes_max + .as_bytes_u64(); + assert_eq!( + config_default, + partitions::EVICTED_RING_BYTES_MAX, + "[partition] evicted_ring_bytes_max default drifted from \ + partitions::EVICTED_RING_BYTES_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 f6a10f349..e76da4408 100644 --- a/core/server-ng/src/partition_helpers.rs +++ b/core/server-ng/src/partition_helpers.rs @@ -523,6 +523,14 @@ pub async fn build_partition_fresh( } let mut partition = IggyPartition::new(stats, consensus); + // Surface the evicted-ring ceilings from config onto the fresh journal. + // IggyPartition::new has already disabled retention for single-replica + // groups (nobody to serve), so this only sizes the multi-replica ring; the + // caps are inert while retention is off. + partition.log.journal().inner.set_ring_caps( + config.partition.evicted_ring_capacity, + config.partition.evicted_ring_bytes_max.as_bytes_u64(), + ); partition.set_partition_dir(config.system.get_partition_path( stream_id, topic_id, diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 60803f5b5..a89be1ff0 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -56,7 +56,7 @@ use partitions::{IggyPartition, IggyPartitions, PollFragments, PollingArgs, Poll use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId}; use server_common::{MESSAGE_ALIGN, Message, MessageBag, iobuf::Frozen}; use shards_table::ShardsTable; -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; use std::collections::{HashMap, VecDeque}; use std::future::Future; use std::rc::Rc; @@ -708,12 +708,16 @@ impl ShardFrame { } } -/// Prepares served per `RequestPrepares` round. The per-peer bus queues -/// are bounded (`peer_queue_capacity`, 256 by default) and overrun frames -/// drop silently, so an unbounded burst loses its own tail; the receiver -/// pulls the window chunk by chunk instead (each walked `RepairDone` -/// immediately requests the next chunk while progress holds). -const REPAIR_CHUNK_MAX: u64 = 128; +/// Prepares served per `RequestPrepares` round. +/// +/// The per-peer bus queues are bounded (`peer_queue_capacity`, 256 by default) +/// and overrun frames drop silently, so an unbounded burst loses its own tail; +/// the receiver pulls the window chunk by chunk instead (each walked +/// `RepairDone` immediately requests the next chunk while progress holds). +/// +/// Runtime default; server-ng overrides the live ceiling per shard from +/// `[cluster] repair_chunk_max` at bootstrap. +pub const REPAIR_CHUNK_MAX: u64 = 128; /// One in-flight metadata journal-repair stream (shard 0 only). #[derive(Debug, Clone, Copy)] @@ -824,6 +828,16 @@ where /// `ReconcileOp::InsertOwned` lands. Bounded per namespace; overflow /// drops the frame (at-least-once: client/primary retries recover). pending_partition_frames: RefCell<HashMap<IggyNamespace, Vec<Message<GenericHeader>>>>, + + /// Live ceiling on prepares served per `RequestPrepares` round. Defaults + /// to [`REPAIR_CHUNK_MAX`]; server-ng overrides it from + /// `[cluster] repair_chunk_max` at bootstrap. + repair_chunk_max: Cell<u64>, + + /// Live stalled-repair retry threshold in consensus ticks. Defaults to + /// [`partitions::REPAIR_RETRY_TICKS`]; server-ng overrides it from + /// `[cluster] repair_retry_interval` at bootstrap. + repair_retry_ticks: Cell<u32>, } impl<B, MJ, S, M, T> IggyShard<B, MJ, S, M, T> @@ -916,9 +930,25 @@ where reconcile_queue: RefCell::new(VecDeque::new()), pending_partition_frames: RefCell::new(HashMap::new()), metadata_repair: RefCell::new(None), + repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX), + repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS), }) } + /// Override the stalled-repair retry threshold (consensus ticks) from + /// configuration. Called once per shard at bootstrap; the simulator and + /// tests keep the compile-time [`partitions::REPAIR_RETRY_TICKS`] default. + pub fn set_repair_retry_ticks(&self, ticks: u32) { + self.repair_retry_ticks.set(ticks); + } + + /// Override the per-round repair-serving chunk ceiling from configuration. + /// Called once per shard at bootstrap; the simulator and tests keep the + /// compile-time [`REPAIR_CHUNK_MAX`] default. + pub fn set_repair_chunk_max(&self, chunk: u64) { + self.repair_chunk_max.set(chunk); + } + /// Hand a metadata consensus submit (login/logout) to shard 0. /// /// Sends a [`LifecycleFrame::MetadataSubmit`] into shard 0's inbox. The @@ -1123,6 +1153,8 @@ where reconcile_queue: RefCell::new(VecDeque::new()), pending_partition_frames: RefCell::new(HashMap::new()), metadata_repair: RefCell::new(None), + repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX), + repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS), } } @@ -2036,6 +2068,9 @@ where { let header = *msg.header(); let target = header.replica; + // Snapshot the config-overridable chunk ceiling once; both plane + // branches below serve the same per-round window. + let repair_chunk_max = self.repair_chunk_max.get(); let planes = self.plane.inner(); if let Some(ref consensus) = planes.0.consensus && consensus.namespace() == header.namespace @@ -2097,7 +2132,7 @@ where ) .await; } - let chunk_end = to_op.min(from_op.saturating_add(REPAIR_CHUNK_MAX - 1)); + let chunk_end = to_op.min(from_op.saturating_add(repair_chunk_max - 1)); let mut served_through = from_op.saturating_sub(1); for op in from_op..=chunk_end { #[allow(clippy::cast_possible_truncation)] @@ -2157,7 +2192,7 @@ where .await; from_op = retained_from; } - let chunk_end = to_op.min(from_op.saturating_add(REPAIR_CHUNK_MAX - 1)); + let chunk_end = to_op.min(from_op.saturating_add(repair_chunk_max - 1)); let mut served_through = from_op.saturating_sub(1); for op in from_op..=chunk_end { let Some(entry) = partition.log.journal().inner.repair_entry(op) else { @@ -2541,6 +2576,7 @@ where >, { let partitions = self.plane.partitions(); + let repair_retry_ticks = self.repair_retry_ticks.get(); // Fan out over every group (each partition's heartbeat/retransmit timer // must advance), so the keyed single-namespace lookup the control-frame // handlers use does not apply here. The namespaces are snapshotted into @@ -2579,7 +2615,7 @@ where return None; } session.idle_ticks += 1; - if session.idle_ticks < partitions::REPAIR_RETRY_TICKS { + if session.idle_ticks < repair_retry_ticks { return None; } session.idle_ticks = 0; @@ -2693,6 +2729,7 @@ where // Stall retry, mirroring `tick_partitions`: a lost repair frame must // not wedge the session forever. + let repair_retry_ticks = self.repair_retry_ticks.get(); let stalled = { let mut session = self.metadata_repair.borrow_mut(); session.as_mut().and_then(|session| { @@ -2700,7 +2737,7 @@ where return None; } session.idle_ticks += 1; - if session.idle_ticks < partitions::REPAIR_RETRY_TICKS { + if session.idle_ticks < repair_retry_ticks { return None; } session.idle_ticks = 0;
