refactor(server): consensus restore, fail-stop, linear cleanup batch

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,
         }
     }
 }

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(),
         }
     }
 

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(),
         }
     }
 }

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 {

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(),
         }
     }
 

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.

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)
-        )
-    }
 }

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