diff --git a/core/configs/src/common/server.rs b/core/configs/src/common/server.rs index 315abeadc5..887cfe8ab3 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 a1874a6406..d71f86ae6b 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 1705e861ba..f752d32659 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 ee67e1097c..1fda606d12 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 4a366076dd..369db555a9 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 6d14d4b6b9..4649af8290 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 dcf4eb0b92..df15b43931 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>, + /// 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 f420aeeb30..d5046c5091 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 ' 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 01ef2d1fab..f3cf877cef 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>` + 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, shutdown_flag: Arc, 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 9251bf925b..d3821b15b3 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 f6a8e63e36..e2547efc7d 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 = 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 b456a30b81..fbf99df6bb 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::(); + let mut msg = Message::::new(header_size); + let header = bytemuck::checked::try_from_bytes_mut::( + &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::(); + let mut msg = Message::::new(header_size); + let header = bytemuck::checked::try_from_bytes_mut::( + &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_common/src/buffer.rs b/core/server_common/src/buffer.rs index aa84101b77..79b018716f 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 ed8ea36373..8fb8cc8422 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::().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 { - // 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 { .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(value: Option<&str>, f: impl FnOnce() -> R) -> R { + fn with_env(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(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 c8fa01d42b..f7b7106586 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 b55c1c64a8..a30ed4b2ec 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 27535c2e9a..0406210b14 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 4e20691b0e..a51dfa25fa 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 3de3100b65..041a7c8860 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 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, 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 2764b96da6..362f3a144f 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