fix(server): truncate torn segment tails instead of resurrecting them

core/common/src/error/iggy_error.rs

@@ -453,6 +453,11 @@ pub enum IggyError {
     InvalidOffset(u64) = 4100,
     #[error("Invalid reserved field value: {0}, expected: 0")]
     InvalidReservedField(u64) = 4101,
+    /// The on-disk segment file length disagrees with the recovered bounds the
+    /// writer was seeded with; appending would corrupt the segment, so the
+    /// open fails instead. Field order: `(on_disk, expected)`.
+    #[error("Segment file size on disk: {0} does not match expected size: {1}")]
+    SegmentSizeMismatchAtOpen(u64, u64) = 4102,
     #[error("Consumer group with ID: {0} for topic with ID: {1} was not found.")]
     ConsumerGroupIdNotFound(Identifier, Identifier) = 5000,
     #[error("Invalid consumer group ID")]

core/common/src/lib.rs

@@ -40,6 +40,17 @@ pub use consumer_group_client_state::ConsumerGroupClientState;
 /// partition id (those are small, dense, zero-based), so it can't collide with
 /// a genuine end-of-partition empty poll, which echoes the real partition id.
 pub const RESYNC_REQUIRED_PARTITION_SENTINEL: u32 = u32::MAX;
+
+/// Frozen ceiling on `message_bus.max_message_size`, the knob that caps a
+/// single framed wire message and with it the widest batch record any
+/// admission path can persist. Frozen rather than knob-derived because
+/// boot-time segment recovery sizes fixed scan and allocation limits from the
+/// widest LEGAL record: a limit read from the live knob would change meaning
+/// between boots and refuse partitions written under an older value. Config
+/// validation rejects a knob above this at boot; raising it is a
+/// compatibility decision, not a tuning change, since segments written under
+/// a larger value would exceed what recovery on an older build accepts.
+pub const MAX_MESSAGE_SIZE_UPPER_BYTES: u64 = 256 * 1024 * 1024;
 pub use http::consumer_groups::*;
 pub use http::consumer_offsets::*;
 pub use http::messages::*;

core/configs/src/server_config/message_bus.rs

@@ -54,7 +54,7 @@
 use super::COMPONENT;
 use crate::ConfigurationError;
 use configs::ConfigEnv;
-use iggy_common::{IggyByteSize, IggyDuration, Validatable};
+use iggy_common::{IggyByteSize, IggyDuration, MAX_MESSAGE_SIZE_UPPER_BYTES, Validatable};
 use serde::{Deserialize, Serialize};
 use serde_with::{DisplayFromStr, serde_as};
 
@@ -145,6 +145,16 @@ impl Validatable<ConfigurationError> for MessageBusConfig {
             eprintln!("{COMPONENT} message_bus.max_message_size must be > 0");
             return Err(ConfigurationError::InvalidConfigurationValue);
         }
+        if self.max_message_size.as_bytes_u64() > MAX_MESSAGE_SIZE_UPPER_BYTES {
+            eprintln!(
+                "{COMPONENT} message_bus.max_message_size ({}) exceeds the frozen ceiling of \
+                 {MAX_MESSAGE_SIZE_UPPER_BYTES} bytes: boot-time segment recovery derives fixed \
+                 scan and allocation limits from the widest legal wire frame, so batches admitted \
+                 above the ceiling would be refused as implausible by recovery on a later boot",
+                self.max_message_size.as_bytes_u64()
+            );
+            return Err(ConfigurationError::InvalidConfigurationValue);
+        }
         if self.handshake_grace.as_micros() == 0 {
             eprintln!("{COMPONENT} message_bus.handshake_grace must be > 0");
             return Err(ConfigurationError::InvalidConfigurationValue);
@@ -213,6 +223,20 @@ mod tests {
         assert!(c.validate().is_err());
     }
 
+    #[test]
+    fn rejects_max_message_size_above_frozen_ceiling() {
+        let mut c = baseline();
+        c.max_message_size = IggyByteSize::from(MAX_MESSAGE_SIZE_UPPER_BYTES + 1);
+        assert!(c.validate().is_err());
+    }
+
+    #[test]
+    fn accepts_max_message_size_at_frozen_ceiling() {
+        let mut c = baseline();
+        c.max_message_size = IggyByteSize::from(MAX_MESSAGE_SIZE_UPPER_BYTES);
+        assert!(c.validate().is_ok());
+    }
+
     /// Tripwire: pins the local copy of `IOV_MAX_LIMIT` against the
     /// runtime crate's value. If `core/message_bus` ever bumps its
     /// `IOV_MAX_LIMIT`, this test fails the configs build until the

core/integration/tests/cluster/crash_recovery_corruption.rs

@@ -302,8 +302,8 @@ fn flip_interior_byte(path: &Path, numerator: u64, denominator: u64) {
 }
 
 /// Byte-compare the partition segment `.log` files across all nodes, panicking
-/// with `defect` (the named bug) plus a per-file diff on divergence.
-fn assert_segment_logs_identical(data_paths: &[PathBuf], defect: &str) {
+/// with the violated `invariant` plus a per-file diff on divergence.
+fn assert_segment_logs_identical(data_paths: &[PathBuf], invariant: &str) {
     let per_node: Vec<_> = data_paths
         .iter()
         .map(|root| disk::collect_comparable_files(root, false))
@@ -324,27 +324,19 @@ fn assert_segment_logs_identical(data_paths: &[PathBuf], defect: &str) {
             }
         }
     }
-    assert!(problems.is_empty(), "{defect}:\n{}", problems.join("\n"));
+    assert!(problems.is_empty(), "{invariant}:\n{}", problems.join("\n"));
 }
 
-/// RED SPEC, expected to FAIL: a torn tail of garbage on the active segment
-/// `.log` must be truncated for good by recovery. Recovery does walk whole
-/// batches and computes the correct pre-garbage size, but reopening the
-/// segment re-stats the RAW file length into the writer's size counter
-/// (`MessagesWriter::new` with `file_exists = true`), while the recovered
-/// segment metadata keeps the walked size. Post-recovery appends then land at
-/// the raw position (after the resurrected garbage) while their index entries
-/// record the walked position, so every new index entry points below where
-/// its bytes physically landed and the NEXT recovery refuses the segment as
-/// message/index divergence. One torn tail on one replica permanently bricks
-/// that node's next restart.
-///
-/// The refusal also provokes a distinct defect this spec does not assert:
-/// the owner shard's boot refusal cascades the surviving shards into
-/// metadata STM read-handle panics (`ShardBootstrapBarrierAborted`, then
-/// `ShardPumpDied`) instead of a clean fail-stop.
-// TODO(hubcio): fix this test
-#[ignore = "torn .log tail re-stated into the size counter bricks the next restart"]
+/// A torn tail of garbage on the active segment `.log` must be truncated for
+/// good by recovery: the walked bounds govern both the on-disk length and the
+/// reopened write cursor, so post-recovery appends land exactly where their
+/// index entries point. The spec asserts the outcomes that used to break: the
+/// node survives its NEXT restart (a stale cursor bricked it as message/index
+/// divergence), every acked offset still reads back, and the at-rest `.log`
+/// bytes stay identical across replicas (resurrected garbage diverged them).
+// TODO(hubcio): both torn-tail specs tear a BACKUP; add a primary-side
+// variant (tear the leader's segment, restart it) since the leader path
+// exercises different reopen and catch-up code.
 #[iggy_harness(cluster_nodes = 3)]
 async fn given_a_torn_segment_tail_when_a_node_recovers_should_keep_size_counter_consistent(
     harness: &mut TestHarness,
@@ -364,7 +356,15 @@ async fn given_a_torn_segment_tail_when_a_node_recovers_should_keep_size_counter
         .await;
     }
 
-    let (_, backup) = pick_backup(harness).await;
+    // The setup client is pinned to node 0, which is the backup whenever the
+    // leader sits elsewhere; harness clients never reconnect, so swap to a
+    // producer on the leader, which stays up across both backup restarts.
+    drop(client);
+    let (leader, backup) = pick_backup(harness).await;
+    let client = harness
+        .root_client_for_node(leader)
+        .await
+        .expect("connect a producer to the leader");
     harness.stop_node(backup).expect("stop the backup");
     let segment_log = find_active_segment_file(&harness.node(backup).data_path(), "log");
     append_garbage(&segment_log, TORN_LOG_GARBAGE);
@@ -388,13 +388,8 @@ async fn given_a_torn_segment_tail_when_a_node_recovers_should_keep_size_counter
 
     harness.restart_node(backup).unwrap_or_else(|error| {
         panic!(
-            "the second recovery over a repaired torn tail must boot cleanly, but the \
-             segment reopen re-stats the raw file length (garbage included) into the \
-             messages writer while the recovered segment metadata keeps the walked \
-             size, so post-recovery appends land after the resurrected garbage and \
-             their index entries point below where the bytes physically landed; the \
-             next boot then refuses the segment as message/index divergence and the \
-             node cannot restart over its own data; boot error: {error}"
+            "a node must restart cleanly over a segment it repaired and then \
+             appended to; boot error: {error}"
         )
     });
     let nodes: Vec<usize> = (0..harness.cluster_size()).collect();
@@ -416,25 +411,18 @@ async fn given_a_torn_segment_tail_when_a_node_recovers_should_keep_size_counter
         .expect("stop the cluster for the at-rest comparison");
     assert_segment_logs_identical(
         &data_paths,
-        "the torn segment tail was resurrected into the live byte range: recovery \
-         computes the valid size by walking whole batches past the garbage, but the \
-         segment reopen re-stats the raw file length into the shared size counter, \
-         so subsequent appends land after the garbage and this replica's segment \
-         bytes diverge from its peers for the same acked history",
+        "segment .log files must stay byte-identical across replicas after a \
+         torn-tail recovery",
     );
 }
 
