This is an automated email from the ASF dual-hosted git repository.
numinnex 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 8ed41c7db refactor(server): consensus restore, fail-stop, linear
cleanup batch (#3911)
8ed41c7db is described below
commit 8ed41c7db2221d0e1fb3b2859e0586afc172f3db
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Thu Aug 20 12:42:41 2026 +0200
refactor(server): consensus restore, fail-stop, linear cleanup batch (#3911)
---
core/configs/src/common/server.rs | 14 +-
core/configs/src/server_config/cluster.rs | 37 +++++
core/configs/src/server_config/defaults.rs | 4 +-
core/metadata/src/impls/metadata.rs | 9 +-
core/partitions/src/iggy_partition.rs | 9 +-
core/partitions/src/lib.rs | 6 +-
core/partitions/src/types.rs | 82 +++++-----
core/server/config.toml | 13 ++
core/server/src/bootstrap.rs | 81 ++++++----
core/server/src/dispatch.rs | 7 +-
core/server/src/main.rs | 4 +-
core/server/src/partition_reconciler.rs | 233 ++++++++++++++++++++++++++++-
core/server/src/responses.rs | 176 +++++++++++++++++-----
core/server_common/src/buffer.rs | 4 +-
core/server_common/src/executor.rs | 100 +++++++++++--
core/server_common/src/lib.rs | 2 +-
core/server_common/src/memory_pool.rs | 23 ++-
core/shard/src/config.rs | 9 +-
core/shard/src/lib.rs | 8 +-
core/simulator/src/bin/simulator-ui.rs | 4 +-
core/simulator/src/bin/workload-fuzz.rs | 4 +-
core/simulator/src/lib.rs | 56 +++----
core/simulator/src/replica.rs | 3 +-
23 files changed, 671 insertions(+), 217 deletions(-)
diff --git a/core/configs/src/common/server.rs
b/core/configs/src/common/server.rs
index 315abeadc..887cfe8ab 100644
--- a/core/configs/src/common/server.rs
+++ b/core/configs/src/common/server.rs
@@ -20,7 +20,7 @@ use iggy_common::{IggyByteSize, IggyDuration};
use serde::{Deserialize, Serialize};
use serde_with::DisplayFromStr;
use serde_with::serde_as;
-use server_common::MemoryPoolConfigOther;
+use server_common::MemoryPoolSettings;
use server_common::log::{TelemetryEndpointSettings, TelemetrySettings};
pub use server_common::log::TelemetryTransport;
@@ -34,12 +34,12 @@ pub struct MemoryPoolConfig {
pub bucket_capacity: u32,
}
-impl MemoryPoolConfig {
- pub fn into_other(&self) -> MemoryPoolConfigOther {
- MemoryPoolConfigOther {
- enabled: self.enabled,
- size: self.size,
- bucket_capacity: self.bucket_capacity,
+impl From<&MemoryPoolConfig> for MemoryPoolSettings {
+ fn from(config: &MemoryPoolConfig) -> Self {
+ Self {
+ enabled: config.enabled,
+ size: config.size,
+ bucket_capacity: config.bucket_capacity,
}
}
}
diff --git a/core/configs/src/server_config/cluster.rs
b/core/configs/src/server_config/cluster.rs
index a1874a640..d71f86ae6 100644
--- a/core/configs/src/server_config/cluster.rs
+++ b/core/configs/src/server_config/cluster.rs
@@ -299,6 +299,41 @@ pub struct ClusterConfig {
/// Replica-to-replica TLS settings for the consensus (`tcp_replica`) port.
#[serde(default)]
pub tls: ClusterTlsConfig,
+ /// Shard-0 coordinator placement tunables.
+ #[serde(default)]
+ pub coordinator: ClusterCoordinatorConfig,
+}
+
+/// Placement tunables for the shard-0 coordinator, converted into the shard
+/// crate's `CoordinatorConfig` at bootstrap (the domain type lives there
+/// because `configs` and `shard` share no dependency edge).
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
+#[serde(deny_unknown_fields)]
+pub struct ClusterCoordinatorConfig {
+ /// When `total_shards > 1`, exclude shard 0 from replica placement.
+ /// Shard 0 already hosts the coordinator, the metadata writer, and both
+ /// listeners; replicas are long-lived steady flows, so offload them to
+ /// peer shards by default.
+ #[serde(default = "default_skip_shard_zero_for_replicas")]
+ pub skip_shard_zero_for_replicas: bool,
+ /// When `total_shards > 1`, exclude shard 0 from client placement.
+ /// Default false: client connections are short-lived and benefit from
+ /// shard-0 parallelism more than replicas do.
+ #[serde(default)]
+ pub skip_shard_zero_for_clients: bool,
+}
+
+fn default_skip_shard_zero_for_replicas() -> bool {
+ true
+}
+
+impl Default for ClusterCoordinatorConfig {
+ fn default() -> Self {
+ Self {
+ skip_shard_zero_for_replicas:
default_skip_shard_zero_for_replicas(),
+ skip_shard_zero_for_clients: false,
+ }
+ }
}
/// Replica-to-replica authentication for the consensus (`tcp_replica`) port.
@@ -1451,6 +1486,7 @@ mod tests {
previous_shared_secret:
"retiring-psk-MUST-NOT-be-persisted".to_owned(),
},
tls: ClusterTlsConfig::default(),
+ coordinator: ClusterCoordinatorConfig::default(),
};
let serialized = serde_json::to_string(&config).expect("serialize
cluster config");
assert!(
@@ -1783,6 +1819,7 @@ mod cluster_validate_tests {
nodes,
auth: ClusterAuthConfig::default(),
tls: ClusterTlsConfig::default(),
+ coordinator: ClusterCoordinatorConfig::default(),
}
}
diff --git a/core/configs/src/server_config/defaults.rs
b/core/configs/src/server_config/defaults.rs
index 1705e861b..f752d3265 100644
--- a/core/configs/src/server_config/defaults.rs
+++ b/core/configs/src/server_config/defaults.rs
@@ -24,7 +24,8 @@
//! [`crate::common::defaults`].
use super::cluster::{
- ClusterAuthConfig, ClusterConfig, ClusterNodeConfig, ClusterTlsConfig,
TransportPorts,
+ ClusterAuthConfig, ClusterConfig, ClusterCoordinatorConfig,
ClusterNodeConfig,
+ ClusterTlsConfig, TransportPorts,
};
use super::message_bus::MessageBusConfig;
use super::metadata::MetadataConfig;
@@ -148,6 +149,7 @@ impl Default for ClusterConfig {
.collect(),
auth: ClusterAuthConfig::default(),
tls: ClusterTlsConfig::default(),
+ coordinator: ClusterCoordinatorConfig::default(),
}
}
}
diff --git a/core/metadata/src/impls/metadata.rs
b/core/metadata/src/impls/metadata.rs
index ee67e1097..1fda606d1 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -1138,10 +1138,6 @@ where
return;
}
- // TODO add assertions for valid state here.
-
- // TODO handle gap in ops.
-
// Verify hash chain integrity BEFORE checkpoint.
`checkpoint_if_needed`
// can drain WAL entries, making previous_header return None.
if let Some(previous) = journal.handle().previous_header(&header) {
@@ -3423,12 +3419,11 @@ where
let header = *message.header();
- // TODO: calculate the index;
#[allow(clippy::cast_possible_truncation)]
- let idx = header.op as usize;
+ let op = header.op as usize;
assert_eq!(header.command, Command::Prepare);
assert!(
- journal.handle().header(idx).is_some(),
+ journal.handle().header(op).is_some(),
"replicate: prepare must be durable in local journal before
chain-forward"
);
if let Err(e) = replicate_to_next_in_chain(consensus, message).await {
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 4a366076d..369db555a 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -3704,10 +3704,10 @@ where
let enforce_fsync = self.effective_enforce_fsync(config);
let preallocate_segments = self.effective_preallocate_segments(config);
let segment = Segment::new(start_offset, segment_size);
- // `PartitionsConfig::get_messages_path` is a stub (`/tmp/iggy_stub`);
- // the partition's real directory is only known to the server config
- // that created the initial segment, so derive the rotated paths from
- // the active writer's location.
+ // Prefer the active writer's location: a per-topic path override or a
+ // config change after the initial segment was created must not scatter
+ // one partition's segments across two directories. The config layout
+ // only decides for a partition with no writer yet.
let (messages_path, index_path) = self.partition_dir().map_or_else(
|| {
(
@@ -6531,6 +6531,7 @@ mod tests {
segment_size: IggyByteSize::from(1024 * 1024),
preallocate_segments: false,
encryptor: None,
+ path_layout: crate::PartitionPathLayout::default(),
}
}
diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs
index 6d14d4b6b..4649af829 100644
--- a/core/partitions/src/lib.rs
+++ b/core/partitions/src/lib.rs
@@ -46,9 +46,9 @@ pub use segment::Segment;
use server_common::Message;
pub use server_common::send_messages::{IggyMessage, IggyMessageHeader,
IggyMessages};
pub use types::{
- AppendResult, Fragment, PartitionOffsets, PartitionsConfig, PollFragments,
PollQueryResult,
- PollingArgs, PollingConsumer, REPAIR_RETRY_TICKS, RepairConclusion,
RepairSession,
- SendMessagesResult,
+ AppendResult, Fragment, PartitionOffsets, PartitionPathLayout,
PartitionsConfig, PollFragments,
+ PollQueryResult, PollingArgs, PollingConsumer, REPAIR_RETRY_TICKS,
RepairConclusion,
+ RepairSession, SendMessagesResult,
};
/// Partition-level data plane operations.
diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs
index dcf4eb0b9..df15b4393 100644
--- a/core/partitions/src/types.rs
+++ b/core/partitions/src/types.rs
@@ -256,6 +256,32 @@ pub enum RepairConclusion {
FloorRefused { floor: u64, to_op: u64 },
}
+/// Where partition directories live on disk, mirroring the server's
+/// `SystemConfig` path scheme so segment files created by the partition plane
+/// land next to the ones the server bootstrap created.
+#[derive(Debug, Clone)]
+pub struct PartitionPathLayout {
+ /// `{system.path}/{stream.path}`: the directory holding per-stream dirs.
+ pub streams_root: String,
+ /// Directory name of the per-topic level (`topic.path`).
+ pub topics_dir: String,
+ /// Directory name of the per-partition level (`partition.path`).
+ pub partitions_dir: String,
+}
+
+/// Synthetic layout for tests and the simulator, where paths only key the
+/// sim storage and never touch a real filesystem. The server always wires
+/// the real layout from its `SystemConfig`.
+impl Default for PartitionPathLayout {
+ fn default() -> Self {
+ Self {
+ streams_root: "/tmp/iggy_stub/streams".to_string(),
+ topics_dir: "topics".to_string(),
+ partitions_dir: "partitions".to_string(),
+ }
+ }
+}
+
/// Configuration for partition operations.
///
/// Mirrors the relevant fields from the server's `PartitionConfig` and
@@ -286,6 +312,8 @@ pub struct PartitionsConfig {
/// decrypts uniformly whether a fragment came from the resident journal
/// or from disk.
pub encryptor: Option<Arc<EncryptorKind>>,
+ /// On-disk location scheme for partition directories.
+ pub path_layout: PartitionPathLayout,
}
impl PartitionsConfig {
@@ -296,14 +324,15 @@ impl PartitionsConfig {
topic_id: usize,
partition_id: usize,
) -> String {
-
format!("/tmp/iggy_stub/streams/{stream_id}/topics/{topic_id}/partitions/{partition_id}")
+ format!(
+ "{}/{stream_id}/{}/{topic_id}/{}/{partition_id}",
+ self.path_layout.streams_root,
+ self.path_layout.topics_dir,
+ self.path_layout.partitions_dir,
+ )
}
/// Constructs the file path for segment messages.
- ///
- /// TODO: This is a stub waiting for completion of issue to move server
config
- /// to shared module. Real implementation should use:
- ///
`{base_path}/{streams_path}/{stream_id}/{topics_path}/{topic_id}/{partitions_path}/{partition_id}/{start_offset:0>20}.log`
#[must_use]
pub fn get_messages_path(
&self,
@@ -319,10 +348,6 @@ impl PartitionsConfig {
}
/// Constructs the file path for segment indexes.
- ///
- /// TODO: This is a stub waiting for completion of issue to move server
config
- /// to shared module. Real implementation should use:
- ///
`{base_path}/{streams_path}/{stream_id}/{topics_path}/{topic_id}/{partitions_path}/{partition_id}/{start_offset:0>20}.index`
#[must_use]
pub fn get_index_path(
&self,
@@ -336,43 +361,4 @@ impl PartitionsConfig {
self.get_partition_path(stream_id, topic_id, partition_id)
)
}
-
- #[must_use]
- pub fn get_offsets_path(
- &self,
- stream_id: usize,
- topic_id: usize,
- partition_id: usize,
- ) -> String {
- format!(
- "{}/offsets",
- self.get_partition_path(stream_id, topic_id, partition_id)
- )
- }
-
- #[must_use]
- pub fn get_consumer_offsets_path(
- &self,
- stream_id: usize,
- topic_id: usize,
- partition_id: usize,
- ) -> String {
- format!(
- "{}/consumers",
- self.get_offsets_path(stream_id, topic_id, partition_id)
- )
- }
-
- #[must_use]
- pub fn get_consumer_group_offsets_path(
- &self,
- stream_id: usize,
- topic_id: usize,
- partition_id: usize,
- ) -> String {
- format!(
- "{}/groups",
- self.get_offsets_path(stream_id, topic_id, partition_id)
- )
- }
}
diff --git a/core/server/config.toml b/core/server/config.toml
index f420aeeb3..d5046c509 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -622,6 +622,19 @@ key_file = ""
# use hostnames in the roster if the certificates only have DNS SANs.
ca_file = ""
+# Shard-0 coordinator placement.
+[cluster.coordinator]
+# When the server runs more than one shard, exclude shard 0 from replica
+# placement. Shard 0 already hosts the coordinator, the metadata writer, and
+# both listeners; replica connections are long-lived steady flows, so they are
+# offloaded to peer shards by default.
+skip_shard_zero_for_replicas = true
+
+# When the server runs more than one shard, exclude shard 0 from client
+# placement. Off by default: client connections are short-lived and benefit
+# from shard-0 parallelism more than replicas do.
+skip_shard_zero_for_clients = false
+
# Full roster of cluster members. Byte-identical on every node. The running
# node's identity is resolved at launch from the '--replica-id <N>' CLI
# flag, which selects the entry in this list that describes the current
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 01ef2d1fa..f3cf877ce 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -989,7 +989,12 @@ async fn shard_main(
let poll_interval =
config.system.sharding.shutdown_poll_interval.get_duration();
let shutdown_flag_for_handoff = Arc::clone(&shutdown_flag);
- spawn_shutdown_watchdog(Rc::clone(&bus), shutdown_flag, drain_timeout,
poll_interval);
+ let mut shutdown_watchdog = Some(spawn_shutdown_watchdog(
+ Rc::clone(&bus),
+ shutdown_flag,
+ drain_timeout,
+ poll_interval,
+ ));
// Metadata bootstrap is single-writer: shard 0 owns the WAL and the
// only `WriteHandle`-bearing `MuxStateMachine`. Peer shards receive
@@ -1194,20 +1199,17 @@ async fn shard_main(
);
// Re-check the cross-thread shutdown flag here, *before* spawning the
- // message pump. A sibling shard may have failed in the window between
- // the metadata broadcast and this point; gating before spawn keeps the
- // bus' `background_tasks` vec empty on the shutdown path. Spawn-then-
- // check would leave `bus.track_background(pump_handle)` registering a
- // `JoinHandle` that only `bus.shutdown()` drains, but the watchdog
- // driving `bus.shutdown()` is `.detach()`'d (see TODO at
- // `spawn_shutdown_watchdog`) and may not be scheduled before this
- // function returns `Ok(())` and the compio runtime drops, cancelling
- // the pump mid-`write_vectored_all`.
- //
- // Without this gate shard 0 would also still open TCP/QUIC/WS
+ // message pump: it keeps the bus' `background_tasks` vec empty on the
+ // shutdown path, and shard 0 would otherwise still open TCP/QUIC/WS
// listeners for a server that is already tearing down, briefly
// accepting connections that immediately get torn by the watchdog.
+ //
+ // The flag is set, so the watchdog is (about to be) driving
+ // `bus.shutdown()`; await it so the runtime does not drop mid-drain.
if shutdown_flag_for_handoff.load(Ordering::Relaxed) {
+ if let Some(watchdog) = shutdown_watchdog.take() {
+ let _ = watchdog.await;
+ }
return Ok(());
}
@@ -1420,6 +1422,12 @@ async fn shard_main(
// The bind failure is the primary fault; the drain verdict only
// matters for the log it emits.
let _ = await_pump_drain(pump_handle.take(), config,
shard_id).await;
+ // Neither the flag nor the bus token has fired yet on this path,
+ // so the watchdog is still idle-looping; awaiting it would hang.
+ // Detach and let `run_shard_thread`'s unwind flip the flag.
+ if let Some(watchdog) = shutdown_watchdog.take() {
+ watchdog.detach();
+ }
return Err(error);
}
@@ -1446,7 +1454,15 @@ async fn shard_main(
let _ = tx.try_send(());
}
- await_pump_drain(pump_handle.take(), config, shard_id).await?;
+ // Await the watchdog even when the drain verdict is an error: the token
+ // has fired, so it either stands down within one poll interval or is
+ // mid-`bus.shutdown()`, and dropping it there truncates in-flight
+ // `ClientForwardFailed` replies.
+ let pump_verdict = await_pump_drain(pump_handle.take(), config,
shard_id).await;
+ if let Some(watchdog) = shutdown_watchdog.take() {
+ let _ = watchdog.await;
+ }
+ pump_verdict?;
info!(shard = shard_id, "server shard exited cleanly");
Ok(())
@@ -1662,16 +1678,27 @@ async fn await_bootstrap_complete(
/// is the only Send signal we have; the bus' shutdown machinery is
/// `!Send` (`Rc<Cell<bool>>` + per-shard `async_channel`), so it must be
/// triggered from within the runtime that owns the bus.
+///
+/// The caller owns the returned handle and must await it on the exit paths
+/// where shutdown is in progress (flag set or bus token triggered):
+/// dropping it there cancels the watchdog mid-`bus.shutdown()`, truncating
+/// in-flight `ClientForwardFailed` replies (terminal per `SendError` docs).
+/// It cannot go through `bus.track_background` instead: the watchdog itself
+/// drives `bus.shutdown()`, and the bg-drain loop in `shutdown()` would
+/// re-enter awaiting the watchdog's own pending shutdown call
+/// (self-deadlock). The await is bounded: once the token fires the loop
+/// stands down within one poll interval, and the shutdown call itself is
+/// capped by `drain_timeout`.
#[allow(clippy::needless_pass_by_value)]
fn spawn_shutdown_watchdog(
bus: Rc<IggyMessageBus>,
shutdown_flag: Arc<AtomicBool>,
drain_timeout: Duration,
poll_interval: Duration,
-) {
+) -> compio::runtime::JoinHandle<()> {
let bus_for_task = Rc::clone(&bus);
let bus_token = bus.token();
- let watchdog = compio::runtime::spawn(async move {
+ compio::runtime::spawn(async move {
loop {
if shutdown_flag.load(Ordering::Relaxed) {
break;
@@ -1684,19 +1711,7 @@ fn spawn_shutdown_watchdog(
compio::time::sleep(poll_interval).await;
}
let _ = bus_for_task.shutdown(drain_timeout).await;
- });
- // TODO(hubcio): `.detach()` races bus shutdown: when `bus.token()` is
- // triggered, `shard_main` returns and the runtime drops the watchdog
- // mid-`bus.shutdown()`, truncating in-flight `ClientForwardFailed`
- // replies (terminal per `SendError` docs). Cannot use
- // `bus.track_background(watchdog)` here because the watchdog itself
- // drives `bus.shutdown()`, and the bg-drain loop in `shutdown()`
- // would re-enter awaiting the watchdog's own pending shutdown call
- // (self-deadlock). Fix: extract a `core/task_registry` crate mirroring
- // `core/server`'s task-tracking mechanism, share it between the bus
- // and server so background tasks can be reaped without coupling
- // to the bus shutdown order.
- watchdog.detach();
+ })
}
/// Copy the configured cluster roster plus this node's own client ports into
@@ -1790,6 +1805,11 @@ async fn build_shard_for_thread(
segment_size:
IggyByteSize::from(iggy_common::DEFAULT_SEGMENT_SIZE),
preallocate_segments: iggy_common::DEFAULT_PREALLOCATE_SEGMENTS,
encryptor,
+ path_layout: partitions::PartitionPathLayout {
+ streams_root: config.system.get_streams_path(),
+ topics_dir: config.system.topic.path.clone(),
+ partitions_dir: config.system.partition.path.clone(),
+ },
},
owned_partitions_capacity,
);
@@ -2007,7 +2027,10 @@ async fn build_shard_for_thread(
shard::ReplicaTopology::new(topology.self_replica_id,
topology.replica_count),
Rc::clone(&bus),
),
- CoordinatorConfig::default(),
+ CoordinatorConfig {
+ skip_shard_zero_for_replicas:
config.cluster.coordinator.skip_shard_zero_for_replicas,
+ skip_shard_zero_for_clients:
config.cluster.coordinator.skip_shard_zero_for_clients,
+ },
metrics,
)
.build()
diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs
index 9251bf925..d3821b15b 100644
--- a/core/server/src/dispatch.rs
+++ b/core/server/src/dispatch.rs
@@ -3842,7 +3842,7 @@ mod tests {
use metadata::stm::stream::Streams;
use metadata::stm::user::Users;
use metadata::{IggyMetadata, MuxStateMachine};
- use partitions::{IggyPartitions, PartitionsConfig};
+ use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig};
use server_common::iobuf::Frozen;
use server_common::sharding::ShardId;
use server_common::{MESSAGE_ALIGN, Message, MessageBag};
@@ -3999,6 +3999,7 @@ mod tests {
segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
preallocate_segments: false,
encryptor: None,
+ path_layout: PartitionPathLayout::default(),
},
);
TestShard::without_inbox(
@@ -4165,6 +4166,7 @@ mod tests {
segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
preallocate_segments: false,
encryptor: None,
+ path_layout: PartitionPathLayout::default(),
},
);
let shard = Rc::new(TestShard::without_inbox(
@@ -4286,6 +4288,7 @@ mod tests {
segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
preallocate_segments: false,
encryptor: None,
+ path_layout: PartitionPathLayout::default(),
},
);
let shard = Rc::new(TestShard::without_inbox(
@@ -4410,6 +4413,7 @@ mod tests {
segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
preallocate_segments: false,
encryptor: None,
+ path_layout: PartitionPathLayout::default(),
},
);
let shard = Rc::new(TestShard::without_inbox(
@@ -4474,6 +4478,7 @@ mod tests {
segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
preallocate_segments: false,
encryptor: None,
+ path_layout: PartitionPathLayout::default(),
},
);
// Real sender ring so the staged deny is observable: the test holds
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index f6a8e63e3..e2547efc7 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -71,7 +71,9 @@ fn main() -> Result<(), ServerError> {
let bootstrap_result: Result<ServerConfig, ServerError> =
bootstrap_runtime.block_on(async {
let config = load_config().await?;
prepare_runtime_dirs(&config, &mut logging, args.fresh).await?;
-
server_common::MemoryPool::init_pool(&config.system.memory_pool.into_other());
+ let memory_pool_settings =
+
server_common::MemoryPoolSettings::from(&config.system.memory_pool);
+ server_common::MemoryPool::init_pool(&memory_pool_settings);
Ok(config)
});
diff --git a/core/server/src/partition_reconciler.rs
b/core/server/src/partition_reconciler.rs
index b456a30b8..fbf99df6b 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -1183,8 +1183,8 @@ mod tests {
PurgeTopicRequest,
};
use iggy_binary_protocol::{
- Command, Operation, PrepareHeader, ReplyHeader, RoutedRequestHeader,
WireIdentifier,
- WireOptions,
+ Command, Operation, PrepareHeader, RepairRangeReplyHeader, ReplyHeader,
+ RequestPreparesHeader, RoutedRequestHeader, WireIdentifier,
WireOptions,
};
use message_bus::IggyMessageBus;
use metadata::IggyMetadata;
@@ -1193,7 +1193,7 @@ mod tests {
use metadata::stm::StateMachine;
use metadata::stm::stream::Streams;
use metadata::stm::user::Users;
- use partitions::{IggyPartitions, PartitionsConfig};
+ use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig,
RepairSession};
use server_common::sharding::{IggyNamespace, ShardId};
use server_common::{Message, MessageBag};
use shard::shards_table::{PapayaShardsTable, ShardsTable,
calculate_shard_assignment};
@@ -1318,6 +1318,54 @@ mod tests {
MessageBag::Request(msg)
}
+ /// Build a partition-plane `RepairRangeReply` as the serving peer would
+ /// send it. Only the fields the receive path reads are stamped: routing
+ /// (`group`), session (`nonce`), and the verdict (`command`, `op`).
+ fn build_repair_range_reply(
+ namespace: IggyNamespace,
+ command: Command,
+ nonce: u128,
+ op: u64,
+ ) -> MessageBag {
+ let header_size = size_of::<RepairRangeReplyHeader>();
+ let mut msg = Message::<RepairRangeReplyHeader>::new(header_size);
+ let header =
bytemuck::checked::try_from_bytes_mut::<RepairRangeReplyHeader>(
+ &mut msg.as_mut_slice()[..header_size],
+ )
+ .expect("zeroed bytes form a valid RepairRangeReplyHeader");
+ header.command = command;
+ header.size = u32::try_from(header_size).expect("header size fits
u32");
+ header.nonce = nonce;
+ header.op = op;
+ header.group = namespace.inner();
+ MessageBag::RepairRangeReply(msg)
+ }
+
+ /// Build a partition-plane `RequestPrepares` as a rejoining peer would
+ /// send it. `replica` is the requester the serve path replies to.
+ fn build_request_prepares(
+ namespace: IggyNamespace,
+ replica: u8,
+ nonce: u128,
+ from_op: u64,
+ to_op: u64,
+ ) -> MessageBag {
+ let header_size = size_of::<RequestPreparesHeader>();
+ let mut msg = Message::<RequestPreparesHeader>::new(header_size);
+ let header =
bytemuck::checked::try_from_bytes_mut::<RequestPreparesHeader>(
+ &mut msg.as_mut_slice()[..header_size],
+ )
+ .expect("zeroed bytes form a valid RequestPreparesHeader");
+ header.command = Command::RequestPrepares;
+ header.size = u32::try_from(header_size).expect("header size fits
u32");
+ header.replica = replica;
+ header.nonce = nonce;
+ header.from_op = from_op;
+ header.to_op = to_op;
+ header.group = namespace.inner();
+ MessageBag::RequestPrepares(msg)
+ }
+
fn assignment(partition_id: u32, consensus_group_id: u64) ->
CreatedPartitionAssignment {
CreatedPartitionAssignment {
partition_id,
@@ -1468,6 +1516,7 @@ mod tests {
segment_size:
iggy_common::IggyByteSize::from(iggy_common::DEFAULT_SEGMENT_SIZE),
preallocate_segments: false,
encryptor: None,
+ path_layout: PartitionPathLayout::default(),
},
);
let shards_table = PapayaShardsTable::new();
@@ -2285,6 +2334,184 @@ mod tests {
);
}
+ /// Receive half of the purge gate in `on_repair_range_reply`: while a
+ /// committed purge has not applied locally, a repair verdict must be
+ /// deferred wholesale -- installing the peer's floor against pre-purge
+ /// segments silently loses the post-purge batches (offsets restarting at
+ /// 0 flush-skip below the stale durable line).
+ #[compio::test]
+ async fn repair_completion_defers_until_committed_purge_applies() {
+ const NONCE: u128 = 7;
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-repair-gate");
+ seed_topic(&mux, 2, 0, "topic-repair-gate", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ reconcile_pass(&ctx).await;
+
+ let ns = IggyNamespace::new(0, 0, 0);
+ shard
+ .plane
+ .partitions()
+ .get_mut_by_ns(&ns)
+ .expect("partition is materialised")
+ .repair = Some(RepairSession {
+ nonce: NONCE,
+ to_op: 5,
+ floor: None,
+ peer: 1,
+ first_batch_offset: None,
+ idle_ticks: 0,
+ });
+
+ // Committed purge: generation 1 > applied 0.
+ let purge = PurgeTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ };
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .update(build_prepare(3, Operation::PurgeTopic, &purge))
+ .expect("PurgeTopic apply succeeds");
+
+ let deferred_before = shard
+ .metrics()
+ .partition_repair_serves_deferred_purge_value();
+ shard
+ .on_message(build_repair_range_reply(
+ ns,
+ Command::RangeEvicted,
+ NONCE,
+ 4,
+ ))
+ .await;
+ let session = shard
+ .plane
+ .partitions()
+ .get_mut_by_ns(&ns)
+ .expect("partition survives the deferral")
+ .repair
+ .expect("deferral must leave the repair session armed");
+ assert_eq!(
+ session.floor, None,
+ "a deferred RangeEvicted must not install the peer's floor"
+ );
+ assert_eq!(
+ shard
+ .metrics()
+ .partition_repair_serves_deferred_purge_value(),
+ deferred_before + 1,
+ "the deferral must be visible on the purge-deferred counter"
+ );
+
+ // Apply the purge; the same frame now lands.
+ let partitions_config = shard.plane.partitions().config().clone();
+ shard
+ .plane
+ .partitions()
+ .get_mut_by_ns(&ns)
+ .expect("purged partition is materialised")
+ .purge(&partitions_config, 1)
+ .await
+ .expect("apply staged purge");
+ shard
+ .on_message(build_repair_range_reply(
+ ns,
+ Command::RangeEvicted,
+ NONCE,
+ 4,
+ ))
+ .await;
+ let session = shard
+ .plane
+ .partitions()
+ .get_mut_by_ns(&ns)
+ .expect("partition survives the retry")
+ .repair
+ .expect("RangeEvicted records the floor but keeps the session");
+ assert_eq!(
+ session.floor,
+ Some(3),
+ "after the purge applies, the retried frame must install the floor"
+ );
+ assert_eq!(
+ shard
+ .metrics()
+ .partition_repair_serves_deferred_purge_value(),
+ deferred_before + 1,
+ "the retried frame must pass the gate without another deferral"
+ );
+ }
+
+ /// Serve half of the purge gate in `on_request_prepares`: while a
+ /// committed purge has not applied locally, the journal still holds
+ /// pre-purge entries with no floor to fence them, so serving a rejoiner
+ /// must be deferred (no reply; the requester's stall retry re-asks).
+ #[compio::test]
+ async fn repair_serve_defers_until_committed_purge_applies() {
+ const NONCE: u128 = 11;
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-serve-gate");
+ seed_topic(&mux, 2, 0, "topic-serve-gate", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ reconcile_pass(&ctx).await;
+
+ let ns = IggyNamespace::new(0, 0, 0);
+ let purge = PurgeTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ };
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .update(build_prepare(3, Operation::PurgeTopic, &purge))
+ .expect("PurgeTopic apply succeeds");
+
+ let deferred_before = shard
+ .metrics()
+ .partition_repair_serves_deferred_purge_value();
+ shard
+ .on_message(build_request_prepares(ns, 1, NONCE, 1, 5))
+ .await;
+ assert_eq!(
+ shard
+ .metrics()
+ .partition_repair_serves_deferred_purge_value(),
+ deferred_before + 1,
+ "an unapplied purge must defer the serve"
+ );
+
+ let partitions_config = shard.plane.partitions().config().clone();
+ shard
+ .plane
+ .partitions()
+ .get_mut_by_ns(&ns)
+ .expect("purged partition is materialised")
+ .purge(&partitions_config, 1)
+ .await
+ .expect("apply staged purge");
+ shard
+ .on_message(build_request_prepares(ns, 1, NONCE, 1, 5))
+ .await;
+ assert_eq!(
+ shard
+ .metrics()
+ .partition_repair_serves_deferred_purge_value(),
+ deferred_before + 1,
+ "once the purge applies, the retried request must be served, not
deferred"
+ );
+ }
+
/// Permanent-tombstone-wedge regression: a teardown whose disk delete
/// fails sets the tombstone and removes the `shards_table` row but never
/// enqueues `ConfirmRemove`, so the tombstone never lifts. If the same
diff --git a/core/server/src/responses.rs b/core/server/src/responses.rs
index 6b7fc4ec5..5f9073ed5 100644
--- a/core/server/src/responses.rs
+++ b/core/server/src/responses.rs
@@ -703,6 +703,51 @@ where
}
}
+/// `(streams, topics, partitions, segments, message bytes, messages)` for the
+/// whole node, from committed metadata plus the shared stats registry.
+///
+/// Segments are summed PER PARTITION through the same floor the detail
+/// responses apply (see [`partition_response`]), not from the stream's
rolled-up
+/// counter: that counter only advances once a partition materialises, which
+/// trails its commit by a reconciler pass. Summing it made `[stats]` report
+/// fewer segments than `get_topic` did for the same partitions, and let the
+/// total climb between two reads with no write in between.
+fn aggregate_stats_totals(
+ streams: &metadata::stm::stream::StreamsInner,
+) -> Result<(u32, u32, u32, u32, u64, u64), IggyError> {
+ let mut topics_count = 0u32;
+ let mut partitions_count = 0u32;
+ let mut segments_count = 0u32;
+ let mut messages_size_bytes = 0u64;
+ let mut messages_count = 0u64;
+ for (_, stream) in &streams.items {
+ topics_count =
topics_count.saturating_add(usize_to_u32(stream.topics.len())?);
+ messages_size_bytes =
+
messages_size_bytes.saturating_add(stream.stats.size_bytes_inconsistent());
+ messages_count =
messages_count.saturating_add(stream.stats.messages_count_inconsistent());
+ for (_, topic) in &stream.topics {
+ partitions_count =
+
partitions_count.saturating_add(usize_to_u32(topic.partitions.len())?);
+ for partition in &topic.partitions {
+ segments_count =
segments_count.saturating_add(partition_segments_count(
+ streams,
+ stream.id,
+ topic.id,
+ partition.id,
+ ));
+ }
+ }
+ }
+ Ok((
+ usize_to_u32(streams.items.len())?,
+ topics_count,
+ partitions_count,
+ segments_count,
+ messages_size_bytes,
+ messages_count,
+ ))
+}
+
fn build_stats_response<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
clients_count: u32,
@@ -726,36 +771,7 @@ where
.metadata()
.mux_stm
.streams()
- .read(|streams| -> Result<_, IggyError> {
- let mut topics_count = 0u32;
- let mut partitions_count = 0u32;
- let mut segments_count = 0u32;
- let mut messages_size_bytes = 0u64;
- let mut messages_count = 0u64;
- for (_, stream) in &streams.items {
- topics_count =
topics_count.saturating_add(usize_to_u32(stream.topics.len())?);
- messages_size_bytes =
-
messages_size_bytes.saturating_add(stream.stats.size_bytes_inconsistent());
- messages_count =
-
messages_count.saturating_add(stream.stats.messages_count_inconsistent());
- // Segment counts roll up from the partition plane through
- // the shared stats registry (partition -> topic -> stream).
- segments_count =
-
segments_count.saturating_add(stream.stats.segments_count_inconsistent());
- for (_, topic) in &stream.topics {
- partitions_count =
-
partitions_count.saturating_add(usize_to_u32(topic.partitions.len())?);
- }
- }
- Ok((
- usize_to_u32(streams.items.len())?,
- topics_count,
- partitions_count,
- segments_count,
- messages_size_bytes,
- messages_count,
- ))
- })?;
+ .read(aggregate_stats_totals)?;
let consumer_groups_count = usize_to_u32(
shard
.plane
@@ -1325,6 +1341,29 @@ fn topic_header(topic: &metadata::stm::stream::Topic) ->
Result<StreamTopicHeade
})
}
+/// Segments a committed partition reports before its storage exists.
+/// [`partition_response`] carries the reasoning.
+const MATERIALIZED_SEGMENTS_FLOOR: u32 = 1;
+
+/// Segments one committed partition reports. The single source for every
+/// client-facing segment count, so the `[stats]` total and the per-partition
+/// detail cannot disagree about the same partition.
+fn partition_segments_count(
+ streams: &metadata::stm::stream::StreamsInner,
+ stream_id: usize,
+ topic_id: usize,
+ partition_id: usize,
+) -> u32 {
+ streams
+ .stats_registry
+ .partition_get(stream_id, topic_id, partition_id)
+ .map_or(MATERIALIZED_SEGMENTS_FLOOR, |stats| {
+ stats
+ .segments_count_inconsistent()
+ .max(MATERIALIZED_SEGMENTS_FLOOR)
+ })
+}
+
fn partition_response(
streams: &metadata::stm::stream::StreamsInner,
stream_id: usize,
@@ -1353,19 +1392,17 @@ fn partition_response(
let stats = streams
.stats_registry
.partition_get(stream_id, topic_id, partition.id);
- let (segments_count, current_offset, size_bytes, messages_count) =
- stats.map_or((1, 0, 0, 0), |stats| {
- (
- stats.segments_count_inconsistent().max(1),
- stats.current_offset(),
- stats.size_bytes_inconsistent(),
- stats.messages_count_inconsistent(),
- )
- });
+ let (current_offset, size_bytes, messages_count) = stats.map_or((0, 0, 0),
|stats| {
+ (
+ stats.current_offset(),
+ stats.size_bytes_inconsistent(),
+ stats.messages_count_inconsistent(),
+ )
+ });
Ok(PartitionResponse {
id: usize_to_u32(partition.id)?,
created_at: partition.created_at.as_micros(),
- segments_count,
+ segments_count: partition_segments_count(streams, stream_id, topic_id,
partition.id),
current_offset,
size_bytes,
messages_count,
@@ -1857,6 +1894,65 @@ mod tests {
assert_eq!(live.size_bytes, 64);
}
+ #[test]
+ fn stats_totals_count_every_committed_partition_before_it_materialises() {
+ use iggy_common::{StreamStats, TopicStats};
+ use metadata::stm::stream::{Partition, Stream, StreamsInner, Topic};
+ use std::sync::atomic::AtomicUsize;
+
+ let created_at = IggyTimestamp::from(1u64);
+ let mut streams = StreamsInner::new();
+ let mut stream = Stream::new(Arc::from("stream"), created_at);
+ let topic_stats = Arc::new(TopicStats::new(stream.stats.clone()));
+ stream.topics.insert(Topic {
+ id: 0,
+ name: Arc::from("topic"),
+ created_at,
+ message_expiry: iggy_common::IggyExpiry::NeverExpire,
+ compression_algorithm: iggy_common::CompressionAlgorithm::None,
+ max_topic_size: iggy_common::MaxTopicSize::Unlimited,
+ options: iggy_common::ResourceOptions::default(),
+ stats: topic_stats.clone(),
+ partitions: vec![
+ Partition::new(0, 1, created_at, 0),
+ Partition::new(1, 1, created_at, 0),
+ ],
+ round_robin_counter: Arc::new(AtomicUsize::new(0)),
+ consumer_groups: ahash::AHashMap::default(),
+ consumer_group_index: ahash::AHashMap::default(),
+ next_consumer_group_id: 0,
+ });
+ streams.items.insert(stream);
+
+ // Only partition 0 has materialised. Counting the stream's rolled-up
+ // counter reported 1 here, so a caller polling `[stats]` twice saw the
+ // total climb to 2 with no write in between (and `get_topic` already
+ // reported 2 for the same partitions).
+ let stats = streams.stats_registry.partition(0, 0, 0, topic_stats);
+ stats.increment_segments_count(1);
+
+ let (_, _, partitions, segments, _, _) =
+ aggregate_stats_totals(&streams).expect("totals aggregate");
+ assert_eq!(partitions, 2);
+ assert_eq!(
+ segments, 2,
+ "an unmaterialised partition must contribute the same floor the
detail response reports"
+ );
+
+ // Materialising the second partition changes nothing: the total was
+ // already the steady-state answer.
+ let late = streams.stats_registry.partition(
+ 0,
+ 0,
+ 1,
+ Arc::new(TopicStats::new(Arc::new(StreamStats::default()))),
+ );
+ late.increment_segments_count(1);
+ let (_, _, _, segments_after, _, _) =
+ aggregate_stats_totals(&streams).expect("totals aggregate");
+ assert_eq!(segments_after, 2);
+ }
+
#[test]
fn topic_header_echoes_stored_size_and_expiry_verbatim() {
use iggy_common::{
diff --git a/core/server_common/src/buffer.rs b/core/server_common/src/buffer.rs
index aa84101b7..79b018716 100644
--- a/core/server_common/src/buffer.rs
+++ b/core/server_common/src/buffer.rs
@@ -434,7 +434,7 @@ mod miri_tests {
#[cfg(not(miri))]
mod split_to {
use super::*;
- use crate::memory_pool::{MemoryPool, MemoryPoolConfigOther};
+ use crate::memory_pool::{MemoryPool, MemoryPoolSettings};
use iggy_common::IggyByteSize;
use serial_test::serial;
use std::str::FromStr;
@@ -444,7 +444,7 @@ mod miri_tests {
fn init_pool_for_split_to_tests() {
MIRI_POOL_INIT.call_once(|| {
- let config = MemoryPoolConfigOther {
+ let config = MemoryPoolSettings {
enabled: true,
size: IggyByteSize::from_str("64MiB").unwrap(),
bucket_capacity: 16,
diff --git a/core/server_common/src/executor.rs
b/core/server_common/src/executor.rs
index ed8ea3637..8fb8cc842 100644
--- a/core/server_common/src/executor.rs
+++ b/core/server_common/src/executor.rs
@@ -20,6 +20,13 @@ use compio::runtime::Runtime;
const DEFAULT_SHARD_RUNTIME_CAPACITY: u32 = 4096;
const SHARD_RUNTIME_CAPACITY_ENV: &str = "IGGY_SHARD_RUNTIME_CAPACITY";
+/// How many tasks the runtime polls between driver (io_uring) sweeps. The
+/// default suits the stock task population; a deployment expecting far more
+/// connected clients (each connection is roughly one task) can raise it via
+/// [`SHARD_EVENT_INTERVAL_ENV`] to amortise driver sweeps across more work.
+const DEFAULT_SHARD_EVENT_INTERVAL: usize = 128;
+const SHARD_EVENT_INTERVAL_ENV: &str = "IGGY_SHARD_EVENT_INTERVAL";
+
/// Resolves the per-shard io_uring SQ/CQ capacity from
`IGGY_SHARD_RUNTIME_CAPACITY`,
/// falling back to [`DEFAULT_SHARD_RUNTIME_CAPACITY`] when the var is missing
or
/// fails to parse as `u32`.
@@ -30,12 +37,26 @@ fn shard_capacity_from_env() -> u32 {
.unwrap_or(DEFAULT_SHARD_RUNTIME_CAPACITY)
}
+/// Resolves the runtime event interval from `IGGY_SHARD_EVENT_INTERVAL`,
+/// falling back to [`DEFAULT_SHARD_EVENT_INTERVAL`] when the var is missing,
+/// fails to parse, or is zero (compio treats the interval as a divisor).
+fn shard_event_interval_from_env() -> usize {
+ std::env::var(SHARD_EVENT_INTERVAL_ENV)
+ .ok()
+ .and_then(|v| v.parse::<usize>().ok())
+ .filter(|&interval| interval > 0)
+ .unwrap_or(DEFAULT_SHARD_EVENT_INTERVAL)
+}
+
/// Creates a compio runtime for a shard thread, with shard-specific
`io_uring` flags.
///
/// The per-ring SQ/CQ capacity defaults to `4096` and can be overridden via
the
/// `IGGY_SHARD_RUNTIME_CAPACITY` env var, which the multi-node integration
/// harness sets to `256` so N nodes * M shards fit under an 8 MiB
-/// `RLIMIT_MEMLOCK` budget without `ENOMEM` at ring setup.
+/// `RLIMIT_MEMLOCK` budget without `ENOMEM` at ring setup. The runtime event
+/// interval defaults to `128` and can be overridden via
+/// `IGGY_SHARD_EVENT_INTERVAL` for deployments whose task count (roughly the
+/// connected-client count) makes a different poll-to-sweep ratio pay off.
///
/// # Errors
///
@@ -47,9 +68,6 @@ fn shard_capacity_from_env() -> u32 {
/// Falling back to default flags would silently degrade shard performance -
/// do not add a retry with reduced flags here.
pub fn create_shard_executor() -> Result<Runtime, std::io::Error> {
- // TODO: The event interval tick, could be configured based on the fact
- // How many clients we expect to have connected.
- // This roughly estimates the number of tasks we will create.
let mut proactor = compio::driver::ProactorBuilder::new();
proactor
@@ -57,45 +75,56 @@ pub fn create_shard_executor() -> Result<Runtime,
std::io::Error> {
.coop_taskrun(true)
.taskrun_flag(true);
- // FIXME(hubcio): Only set thread_pool_limit(0) on non-macOS platforms
- // This causes a freeze on macOS with compio fs operations
- // see https://github.com/compio-rs/compio/issues/446
+ // Permanent divergence, not a workaround to revisit: macOS runs compio's
+ // polling driver, which routes fs operations through the blocking pool, so
+ // a zero limit cannot work there by design. Upstream closed
+ // https://github.com/compio-rs/compio/issues/446 by making blocking-pool
+ // dispatch with no workers panic ("the thread pool is needed but no worker
+ // thread is running", compio-driver asyncify.rs) instead of freeze.
+ // io_uring targets keep the zero limit: no blocking pool exists on shard
+ // threads, which `core/partitions` messages_writer relies on to justify
+ // running fallocate inline (`spawn_blocking` would hit that same panic).
#[cfg(not(all(target_os = "macos", target_arch = "aarch64")))]
proactor.thread_pool_limit(0);
compio::runtime::RuntimeBuilder::new()
.with_proactor(proactor.to_owned())
- .event_interval(128)
+ .event_interval(shard_event_interval_from_env())
.build()
}
#[cfg(test)]
mod tests {
use super::{
- DEFAULT_SHARD_RUNTIME_CAPACITY, SHARD_RUNTIME_CAPACITY_ENV,
shard_capacity_from_env,
+ DEFAULT_SHARD_EVENT_INTERVAL, DEFAULT_SHARD_RUNTIME_CAPACITY,
SHARD_EVENT_INTERVAL_ENV,
+ SHARD_RUNTIME_CAPACITY_ENV, shard_capacity_from_env,
shard_event_interval_from_env,
};
use serial_test::serial;
- fn with_capacity_env<R>(value: Option<&str>, f: impl FnOnce() -> R) -> R {
+ fn with_env<R>(name: &str, value: Option<&str>, f: impl FnOnce() -> R) ->
R {
// SAFETY: tests in this module are #[serial], so no other thread races
// on the process-wide environment while the guard is active.
- let prev = std::env::var(SHARD_RUNTIME_CAPACITY_ENV).ok();
+ let prev = std::env::var(name).ok();
unsafe {
match value {
- Some(v) => std::env::set_var(SHARD_RUNTIME_CAPACITY_ENV, v),
- None => std::env::remove_var(SHARD_RUNTIME_CAPACITY_ENV),
+ Some(v) => std::env::set_var(name, v),
+ None => std::env::remove_var(name),
}
}
let out = f();
unsafe {
match prev {
- Some(v) => std::env::set_var(SHARD_RUNTIME_CAPACITY_ENV, v),
- None => std::env::remove_var(SHARD_RUNTIME_CAPACITY_ENV),
+ Some(v) => std::env::set_var(name, v),
+ None => std::env::remove_var(name),
}
}
out
}
+ fn with_capacity_env<R>(value: Option<&str>, f: impl FnOnce() -> R) -> R {
+ with_env(SHARD_RUNTIME_CAPACITY_ENV, value, f)
+ }
+
#[test]
#[serial]
fn shard_capacity_from_env_uses_parsed_value() {
@@ -127,4 +156,45 @@ mod tests {
assert_eq!(shard_capacity_from_env(),
DEFAULT_SHARD_RUNTIME_CAPACITY);
});
}
+
+ #[test]
+ #[serial]
+ fn shard_event_interval_from_env_uses_parsed_value() {
+ with_env(SHARD_EVENT_INTERVAL_ENV, Some("512"), || {
+ assert_eq!(shard_event_interval_from_env(), 512);
+ });
+ }
+
+ #[test]
+ #[serial]
+ fn shard_event_interval_from_env_falls_back_when_unset() {
+ with_env(SHARD_EVENT_INTERVAL_ENV, None, || {
+ assert_eq!(
+ shard_event_interval_from_env(),
+ DEFAULT_SHARD_EVENT_INTERVAL
+ );
+ });
+ }
+
+ #[test]
+ #[serial]
+ fn shard_event_interval_from_env_falls_back_on_unparsable() {
+ with_env(SHARD_EVENT_INTERVAL_ENV, Some("not-a-number"), || {
+ assert_eq!(
+ shard_event_interval_from_env(),
+ DEFAULT_SHARD_EVENT_INTERVAL
+ );
+ });
+ }
+
+ #[test]
+ #[serial]
+ fn shard_event_interval_from_env_falls_back_on_zero() {
+ with_env(SHARD_EVENT_INTERVAL_ENV, Some("0"), || {
+ assert_eq!(
+ shard_event_interval_from_env(),
+ DEFAULT_SHARD_EVENT_INTERVAL
+ );
+ });
+ }
}
diff --git a/core/server_common/src/lib.rs b/core/server_common/src/lib.rs
index c8fa01d42..f7b710658 100644
--- a/core/server_common/src/lib.rs
+++ b/core/server_common/src/lib.rs
@@ -39,7 +39,7 @@ pub use consensus_message::{
MutableBacking, RequestBacking, RequestBackingKind, ResponseBacking,
ResponseBackingKind,
};
pub use executor::create_shard_executor;
-pub use memory_pool::{MEMORY_POOL, MemoryPool, MemoryPoolConfigOther,
memory_pool};
+pub use memory_pool::{MEMORY_POOL, MemoryPool, MemoryPoolSettings,
memory_pool};
pub use segment_storage::{
IndexReader, IndexWriter, MessagesReader, MessagesWriter, SegmentStorage,
};
diff --git a/core/server_common/src/memory_pool.rs
b/core/server_common/src/memory_pool.rs
index b55c1c64a..a30ed4b2e 100644
--- a/core/server_common/src/memory_pool.rs
+++ b/core/server_common/src/memory_pool.rs
@@ -70,12 +70,11 @@ pub fn memory_pool() -> &'static MemoryPool {
.expect("Memory pool not initialized - MemoryPool::init_pool should be
called first")
}
-// TODO: Extract shared domain types (IggyByteSize, IggyDuration, etc.) into
an `iggy_types`
-// leaf crate so `iggy_common` can depend on `configs` directly. That lets us
delete this
-// duplicate and use `configs::server::MemoryPoolConfig` here instead.
-/// Configuration for the memory pool.
+/// Pool settings, converted from the serde-facing config type in
+/// `configs::server::MemoryPoolConfig` (which depends on this crate, so the
+/// domain type has to live here - same split as
[`crate::log::TelemetrySettings`]).
#[derive(Debug)]
-pub struct MemoryPoolConfigOther {
+pub struct MemoryPoolSettings {
/// Whether the pool is enabled.
pub enabled: bool,
/// Maximum size of the pool.
@@ -162,11 +161,11 @@ impl MemoryPool {
}
}
- /// Initialize the global pool from the given config.
- pub fn init_pool(config: &MemoryPoolConfigOther) {
- let is_enabled = config.enabled;
- let memory_limit = config.size.as_bytes_usize();
- let bucket_capacity = config.bucket_capacity as usize;
+ /// Initialize the global pool from the given settings.
+ pub fn init_pool(settings: &MemoryPoolSettings) {
+ let is_enabled = settings.enabled;
+ let memory_limit = settings.size.as_bytes_usize();
+ let bucket_capacity = settings.bucket_capacity as usize;
let _ =
MEMORY_POOL.get_or_init(|| MemoryPool::new(is_enabled,
memory_limit, bucket_capacity));
@@ -502,12 +501,12 @@ mod tests {
fn initialize_pool_for_tests() {
TEST_INIT.call_once(|| {
- let config = MemoryPoolConfigOther {
+ let settings = MemoryPoolSettings {
enabled: true,
size: IggyByteSize::from_str("4GiB").unwrap(),
bucket_capacity: 8192,
};
- MemoryPool::init_pool(&config);
+ MemoryPool::init_pool(&settings);
});
}
diff --git a/core/shard/src/config.rs b/core/shard/src/config.rs
index 27535c2e9..0406210b1 100644
--- a/core/shard/src/config.rs
+++ b/core/shard/src/config.rs
@@ -17,11 +17,10 @@
//! Runtime tunables for the shard-0 coordinator.
//!
-//! TODO: move this module into `core/configs` (as a `CoordinatorConfig`
-//! section nested under `ClusterConfig`) once downstream bootstrap
-//! wiring that constructs [`crate::coordinator::ShardZeroCoordinator`]
-//! from `ServerConfig` lands. Kept in-crate for now to avoid churning
-//! the configs crate ahead of that wiring.
+//! The serde-facing section lives in `core/configs` as
+//! `[cluster.coordinator]` (`ClusterCoordinatorConfig`); the server's
+//! bootstrap converts it into this domain type. The split exists because
+//! `configs` and `shard` share no dependency edge.
/// Tunables for [`crate::coordinator::ShardZeroCoordinator`].
#[derive(Debug, Clone)]
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 4e20691b0..a51dfa25f 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -4611,11 +4611,9 @@ where
// Defer the whole reply: the purge is one reconciler wake away and
// resets the line to `None`, and the stall retry re-asks, so the peer
// re-emits both `RangeEvicted` and `RepairDone` for the same window.
- // TODO(hubcio): no direct test drives this gate --
`on_repair_range_reply`
- // is only reachable through the real message bus and the shard crate
has
- // no fixture for it (the serve-side gate shares the gap). The loss
shape
- // is pinned at the partition level instead; a bus fixture would let
both
- // gates be exercised end to end.
+ // Pinned by `repair_completion_defers_until_committed_purge_applies`
+ // (server crate, partition_reconciler tests), driven through the pub
+ // `on_message` entry; the serve-side twin has its own pin there.
let committed_purge = self
.plane
.metadata()
diff --git a/core/simulator/src/bin/simulator-ui.rs
b/core/simulator/src/bin/simulator-ui.rs
index 3de3100b6..041a7c886 100644
--- a/core/simulator/src/bin/simulator-ui.rs
+++ b/core/simulator/src/bin/simulator-ui.rs
@@ -20,7 +20,7 @@ use iggy_binary_protocol::ReplyHeader;
use iggy_common::{IggyByteSize, PollingStrategy};
use partitions::{PollingArgs, PollingConsumer};
use server_common::sharding::IggyNamespace;
-use server_common::{MemoryPool, MemoryPoolConfigOther, Message};
+use server_common::{MemoryPool, MemoryPoolSettings, Message};
use simulator::Simulator;
use simulator::client::SimClient;
use simulator::packet::PacketSimulatorOptions;
@@ -41,7 +41,7 @@ fn step_until_reply(sim: &mut Simulator, max_ticks: u64) ->
Vec<Message<ReplyHea
fn main() {
// PooledBuffer::from (used by poll_messages) panics if the global pool is
uninitialized.
// Disabled pooling just falls through to the system allocator.
- MemoryPool::init_pool(&MemoryPoolConfigOther {
+ MemoryPool::init_pool(&MemoryPoolSettings {
enabled: false,
size: IggyByteSize::from(0u64),
bucket_capacity: 1,
diff --git a/core/simulator/src/bin/workload-fuzz.rs
b/core/simulator/src/bin/workload-fuzz.rs
index fc495f8af..11d8c865f 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -38,7 +38,7 @@
use clap::Parser;
use iggy_common::IggyByteSize;
use server_common::sharding::IggyNamespace;
-use server_common::{MemoryPool, MemoryPoolConfigOther};
+use server_common::{MemoryPool, MemoryPoolSettings};
use simulator::Simulator;
use simulator::client::SimClient;
use simulator::packet::PacketSimulatorOptions;
@@ -102,7 +102,7 @@ fn main() {
// poll_messages / reply paths panic without an initialized pool; disabled
// pooling falls through to the system allocator.
- MemoryPool::init_pool(&MemoryPoolConfigOther {
+ MemoryPool::init_pool(&MemoryPoolSettings {
enabled: false,
size: IggyByteSize::from(0u64),
bucket_capacity: 1,
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index fa5b3e316..17f24a620 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -994,7 +994,7 @@ mod tests {
/// heartbeat timeout and elect a new primary via view change.
#[test]
fn view_change_after_primary_crash() {
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1092,7 +1092,7 @@ mod tests {
/// Impossible before the superblock, since a rebuilt consensus starts at
view 0.
#[test]
fn
given_advanced_view_when_metadata_replica_restarts_should_recover_view_from_superblock()
{
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1202,7 +1202,7 @@ mod tests {
#[test]
fn
given_committed_metadata_when_solo_replica_restarts_should_recover_from_own_wal()
{
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1300,7 +1300,7 @@ mod tests {
#[test]
fn
given_registered_client_when_solo_replica_restarts_should_recover_session_from_own_wal()
{
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1367,7 +1367,7 @@ mod tests {
// split-brain on its restart. The positive control is
// `given_advanced_view_when_metadata_replica_restarts_...`, which
elects a new
// primary from the same crash with healthy superblocks.
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1456,7 +1456,7 @@ mod tests {
/// dedup if they need at-most-once-per-payload.
#[test]
fn failover_retry_re_executes_under_at_least_once() {
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1571,7 +1571,7 @@ mod tests {
/// `handle_commit_message_timeout` used to assert `commit_min ==
commit_max`.
#[test]
fn view_change_behind_backup_becomes_primary() {
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1639,7 +1639,7 @@ mod tests {
/// and workload) produces an identical reply-header sequence.
#[test]
fn workload_replay_is_deterministic() {
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1696,7 +1696,7 @@ mod tests {
};
use strum::{EnumCount, IntoEnumIterator};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1766,7 +1766,7 @@ mod tests {
};
use strum::{EnumCount, IntoEnumIterator};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1832,7 +1832,7 @@ mod tests {
options::{ActionWeights, WorkloadOptions},
};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1905,7 +1905,7 @@ mod tests {
oracle,
};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -1963,7 +1963,7 @@ mod tests {
options::{ActionWeights, WorkloadOptions},
oracle,
};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2029,7 +2029,7 @@ mod tests {
options::{ActionWeights, WorkloadOptions},
};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2119,7 +2119,7 @@ mod tests {
options::{ActionWeights, WorkloadOptions},
};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2212,7 +2212,7 @@ mod tests {
options::{ActionWeights, WorkloadOptions},
};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2394,7 +2394,7 @@ mod tests {
use consensus::MetadataHandle;
use shard::shards_table::{ShardsTable, calculate_shard_assignment};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2445,7 +2445,7 @@ mod tests {
fn peer_shard_resolves_namespace_via_shard0_read_handle() {
use iggy_binary_protocol::WireIdentifier;
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2499,7 +2499,7 @@ mod tests {
/// post-change send commits through the mesh.
#[test]
fn multi_shard_view_change_after_primary_crash() {
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2598,7 +2598,7 @@ mod tests {
/// Schedule hash for `seed` after stepping the consensus plane with no
/// client traffic, with the dispatch shell on or off.
fn consensus_schedule_hash(seed: u64, shell: bool) -> u64 {
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2650,7 +2650,7 @@ mod tests {
/// plus its metadata, log a client in against root, produce one message,
/// then poll. Returns the poll reply's raw bytes and the schedule hash.
fn shell_produce_poll(seed: u64) -> (Vec<u8>, u64) {
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2760,7 +2760,7 @@ mod tests {
use consensus::PartitionsHandle;
use std::panic::{AssertUnwindSafe, catch_unwind};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2859,7 +2859,7 @@ mod tests {
use consensus::PartitionsHandle;
use std::panic::{AssertUnwindSafe, catch_unwind};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -2957,7 +2957,7 @@ mod tests {
use consensus::MetadataHandle;
use journal::{Journal, JournalHandle};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -3069,7 +3069,7 @@ mod tests {
header.group == BLOCKED_NS.load(Ordering::Relaxed)
}
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -3177,7 +3177,7 @@ mod tests {
oracle,
};
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -3295,7 +3295,7 @@ mod view_change_data_loss_tests {
#[test]
fn
given_committed_op_missing_on_next_primary_when_primary_crashes_should_survive_view_change()
{
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
@@ -3393,7 +3393,7 @@ mod view_change_data_loss_tests {
#[test]
fn
given_a_register_inside_the_view_start_persist_when_the_pipeline_rebuilds_should_commit_once()
{
-
server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
enabled: false,
size: iggy_common::IggyByteSize::from(0u64),
bucket_capacity: 1,
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index 2764b96da..362f3a144 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -27,7 +27,7 @@ use metadata::stm::mux::WithFactory;
use metadata::stm::stream::{Streams, StreamsInner};
use metadata::stm::user::{Users, UsersInner};
use metadata::{IggyMetadata, apply_committed_prepare};
-use partitions::{IggyPartitions, PartitionsConfig};
+use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig};
use server::bootstrap::{ShellHandlers, ShellShardHandle, wire_shell_handlers};
use server_common::crypto;
use server_common::sharding::{METADATA_GROUP, ShardId};
@@ -287,6 +287,7 @@ pub fn new_shard(
segment_size: IggyByteSize::from(1024 * 1024 * 1024),
preallocate_segments: false,
encryptor: None,
+ path_layout: PartitionPathLayout::default(),
};
// Shard id is the NODE-LOCAL shard index, never the replica id: the
