refactor(shard): drop journal generic, metrics scrape, poll cursor

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/journal/src/lib.rs

@@ -26,10 +26,7 @@ pub mod local_gate;
 pub mod prepare_journal;
 pub mod superblock;
 
-pub trait Journal<S>
-where
-    S: Storage,
-{
+pub trait Journal {
     type Header;
     type Entry;
     type HeaderRef<'a>: Deref<Target = Self::Header>
@@ -101,8 +98,7 @@ where
 }
 
 pub trait JournalHandle {
-    type Storage: Storage;
-    type Target: Journal<Self::Storage>;
+    type Target: Journal;
 
     fn handle(&self) -> &Self::Target;
 }
@@ -112,7 +108,6 @@ pub trait JournalHandle {
 /// the metadata WAL across a replica restart: the bytes and index survive the
 /// shard being dropped and rebuilt.
 impl<T: JournalHandle> JournalHandle for Rc<T> {
-    type Storage = T::Storage;
     type Target = T::Target;
 
     fn handle(&self) -> &Self::Target {

core/journal/src/prepare_journal.rs

@@ -741,7 +741,7 @@ impl PrepareJournal {
     clippy::cast_sign_loss,
     clippy::future_not_send
 )]
-impl Journal<FileStorage> for PrepareJournal {
+impl Journal for PrepareJournal {
     fn last_op(&self) -> Option<u64> {
         self.last_op.get()
     }
@@ -1162,7 +1162,6 @@ impl Journal<FileStorage> for PrepareJournal {
 }
 
 impl JournalHandle for PrepareJournal {
-    type Storage = FileStorage;
     type Target = Self;
 
     fn handle(&self) -> &Self::Target {

core/metadata/src/impls/metadata.rs

@@ -945,7 +945,7 @@ where
     B: MessageBus,
     SB: SuperblockStore,
     J: JournalHandle,
-    J::Target: Journal<J::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+    J::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
     M: StreamsFrontend
         + StateMachine<
             Input = Message<PrepareHeader>,
@@ -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) {
@@ -1317,7 +1313,7 @@ where
     B: MessageBus,
     P: Pipeline<Entry = PipelineEntry>,
     J: JournalHandle,
-    J::Target: Journal<J::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+    J::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
     M: StateMachine<Input = Message<PrepareHeader>>,
 {
     fn is_applicable<H>(&self, message: &<VsrConsensus<B, P> as Consensus>::Message<H>) -> bool
@@ -1460,7 +1456,7 @@ where
     B: MessageBus,
     SB: SuperblockStore,
     J: JournalHandle,
-    J::Target: Journal<J::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+    J::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
     M: StreamsFrontend
         + StateMachine<
             Input = Message<PrepareHeader>,
@@ -3094,7 +3090,7 @@ where
     P: Pipeline<Entry = PipelineEntry>,
     SB: SuperblockStore,
     J: JournalHandle,
-    J::Target: Journal<J::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+    J::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
     M: StreamsFrontend
         + StateMachine<
             Input = Message<PrepareHeader>,
@@ -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_index_reader.rs

@@ -170,8 +170,42 @@ impl IggyIndexReader {
         entry_count: u64,
         offset: u64,
     ) -> Result<Option<IggyIndex>, IggyError> {
-        self.lower_bound_by(entry_count, |entry| entry.offset, offset)
-            .await
+        Ok(self
+            .lower_bound_by(entry_count, |entry| entry.offset, offset)
+            .await?
+            .map(|(entry, _)| entry))
+    }
+
+    /// [`Self::offset_lower_bound`] that also reports the successor entry's
+    /// offset (`None` when the match is the last entry), bounding the offset
+    /// interval the match resolves for. Costs one extra entry read; callers
+    /// memoizing the resolution use the bound to answer later in-interval
+    /// queries without reopening the index.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if any probed entry cannot be read.
+    pub async fn offset_lower_bound_with_successor(
+        &self,
+        entry_count: u64,
+        offset: u64,
+    ) -> Result<Option<(IggyIndex, Option<u64>)>, IggyError> {
+        let Some((entry, successor_index)) = self
+            .lower_bound_by(entry_count, |entry| entry.offset, offset)
+            .await?
+        else {
+            return Ok(None);
+        };
+        let successor_offset = if successor_index < entry_count {
+            Some(
+                self.read_entry_at(successor_index * IGGY_INDEX_SIZE as u64)
+                    .await?
+                    .offset,
+            )
+        } else {
+            None
+        };
+        Ok(Some((entry, successor_offset)))
     }
 
     /// Last entry with `timestamp` at or below the target; `None` semantics
@@ -185,16 +219,21 @@ impl IggyIndexReader {
         entry_count: u64,
         timestamp: u64,
     ) -> Result<Option<IggyIndex>, IggyError> {
-        self.lower_bound_by(entry_count, |entry| entry.timestamp, timestamp)
-            .await
+        Ok(self
+            .lower_bound_by(entry_count, |entry| entry.timestamp, timestamp)
+            .await?
+            .map(|(entry, _)| entry))
     }
 
+    /// Binary search for the last entry with `key` at or below `target`,
+    /// returning it with its successor's entry index (the standard
+    /// lower-bound exit: `low` lands on the first entry above the target).
     async fn lower_bound_by(
         &self,
         entry_count: u64,
         key: impl Fn(&IggyIndex) -> u64,
         target: u64,
-    ) -> Result<Option<IggyIndex>, IggyError> {
+    ) -> Result<Option<(IggyIndex, u64)>, IggyError> {
         let mut low = 0u64;
         let mut high = entry_count;
         let mut result = None;
@@ -208,7 +247,7 @@ impl IggyIndexReader {
                 high = middle;
             }
         }
-        Ok(result)
+        Ok(result.map(|entry| (entry, low)))
     }
 }
 
@@ -271,6 +310,37 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    #[compio::test]
+    async fn offset_lower_bound_with_successor_reports_the_interval_ceiling() {
+        let (dir, path) = write_index_file(&entries()).await;
+        let reader = IggyIndexReader::new(&path).await.expect("open index");
+        let count = reader.entry_count().await.expect("entry count");
+
+        let mid = reader
+            .offset_lower_bound_with_successor(count, 25)
+            .await
+            .expect("lookup")
+            .expect("in range");
+        assert_eq!((mid.0.offset, mid.1), (20, Some(30)));
+        let last = reader
+            .offset_lower_bound_with_successor(count, 35)
+            .await
+            .expect("lookup")
+            .expect("in range");
+        assert_eq!(
+            (last.0.offset, last.1),
+            (30, None),
+            "the last entry has no successor to bound its interval",
+        );
+        let below_range = reader
+            .offset_lower_bound_with_successor(count, 5)
+            .await
+            .expect("lookup");
+        assert!(below_range.is_none());
+
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[compio::test]
     async fn timestamp_lower_bound_on_file_returns_predecessor() {
         let (dir, path) = write_index_file(&entries()).await;

core/partitions/src/iggy_partition.rs

@@ -96,7 +96,7 @@ where
     B: MessageBus,
 {
     consensus: VsrConsensus<B>,
-    pub log: SegmentedLog<PartitionJournal<PartitionJournalMemStorage>, PartitionJournalMemStorage>,
+    pub log: SegmentedLog<PartitionJournal<PartitionJournalMemStorage>>,
     /// Highest durably persisted offset.
     pub offset: Arc<AtomicU64>,
     /// Highest offset assigned to prepares that may still only live in the in-memory journal.
@@ -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/journal.rs

@@ -929,7 +929,7 @@ where
     }
 }
 
-impl Journal<PartitionJournalMemStorage> for PartitionJournal<PartitionJournalMemStorage> {
+impl Journal for PartitionJournal<PartitionJournalMemStorage> {
     type Header = PrepareHeader;
     type Entry = JournalBuffer;
     #[rustfmt::skip]

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.