-/// RED SPEC, expected to FAIL: a torn tail on the segment `.index` (10 bytes,
-/// not a multiple of the 24-byte entry stride) must not poison later entries.
-/// The first recovery ignores the partial tail (whole-entry floor division) and
-/// boots, but the index reopen re-stats the RAW length and appends every new
-/// entry at that unaligned position, while every reader addresses entries as
-/// stride-from-0. The next recovery then decodes a garbage-straddling entry as
-/// the last flush and refuses the partition (or serves misresolved reads).
-/// The refusal provokes the same unasserted shard-cascade defect as the torn
-/// `.log` spec above.
-// TODO(hubcio): fix this test
-#[ignore = "torn .index tail misaligns subsequent entries off the 24-byte stride"]
+/// A torn tail on the segment `.index` (10 bytes, not a multiple of the
+/// 24-byte entry stride) must not poison later entries: recovery floors the
+/// index to whole entries and truncates the partial tail off the file, so
+/// every subsequent entry keeps the stride-from-0 addressing readers assume.
+/// The spec asserts the node survives its NEXT restart (an unaligned write
+/// cursor used to make the following recovery decode a garbage-straddling
+/// entry and refuse the partition) and every acked offset still reads back.
 #[iggy_harness(cluster_nodes = 3)]
 async fn given_a_torn_index_tail_when_a_node_recovers_should_not_misalign_subsequent_entries(
     harness: &mut TestHarness,
@@ -454,7 +442,15 @@ async fn given_a_torn_index_tail_when_a_node_recovers_should_not_misalign_subseq
         .await;
     }
 
-    let (_, backup) = pick_backup(harness).await;
+    // The setup client is pinned to node 0, which is the backup whenever the
+    // leader sits elsewhere; harness clients never reconnect, so swap to a
+    // producer on the leader, which stays up across both backup restarts.
+    drop(client);
+    let (leader, backup) = pick_backup(harness).await;
+    let client = harness
+        .root_client_for_node(leader)
+        .await
+        .expect("connect a producer to the leader");
     harness.stop_node(backup).expect("stop the backup");
     let segment_index = find_active_segment_file(&harness.node(backup).data_path(), "index");
     append_garbage(&segment_index, TORN_INDEX_GARBAGE);
@@ -476,11 +472,8 @@ async fn given_a_torn_index_tail_when_a_node_recovers_should_not_misalign_subseq
 
     harness.restart_node(backup).unwrap_or_else(|error| {
         panic!(
-            "the recovery after a torn index tail must not misalign subsequent index \
-             entries: the index reopen re-stats the raw (unaligned) file length and \
-             appends new entries off the 24-byte stride, while recovery reads entries \
-             as stride-from-0, so the second boot decodes a garbage-straddling entry \
-             and refuses the segment; boot error: {error}"
+            "a node must restart cleanly over an index it repaired and then \
+             appended to; boot error: {error}"
         )
     });
 
@@ -489,11 +482,7 @@ async fn given_a_torn_index_tail_when_a_node_recovers_should_not_misalign_subseq
     wait_for_acked_readable(&client, &acked, CONVERGE_TIMEOUT)
         .await
         .unwrap_or_else(|state| {
-            panic!(
-                "every acked offset must poll back after the torn-index recovery \
-                 (index entries misaligned off the 24-byte stride misresolve reads): \
-                 {state}"
-            )
+            panic!("every acked offset must poll back after the torn-index recovery: {state}")
         });
 }
 

core/journal/src/file_storage.rs

@@ -61,6 +61,11 @@ impl FileStorage {
     ///
     /// # Errors
     /// Returns an I/O error if truncation fails.
+    // TODO(hubcio): compio `set_len` submits IORING_OP_FTRUNCATE, which kernels
+    // below 6.9 do not support; the driver then falls back to its blocking
+    // pool, and shard proactors run with `thread_pool_limit(0)`, so the torn
+    // WAL repair panics the shard on such kernels instead of repairing. Use a
+    // synchronous `std::fs` truncate here (boot-time path) or gate on a probe.
     pub async fn truncate(&self, len: u64) -> io::Result<()> {
         let file = unsafe { &*self.file.get() };
         file.set_len(len).await?;

core/partitions/src/iggy_index_writer.rs

@@ -20,7 +20,7 @@ use compio::io::AsyncWriteAtExt;
 use iggy_common::IggyError;
 use std::rc::Rc;
 use std::sync::atomic::{AtomicU64, Ordering};
-use tracing::trace;
+use tracing::{error, trace};
 
 #[derive(Debug)]
 pub struct IggyIndexWriter {
@@ -35,7 +35,8 @@ impl IggyIndexWriter {
     ///
     /// # Errors
     ///
-    /// Returns an error if the file cannot be opened, synchronized, or queried for metadata.
+    /// Returns an error if the file cannot be opened, synchronized, or queried for
+    /// metadata, or if the on-disk length does not match the seeded size counter.
     pub async fn new(
         file_path: &str,
         index_size_bytes: Rc<AtomicU64>,
@@ -63,7 +64,21 @@ impl IggyIndexWriter {
                 .map_err(|_| IggyError::CannotReadFileMetadata)?
                 .len();
 
-            index_size_bytes.store(actual_index_size, Ordering::Relaxed);
+            // Refusal rationale documented on `IggyError::SegmentSizeMismatchAtOpen`.
+            let expected_index_size = index_size_bytes.load(Ordering::Relaxed);
+            if actual_index_size != expected_index_size {
+                error!(
+                    target: "iggy.partitions.storage",
+                    file = file_path,
+                    on_disk_size = actual_index_size,
+                    expected_size = expected_index_size,
+                    "sparse index file size does not match the seeded size at open"
+                );
+                return Err(IggyError::SegmentSizeMismatchAtOpen(
+                    actual_index_size,
+                    expected_index_size,
+                ));
+            }
         }
 
         let size = index_size_bytes.load(Ordering::Relaxed);
@@ -101,13 +116,16 @@ impl IggyIndexWriter {
             .0
             .map_err(|_| IggyError::CannotSaveIndexToSegment)?;
 
-        self.index_size_bytes
-            .fetch_add(len as u64, Ordering::Release);
-
         if self.fsync {
             self.fsync().await?;
         }
 
+        // Advance the write cursor last: if the write or fsync fails, the
+        // counter must stay put so the retry overwrites the same slot instead
+        // of appending a duplicate entry that boot recovery would refuse.
+        self.index_size_bytes
+            .fetch_add(len as u64, Ordering::Release);
+
         trace!(
             target: "iggy.partitions.storage",
             file = self.file_path.as_str(),
@@ -135,3 +153,29 @@ impl IggyIndexWriter {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[compio::test]
+    async fn given_seeded_size_diverging_from_disk_when_opening_existing_file_should_return_size_mismatch_error()
+     {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("segment.index");
+        std::fs::write(&path, [7u8; 96]).unwrap();
+
+        let result = IggyIndexWriter::new(
+            path.to_str().unwrap(),
+            Rc::new(AtomicU64::new(32)),
+            false,
+            true,
+        )
+        .await;
+
+        assert!(matches!(
+            result,
+            Err(IggyError::SegmentSizeMismatchAtOpen(96, 32))
+        ));
+    }
+}