diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs index 52981fba4f..e79d96b897 100644 --- a/core/common/src/error/iggy_error.rs +++ b/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")] diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs index 777b5ef276..7225b2c20f 100644 --- a/core/common/src/lib.rs +++ b/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::*; diff --git a/core/configs/src/server_config/message_bus.rs b/core/configs/src/server_config/message_bus.rs index 91a84d6a01..77c2bbf9d2 100644 --- a/core/configs/src/server_config/message_bus.rs +++ b/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 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 diff --git a/core/integration/tests/cluster/crash_recovery_corruption.rs b/core/integration/tests/cluster/crash_recovery_corruption.rs index fdd5b98226..63a82d2156 100644 --- a/core/integration/tests/cluster/crash_recovery_corruption.rs +++ b/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 = (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}") }); } diff --git a/core/journal/src/file_storage.rs b/core/journal/src/file_storage.rs index b6af782773..1273d27b98 100644 --- a/core/journal/src/file_storage.rs +++ b/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?; diff --git a/core/partitions/src/iggy_index_writer.rs b/core/partitions/src/iggy_index_writer.rs index 1eb11599a1..939652360c 100644 --- a/core/partitions/src/iggy_index_writer.rs +++ b/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, @@ -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)) + )); + } +} diff --git a/core/partitions/src/messages_writer.rs b/core/partitions/src/messages_writer.rs index 4ecaadf509..1d8e8ab24f 100644 --- a/core/partitions/src/messages_writer.rs +++ b/core/partitions/src/messages_writer.rs @@ -45,7 +45,8 @@ impl MessagesWriter { /// /// # 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, messages_size_bytes: Rc, @@ -78,7 +79,21 @@ impl MessagesWriter { .map_err(|_| IggyError::CannotReadFileMetadata)? .len(); - messages_size_bytes.store(actual_messages_size, Ordering::Relaxed); + // Refusal rationale documented on `IggyError::SegmentSizeMismatchAtOpen`. + let expected_messages_size = messages_size_bytes.load(Ordering::Relaxed); + if actual_messages_size != expected_messages_size { + error!( + target: "iggy.partitions.storage", + file = file_path, + on_disk_size = actual_messages_size, + expected_size = expected_messages_size, + "segment messages file size does not match the seeded size at open" + ); + return Err(IggyError::SegmentSizeMismatchAtOpen( + actual_messages_size, + expected_messages_size, + )); + } } Ok(Self { @@ -122,9 +137,9 @@ impl MessagesWriter { /// save then failed. The committed prefix stays resident and is /// re-persisted on the next `commit_messages`; rewinding the cursor makes /// that retry overwrite the same region instead of appending a second copy - /// of the committed batch. The on-disk bytes are left in place (they are - /// committed data the retry overwrites), and after a crash the cursor - /// reinitializes from the file length, so no truncation is needed. + /// of the committed batch. Crash-safe without truncating: the rewound + /// bytes are whole batch records past the last index entry, so boot + /// recovery's indexed walk absorbs them and its truncate is a no-op. pub(crate) fn rewind(&self, bytes: u64) { debug_assert!( bytes <= self.messages_size_bytes.load(Ordering::Relaxed), @@ -256,4 +271,26 @@ mod tests { assert_eq!(writer.file.metadata().await.unwrap().len(), 0); } + + #[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.log"); + std::fs::write(&path, [7u8; 128]).unwrap(); + + let result = MessagesWriter::new( + path.to_str().unwrap(), + Rc::new(AtomicU64::new(129)), + false, + true, + None, + ) + .await; + + assert!(matches!( + result, + Err(IggyError::SegmentSizeMismatchAtOpen(128, 129)) + )); + } } diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index 4e331eb5c1..6031d0e6a8 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -41,8 +41,8 @@ use consensus::{ArtifactProgress, Sequencer as _, StateArtifactHasher, state_art use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset, IggyByteSize}; use journal::superblock::SuperblockStore; use message_bus::MessageBus; -use server_common::SegmentStorage; use server_common::send_messages::decode_batch_slice; +use server_common::{SegmentStorage, yield_to_reactor}; use std::collections::HashSet; use std::fmt; use std::mem::size_of; @@ -2798,22 +2798,6 @@ impl std::error::Error for SpillError {} /// per-poll binary-search fallback. const INDEX_STRIDE_BYTES: usize = 64 * 1024; -/// Hand the core back to the reactor mid-CPU-pass. -/// -/// Reactor only: the consensus tick shares this task as a sibling -/// `select_biased!` arm, and arms are not polled while one arm's body awaits, so -/// yielding here does not unfreeze ticks or heartbeats. -/// -/// A zero-duration timer, NOT a bare self-waking yield: this runtime does not -/// reliably re-poll a task that woke itself from inside its own poll, and a -/// pump that suspends that way stops driving consensus entirely (the frame -/// handler never resumes, ticks stop, the node goes quiet until something else -/// wakes it). Registering with the reactor is what every other yield on these -/// paths does -- the serving side yields through real file reads. -async fn yield_to_reactor() { - compio::time::sleep(std::time::Duration::ZERO).await; -} - /// Chunk size for the offer build's streaming checksum pass. Large enough /// that per-chunk overhead is noise, small enough that the pump yields to /// the reactor many times per segment. diff --git a/core/server/config.toml b/core/server/config.toml index d980987990..93cfc7d3d0 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -463,6 +463,26 @@ archive_expired = false # Unsupported: setting this to `true` aborts boot. recreate_missing_state = false +# At boot, segment recovery walks each partition's segments: bytes after the +# last decodable batch of a genuinely torn tail are physically truncated from +# the .log/.index files, and the index is rebuilt when it was damaged. Only +# the walk of a segment whose index was lost re-checksums batches; with an +# intact index the walk trusts batch headers (decodable, contiguous offsets), +# and bytes before the last index entry are not re-examined at boot at all -- +# at-rest damage there surfaces on the read path via validate_checksum. +# Damage in the middle of a segment, or trailing bytes too large or costly to +# prove torn, is never silently truncated: the partition is refused. With +# peer replicas the refused files are moved to a .fenced.N directory beside +# the partition, which is rebuilt empty and refilled from a peer. With +# replica_count = 1 there is no peer: the refused files stay at their +# original paths, the partition is tombstoned (unrouted, never served +# empty), and the same refusal is re-derived and re-logged on every boot +# until an operator intervenes. Only single-replica refusals with no +# recoverable bytes at stake (a hole from a stray file, an orphaned empty +# segment) still rebuild through .fenced.N. The metadata WAL truncates +# genuinely torn tails too; interior WAL damage or oversized trailing bytes +# refuse boot instead. + # Memory pool configuration [system.memory_pool] # Enables or disables the memory pool (boolean). diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs index 894a8a90bf..5044480e2b 100644 --- a/core/server/src/bootstrap.rs +++ b/core/server/src/bootstrap.rs @@ -29,7 +29,9 @@ use crate::partition_helpers::{ open_partition_superblock, }; use crate::segment_recovery::{RecoveredSegment, load_persisted_segments}; -use crate::server_error::{ServerError, ShardJoinFailure, ShardJoinFailureKind}; +use crate::server_error::{ + PartitionRecoveryRefusal, ServerError, ShardJoinFailure, ShardJoinFailureKind, +}; use crate::session_manager::SessionManager; use compio::runtime::ResumeUnwind; use configs::server::{ServerConfig, ServerSystemConfig}; @@ -51,7 +53,8 @@ use iggy_common::defaults::{ MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH, }; use iggy_common::{ - Aes256GcmEncryptor, EncryptorKind, IggyByteSize, PartitionStats, TopicRuntimeOptions, variadic, + Aes256GcmEncryptor, EncryptorKind, IggyByteSize, IggyError, PartitionStats, + TopicRuntimeOptions, variadic, }; use journal::prepare_journal::PrepareJournal; use journal::superblock::{PingPongSuperblock, SuperblockStore}; @@ -1911,23 +1914,72 @@ async fn build_shard_for_thread( { Ok(partition) => partition, // ONE damaged local chain must not take the node down. The shapes - // this refuses are exactly what a failed state-transfer quarantine - // leaves behind, so fence that group the same way the runtime path - // does -- move its segment files aside, keeping the superblock so it - // cannot re-enter view 0 -- and materialise it fresh. The ordinary + // this refuses are structural -- what a failed state-transfer + // quarantine leaves behind, or damage the recovery walk proved + // inside a segment. What follows depends on whether a peer can + // restore the data. With peers, the segment files are fenced + // aside (keeping the superblock so the group cannot re-enter + // view 0), the group is materialised fresh, and the ordinary // rejoin path (repair, then state transfer on a refused floor) - // recovers its data from a peer. - Err(ServerError::PartitionChainRefused { dir, reason, .. }) => { + // refills it. Single-replica, only the two directory-shape + // refusals (a hole from a stray or half-unlinked file, an + // orphaned empty segment) still fence and rebuild: their segment + // bytes sit intact in quarantine and no damage verdict needs + // surfacing. Every refusal that proved or suspects damage + // tombstones instead, leaving its files exactly where they are: + // a rebuilt empty partition answers polls exactly like a healthy + // empty one and hides the loss, while an unrouted namespace is a + // failure an operator can see. + Err(ServerError::PartitionRecoveryRefused { dir, reason, .. }) => { let partition_dir = dir.to_string_lossy().into_owned(); + let rebuild_for_rejoin = topology.replica_count > 1 + || matches!( + reason, + PartitionRecoveryRefusal::Hole { .. } + | PartitionRecoveryRefusal::EmptyNonTailSegment { .. } + ); error!( stream_id, topic_id, partition_id = partition_metadata.id, partition_dir, %reason, - "refusing the recovered segment chain; fencing this partition and \ - rebuilding it empty for the rejoin path" + "refusing the recovered segment chain" ); + // A pass-A refusal folded nothing into the stats (recovery + // counts only accepted chains), but the hydrate-reopen refusal + // arrives after a fully counted load, so clear them either way. + partition_stats.zero_out_all(); + if !rebuild_for_rejoin { + // No quarantine here, mirroring the superblock arm below: + // a tombstone is only durable if its cause is. Fencing the + // chain aside would leave the next boot zero segments to + // walk, so it would re-seed from the surviving superblock, + // plant a fresh segment, and serve the partition empty + // with no refusal logged. Left at their real paths, the + // same files re-derive this verdict (and this log line) + // every boot, and the reconciler's tombstone gate keeps + // the namespace away from a fresh build, whose + // initial-segment open would truncate the oldest refused + // segment in place. The one refusal whose cause is NOT + // durable is `StorageSizeMismatch`: it fires from the + // reopen right after recovery truncated the same file, so + // the next boot re-walks the already-truncated bytes and, + // unless the length diverges again, accepts the chain + // instead of re-tombstoning -- acceptable for an + // assertion that the filesystem lied about a length. + error!( + stream_id, + topic_id, + partition_id = partition_metadata.id, + partition_dir, + "no peer replica holds this partition's data; leaving the refused \ + segment files in place and tombstoning it instead of serving it \ + empty" + ); + partitions.tombstone(namespace); + continue; + } match partitions::state_transfer::quarantine_segment_files(&partition_dir).await { Ok(fenced_dir) => error!( stream_id, @@ -1958,13 +2010,10 @@ async fn build_shard_for_thread( "failed to quarantine the refused segment files; leaving this \ partition tombstoned rather than rebuilding over them" ); - partition_stats.zero_out_all(); partitions.tombstone(namespace); continue; } } - // The refused load already folded its segment counts in. - partition_stats.zero_out_all(); build_partition_fresh( config, namespace, @@ -2605,13 +2654,14 @@ async fn load_partition( config.partition.evicted_ring_capacity, config.partition.evicted_ring_bytes_max.as_bytes_u64(), ); - partition.set_partition_dir(partition_dir); + partition.set_partition_dir(partition_dir.clone()); // Before the hydrate: the durable record is keyed by incarnation, so a // `purge.gen` left behind by a previous life of this namespace reads 0. partition.set_created_revision(partition_metadata.created_revision); partition.hydrate_applied_purge_generation().await?; hydrate_partition_log( &mut partition, + &partition_dir, stream_id, topic_id, partition_id, @@ -2677,6 +2727,7 @@ async fn load_partition( /// resolved topic option now, which is the whole point of the per-topic move. async fn hydrate_partition_log( partition: &mut IggyPartition>, + partition_dir: &str, stream_id: usize, topic_id: usize, partition_id: usize, @@ -2716,9 +2767,10 @@ async fn hydrate_partition_log( storage.index_writer.as_ref(), ) { let index_path = index_reader.path(); - // Share the storage's size counters: the readers bound reads by - // these atomics, so a writer with a private counter persists bytes - // the readers never learn about. + let start_offset = partition.log.segments()[active_index].start_offset; + // Share the storage's size counters: they are the write cursors. + // A private counter would let the append position diverge from the + // segment bookkeeping that index entries and poll bounds rely on. let messages_size_counter = storage_messages_writer.size_counter(); let index_size_counter = storage_index_writer.size_counter(); partition.log.messages_writers_mut()[active_index] = Some(Rc::new( @@ -2739,7 +2791,14 @@ async fn hydrate_partition_log( error = %source, "failed to initialize persisted messages writer" ); - source + hydrate_reopen_error( + source, + partition_dir, + stream_id, + topic_id, + partition_id, + start_offset, + ) })?, )); partition.log.index_writers_mut()[active_index] = Some(Rc::new( @@ -2754,7 +2813,14 @@ async fn hydrate_partition_log( error = %source, "failed to initialize persisted sparse index writer" ); - source + hydrate_reopen_error( + source, + partition_dir, + stream_id, + topic_id, + partition_id, + start_offset, + ) })?, )); } @@ -2763,6 +2829,42 @@ async fn hydrate_partition_log( Ok(()) } +/// Routes a hydrate-reopen writer failure. The seed-vs-stat divergence guard +/// (`SegmentSizeMismatchAtOpen`) is a post-condition assertion on recovery's +/// own truncation: pass C truncates every file to its recovered size before +/// storage and writers reopen it, so the guard can only fire if the +/// filesystem lied about a length or a change broke that truncate-then-open +/// contract. Kept as defense-in-depth and routed as a structural refusal +/// because a retried boot cannot help. Every other failure here (open, stat, +/// sync) is transient I/O and stays node-fatal: a retried boot can still +/// serve the partition, while fencing would quarantine healthy data (and at +/// `replica_count = 1` tombstone the partition outright). +fn hydrate_reopen_error( + source: IggyError, + partition_dir: &str, + stream_id: usize, + topic_id: usize, + partition_id: usize, + start_offset: u64, +) -> ServerError { + match source { + IggyError::SegmentSizeMismatchAtOpen(on_disk_bytes, expected_bytes) => { + ServerError::PartitionRecoveryRefused { + dir: PathBuf::from(partition_dir), + stream_id, + topic_id, + partition_id, + reason: PartitionRecoveryRefusal::StorageSizeMismatch { + start_offset, + on_disk_bytes, + expected_bytes, + }, + } + } + transient => transient.into(), + } +} + fn resolve_tcp_topology( config: &ServerConfig, current_replica_id: Option, diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index 820d40f5f4..5a2f49fd21 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -353,8 +353,9 @@ pub async fn ensure_initial_segment( ); source })?; - // Share the storage's size counters so reads observe persisted bytes; - // a writer with a private counter grows the file invisibly to readers. + // Share the storage's size counters: they are the write cursors. A private + // counter would let the append position diverge from the segment + // bookkeeping that index entries and poll bounds rely on. let messages_size_counter = storage .messages_writer .as_ref() diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index d3a3b1f80e..99f03ed8a5 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -527,6 +527,7 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool { true } +#[allow(clippy::too_many_lines)] async fn reconcile_additions( ctx: &ReconcilerCtx, target: Vec<(IggyNamespace, u64)>, @@ -598,6 +599,25 @@ async fn reconcile_additions( continue; } + // Tombstoned without ever being materialised: a boot-time damage + // verdict (a refused segment chain, an untrusted superblock) fenced + // the namespace before any partition existed, so no teardown ran and + // no `ConfirmRemove` is coming to lift the tombstone. Building fresh + // would plant segment 0 over the refused files, truncating the + // oldest one, and the partition would then serve empty, hiding + // exactly the loss the tombstone surfaces. Deliberately uncounted: + // nothing lifts this state short of a metadata commit, and a commit + // bumps `Streams::revision`, which forces the next pass past the + // fast-skip. + if partitions.is_tombstoned(&ns) { + trace!( + shard = shard_id, + ns_raw = ns.inner(), + "additions: ns tombstoned before materialisation; refusing to rebuild over fenced files" + ); + continue; + } + let owning_shard = calculate_shard_assignment(&ns, total_shards); if owning_shard != shard_id { // Compare the epoch, not just presence: a delete + recreate @@ -2659,6 +2679,100 @@ mod tests { ); } + /// A namespace tombstoned before it was ever materialised is a boot-time + /// damage verdict (a refused segment chain, an untrusted superblock): its + /// files are still on disk and no `ConfirmRemove` is coming. The + /// additions pass must not rebuild it -- `build_partition_fresh` would + /// plant segment 0 over the refused files and the partition would serve + /// empty -- and it must stay unrouted so the loss stays visible. + #[compio::test] + async fn reconcile_never_rebuilds_tombstoned_unmaterialised_namespace() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-fence"); + seed_topic(&mux, 2, 0, "topic-fence", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + let ns = IggyNamespace::new(0, 0, 0); + let partitions = shard.plane.partitions(); + // Boot-fence shape: the verdict lands during partition load, before + // anything is inserted, so the namespace is tombstoned but absent + // from the map while still in the committed target. + partitions.tombstone(ns); + + reconcile_pass(&ctx).await; + + assert!( + !partitions.contains(&ns), + "tombstoned namespace must not be rebuilt" + ); + assert!( + partitions.is_tombstoned(&ns), + "the boot fence must survive the pass" + ); + assert_eq!( + shard.shards_table().shard_for(ns), + None, + "tombstoned namespace must stay unrouted" + ); + let partition_root = ctx.config.system.get_partition_path(0, 0, 0); + assert!( + !std::path::Path::new(&partition_root).exists(), + "no fresh build may touch the refused files' directory" + ); + } + + /// Backstop at the pump: an `InsertOwned` staged before a tombstone + /// landed must be discarded at apply, not routed. Applying it would put + /// the namespace in `partitions` + `shards_table` while the tombstone + /// stands, and the plane drops requests for tombstoned namespaces + /// without replying, so every client would hang to its read timeout. + #[compio::test] + async fn apply_discards_insert_owned_for_tombstoned_namespace() { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-race"); + seed_topic(&mux, 2, 0, "topic-race", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + let ns = IggyNamespace::new(0, 0, 0); + let partitions = shard.plane.partitions(); + + // Stage the build without applying it, then fence: models a + // tombstone landing between the reconciler pass and the pump drain. + reconcile_once(&ctx).await; + assert!(shard.has_staged_insert_owned(ns)); + partitions.tombstone(ns); + shard.apply_reconcile_ops(); + + assert!( + !partitions.contains(&ns), + "InsertOwned for a tombstoned namespace must be discarded" + ); + assert!( + partitions.is_tombstoned(&ns), + "the discard must not clear the tombstone" + ); + assert_eq!( + shard.shards_table().shard_for(ns), + None, + "the discarded build must not route the namespace" + ); + + // The follow-up pass sees the same tombstone before building, so the + // namespace stays dark instead of looping build-and-discard. + reconcile_pass(&ctx).await; + assert!(!partitions.contains(&ns)); + assert!( + !shard.has_staged_insert_owned(ns), + "no second build may be staged while the tombstone stands" + ); + } + /// Deferral-arms-the-fast-skip regression: a deferred rebuild is pending /// work, but a pass that found nothing else used to report `did_work = /// false` and arm the fast-skip. The wake the pump fires after draining diff --git a/core/server/src/segment_recovery.rs b/core/server/src/segment_recovery.rs index fcaa57acea..10959bf98b 100644 --- a/core/server/src/segment_recovery.rs +++ b/core/server/src/segment_recovery.rs @@ -27,21 +27,70 @@ //! recovery panic). This module is the server-owned loader, reading the same //! 24-byte format its writer emits. -use crate::server_error::{PartitionChainRefusal, ServerError}; +use crate::server_error::{PartitionRecoveryRefusal, ServerError}; use configs::server::ServerConfig; -use iggy_common::{IggyByteSize, IggyError, PartitionStats}; +use iggy_common::{IggyByteSize, IggyError, MAX_MESSAGE_SIZE_UPPER_BYTES, PartitionStats}; use partitions::state_transfer::STAGING_SUFFIX; -use partitions::{IggyIndexReader, Segment}; -use server_common::SegmentStorage; +use partitions::{IggyIndex, IggyIndexReader, Segment}; use server_common::send_messages::{BatchHeader, COMMAND_HEADER_SIZE, decode_batch_slice}; +use server_common::{SegmentStorage, yield_to_reactor}; use std::fs; +use std::io; use std::os::unix::fs::FileExt; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use tracing::{error, warn}; const LOG_EXTENSION: &str = "log"; const INDEX_EXTENSION: &str = "index"; +/// On-disk stride of one sparse index entry (`offset`, `timestamp`, +/// `position`, each a little-endian u64). Mirrors `IGGY_INDEX_SIZE`, which is +/// crate-private to `partitions` alongside the reader and writer that own the +/// format: the reader's `entry_count` floors by it, and this module needs it +/// to turn that count back into bytes, validate whole entries, and emit +/// rebuilt ones. +const SPARSE_INDEX_ENTRY_SIZE: usize = std::mem::size_of::() * 3; + +/// Window for the buffered walk, probe, and index scans. One allocation per +/// partition load, refilled forward on demand; batches larger than this fall +/// back to a single direct read. +const SCAN_WINDOW_CAPACITY: usize = 4 * 1024 * 1024; + +/// Byte stride between rebuilt sparse index entries, mirroring the +/// state-transfer receiver's rebuild policy: lower-bound consumers are +/// correct at ANY density, but a per-batch index over a large segment +/// overshoots the sealed-index residency cap and demotes every sealed poll +/// to on-file binary search, so entries are spaced out. The first walked +/// batch always gets one. +const REBUILT_INDEX_STRIDE_BYTES: u64 = 64 * 1024; + +/// Units the damage probes of one partition load may spend per byte of +/// residue they are asked to classify, at one unit per candidate offset +/// examined. Candidates never outnumber residue bytes, so an honest +/// front-to-back scan always fits under this multiple regardless of residue +/// width; only a shape that re-examines offsets can exhaust it. Deriving the +/// limit from the residue actually present -- rather than from any +/// configuration knob or frozen size constant -- makes it immune to knob +/// changes between boots by construction: no legal segment can be refused +/// because a limit was derived from a value the segment was not written +/// under. Exhaustion refuses recovery; it never falls through to the +/// truncating no-survivor verdict. +const PROBE_BUDGET_UNITS_PER_RESIDUE_BYTE: u64 = 2; + +/// Largest on-disk batch record recovery treats as plausible: the frozen +/// ceiling on `message_bus.max_message_size` -- the widest wire frame any +/// legal configuration admits, validated at boot -- plus one batch header of +/// slack in case an admission path counts its cap against the blob alone. A +/// header claiming more cannot be a real batch, so rejecting it at the +/// header is verdict-identical to reading the claimed bytes and failing the +/// verify, minus a claimed-size allocation and read that a single +/// bit-flipped length field could otherwise drive up to a whole segment. +const MAX_RECOVERABLE_BATCH_BYTES: u64 = MAX_MESSAGE_SIZE_UPPER_BYTES + COMMAND_HEADER_SIZE as u64; + +/// Attempts at finding a free `.fenced.` name, mirroring +/// the partition-level quarantine's bound. +const FENCED_DIR_PROBE_LIMIT: u32 = 1000; + /// A persisted segment recovered from disk: its metadata plus the storage /// handles (readers/writers) opened over its `.log` / `.index` files. pub struct RecoveredSegment { @@ -52,14 +101,30 @@ pub struct RecoveredSegment { /// Loads every persisted segment for a partition, sorted by start offset. /// /// Segment offsets and timestamps are recovered from the 24-byte sparse index -/// (see module docs); segment byte size comes from the `.log` file. The last -/// segment is left unsealed so it can accept further writes. +/// (see module docs); segment byte size comes from walking the `.log` batch +/// chain. Recovery runs in three passes: every segment is bounded first +/// without touching an existing byte (pass A's only write is staging each +/// rebuilt index in a fresh `.staging` file), then the chain guard runs over +/// those bounds, and only an accepted chain is made physical -- torn tails +/// truncated, staged indexes renamed into place, unreadable segments fenced +/// aside -- before storage opens over it. A refusal raised in pass A or B +/// therefore leaves every pre-existing file byte-identical to what boot found +/// (staged scratch is swept at the next boot or quarantined with the fence). +/// Pass C is NOT atomic across segments: its own refusals -- the storage-open +/// guard -- can land after earlier segments in the chain were already +/// truncated. The last segment is left unsealed so it can accept further +/// writes. /// /// # Errors /// -/// Returns an error if the partition directory or a segment's files cannot be -/// read, or if a segment's index references a batch beyond the end of its -/// messages file (torn write). +/// Transient I/O failures (listing, stat, open, read, truncate, fsync) are +/// returned as-is and abort the boot so it can be retried. Structural +/// contradictions -- a holed chain, an index diverging from its log, damage +/// with intact batches after it, residue the damage probe could not classify +/// within its limits -- return +/// [`ServerError::PartitionRecoveryRefused`] so the caller can fence this one +/// partition instead of taking the node down. +#[allow(clippy::too_many_lines)] pub async fn load_persisted_segments( config: &ServerConfig, stream_id: usize, @@ -71,6 +136,12 @@ pub async fn load_persisted_segments( let partition_path = config .system .get_partition_path(stream_id, topic_id, partition_id); + let identity = PartitionIdentity { + partition_path: &partition_path, + stream_id, + topic_id, + partition_id, + }; // ONE directory walk feeds both: the sweep only ever unlinks `.staging` and // orphan `.index` files, never a `.log`, so the log stems it already // collects ARE the post-sweep start-offset set. Note the error policy is @@ -80,8 +151,15 @@ pub async fn load_persisted_segments( start_offsets.sort_unstable(); let max_size = segment_size; + let mut scratch = ScanScratch::default(); - let mut recovered = Vec::with_capacity(start_offsets.len()); + // Pass A: derive every segment's bounds without touching an existing + // byte (the only write is each rebuilt index staged to a fresh + // `.staging` scratch file). Nothing moves until the WHOLE chain is + // accepted, so a refusal raised by a later segment (or by the chain + // guard) leaves the earlier segments' files byte-identical for the + // caller's quarantine to keep. + let mut planned = Vec::with_capacity(start_offsets.len()); for start_offset in start_offsets { let messages_path = config @@ -92,53 +170,123 @@ pub async fn load_persisted_segments( .system .get_index_path(stream_id, topic_id, partition_id, start_offset); - let messages_size = file_len(&messages_path); - let index_size = file_len(&index_path); + let raw_messages_size = file_len(&messages_path)?; let bounds = recover_segment_bounds( + identity, &index_path, &messages_path, start_offset, - messages_size, - stream_id, - topic_id, - partition_id, + raw_messages_size, + &mut scratch, ) .await?; - // `bounds == None` now means the log holds no whole BATCH either (the - // index-less path above already tried walking the log), so there is - // nothing to recover: zeroed sizes make the next append overwrite the - // torn bytes, where counting them with `end_offset == start_offset` - // would fabricate one phantom message for the bootstrap non-empty - // filters and strand undecodable garbage inside the readable range. - // Note this is NOT tail-only -- a torn index is reachable mid-chain on - // the shipped `enforce_fsync = false`, which is why the walk above - // exists rather than refusing the partition. - let (start_timestamp, end_timestamp, end_offset, effective_messages_size) = - if let Some((start_timestamp, end_timestamp, end_offset, walked_size)) = bounds { - (start_timestamp, end_timestamp, end_offset, walked_size) - } else { - if messages_size > 0 { - warn!( - stream_id, - topic_id, - partition_id, - start_offset, - messages_size, - "segment log holds bytes but its index holds no whole \ - entry (torn write); recovering the segment as empty" - ); - } - (0, 0, start_offset, 0) - }; - let effective_index_size = if bounds.is_some() { index_size } else { 0 }; + // `bounds == None` means the log holds no whole batch ANYWHERE: the + // index-less walk tried from byte 0 and the damage probe found no + // surviving batch deeper in the file. There is nothing to serve: + // zeroed sizes seed fresh empty files (pass C fences the unreadable + // originals aside rather than deleting them), where counting the + // bytes with `end_offset == start_offset` would fabricate one + // phantom message for the bootstrap non-empty filters and strand + // undecodable garbage inside the readable range. Note this is NOT + // tail-only -- a torn index is reachable mid-chain on the shipped + // `enforce_fsync = false`, which is why the walk exists rather than + // refusing the partition. + let recovered_empty = bounds.is_none(); + let bounds = bounds.unwrap_or_else(|| { + if raw_messages_size > 0 { + warn!( + stream_id, + topic_id, + partition_id, + start_offset, + messages_size = raw_messages_size, + "segment log holds bytes but no whole batch decodes \ + anywhere in it (torn write); recovering the segment as \ + empty and fencing its files aside" + ); + } + WalkedBounds { + start_timestamp: 0, + end_timestamp: 0, + end_offset: start_offset, + messages_size: 0, + index_size: 0, + rebuilt_index: None, + } + }); + + // Staged now so pass C can install it with one atomic rename, and so + // a long chain never holds more than one rebuilt index in memory. + let rebuilt_index_staging = match &bounds.rebuilt_index { + Some(entries) => Some(stage_rebuilt_index(&index_path, entries)?), + None => None, + }; + + let mut segment = Segment::new(start_offset, max_size); + segment.sealed = true; + segment.start_timestamp = bounds.start_timestamp; + segment.end_timestamp = bounds.end_timestamp; + segment.max_timestamp = bounds.end_timestamp; + segment.end_offset = bounds.end_offset; + segment.size = IggyByteSize::from(bounds.messages_size); + segment.current_position = bounds.messages_size; + + planned.push(PlannedSegment { + segment, + messages_path, + index_path, + index_size: bounds.index_size, + rebuilt_index_staging, + recovered_empty, + }); + } + + if let Some(last) = planned.last_mut() { + last.segment.sealed = false; + } + + // Pass B: the chain guard reads only the planned bounds, so it can refuse + // BEFORE anything is truncated. + ensure_contiguous_chain(identity, &planned)?; + + // Pass C: the chain is accepted; make disk match the bounds and open + // storage over them. + let mut recovered = Vec::with_capacity(planned.len()); + for plan in planned { + let messages_size = plan.segment.size.as_bytes_u64(); + if plan.recovered_empty { + // The pair holds bytes that prove nothing, yet they are the only + // copy of whatever the crash tore: move them aside and seed fresh + // empty files rather than truncating them away. + fence_unrecoverable_segment_files( + identity, + &plan.messages_path, + &plan.index_path, + plan.segment.start_offset, + )?; + } + // Log first, index second: a walk only accepts bounds when a whole + // batch decodes at the last index entry's position, so the walked log + // length strictly exceeds that position and every surviving index + // entry still points inside the shortened log even if a crash lands + // between the two mutations. The staged-rebuild install keeps the + // same property: until its rename lands, the on-disk index still + // holds no whole entry, so a crash between the two re-runs the + // index-less walk over the already-truncated log. + truncate_to(&plan.messages_path, messages_size)?; + if let Some(staging_path) = &plan.rebuilt_index_staging { + install_rebuilt_index(staging_path, &plan.index_path, identity.partition_path)?; + } else { + truncate_to(&plan.index_path, plan.index_size)?; + } let storage = SegmentStorage::new( - &messages_path, - &index_path, - effective_messages_size, - effective_index_size, + &plan.messages_path, + &plan.index_path, + messages_size, + plan.index_size, true, ) .await @@ -147,46 +295,132 @@ pub async fn load_persisted_segments( stream_id, topic_id, partition_id, - path = %messages_path, + path = %plan.messages_path, error = %source, "failed to open persisted segment storage during recovery" ); - source + // The seed-vs-stat guard refusing the open is a post-condition + // assertion on the truncation this pass just performed: it can + // only fire if the filesystem lied about a length or a change + // broke the truncate-then-open contract. Kept as defense-in-depth + // and routed as a structural refusal (fence one partition, not + // the node) because a retried boot cannot help. Everything else + // here is transient I/O and stays node-fatal. + match source { + IggyError::SegmentSizeMismatchAtOpen(on_disk_bytes, expected_bytes) => identity + .refusal(PartitionRecoveryRefusal::StorageSizeMismatch { + start_offset: plan.segment.start_offset, + on_disk_bytes, + expected_bytes, + }), + transient => transient.into(), + } })?; - let mut segment = Segment::new(start_offset, max_size); - segment.sealed = true; - segment.start_timestamp = start_timestamp; - segment.end_timestamp = end_timestamp; - segment.max_timestamp = end_timestamp; - segment.end_offset = end_offset; - segment.size = IggyByteSize::from(effective_messages_size); - segment.current_position = effective_messages_size; - stats.increment_segments_count(1); - stats.increment_size_bytes(effective_messages_size); - if effective_messages_size > 0 { + stats.increment_size_bytes(messages_size); + if messages_size > 0 { // Offsets in a segment are contiguous, so the message count is the // inclusive span between the first (segment start) and last offset. - stats.increment_messages_count(end_offset - start_offset + 1); + stats.increment_messages_count(plan.segment.end_offset - plan.segment.start_offset + 1); } - recovered.push(RecoveredSegment { segment, storage }); + recovered.push(RecoveredSegment { + segment: plan.segment, + storage, + }); } - if let Some(last) = recovered.last_mut() { - last.segment.sealed = false; + Ok(recovered) +} + +/// Identity of the partition being recovered, threaded through the walk for +/// logs and refusal construction. +#[derive(Clone, Copy)] +struct PartitionIdentity<'load> { + partition_path: &'load str, + stream_id: usize, + topic_id: usize, + partition_id: usize, +} + +impl PartitionIdentity<'_> { + fn refusal(&self, reason: PartitionRecoveryRefusal) -> ServerError { + ServerError::PartitionRecoveryRefused { + dir: PathBuf::from(self.partition_path), + stream_id: self.stream_id, + topic_id: self.topic_id, + partition_id: self.partition_id, + reason, + } } +} - ensure_contiguous_chain( - &recovered, - &partition_path, - stream_id, - topic_id, - partition_id, - )?; +/// Pass A output for one segment: the recovered metadata plus what pass C +/// must make true on disk once the whole chain is accepted. +struct PlannedSegment { + segment: Segment, + messages_path: String, + index_path: String, + index_size: u64, + /// Path of the staged rebuilt index pass C renames over `index_path`. + rebuilt_index_staging: Option, + /// The pair holds bytes but nothing in them decodes; pass C fences the + /// files aside and reseeds empty ones instead of truncating. + recovered_empty: bool, +} - Ok(recovered) +/// Work bound shared by every damage probe in one partition load. +/// +/// One unit is charged per candidate byte offset a probe examines, and the +/// limit grows by [`PROBE_BUDGET_UNITS_PER_RESIDUE_BYTE`] units per residue +/// byte a probe is asked to classify. Examining a candidate is flat-cost by +/// construction -- the header decode bails on an undersized length or the +/// first nonzero reserved byte, and the sizes a decoded header CLAIMS gate +/// whether a verify runs at all -- so units track real work without scaling +/// with file size or any knob. +/// +/// Scoped to the LOAD, not to one probe: pass A probes every segment before +/// pass B can refuse the chain, so a per-probe budget would multiply the +/// worst case by the segment count. +#[derive(Default)] +struct ProbeBudget { + limit_units: u64, + spent_units: u64, +} + +impl ProbeBudget { + const fn grow_for_residue(&mut self, residue_bytes: u64) { + self.limit_units = self + .limit_units + .saturating_add(residue_bytes.saturating_mul(PROBE_BUDGET_UNITS_PER_RESIDUE_BYTE)); + } + + /// Charges one candidate; `false` means the budget is exhausted and the + /// probe must give up without a verdict. + const fn charge_candidate(&mut self) -> bool { + self.spent_units = self.spent_units.saturating_add(1); + self.spent_units <= self.limit_units + } +} + +/// Readable bounds recovered for one segment holding data. +struct WalkedBounds { + start_timestamp: u64, + end_timestamp: u64, + end_offset: u64, + messages_size: u64, + index_size: u64, + rebuilt_index: Option>, +} + +/// Reusable buffers for the walk, probe, and index validation scans, plus the +/// probe work budget they share, allocated once per partition load. +#[derive(Default)] +struct ScanScratch { + window: Vec, + spill: Vec, + probe_budget: ProbeBudget, } /// Contiguity guard: recovery takes every `.log` stem in the directory, so a @@ -195,27 +429,17 @@ pub async fn load_persisted_segments( /// chain and push `current_offset` past data this replica does not hold. /// Refuse loudly instead of serving a holed log. /// -/// The refusal names the partition and its directory so the caller can fence -/// THAT group rather than abort the node's boot: the shapes it rejects are -/// exactly what a failed quarantine leaves behind, and one damaged local chain -/// must not take the whole node down. +/// Runs on the planned bounds alone, BEFORE any truncation, so the segment +/// files a refusal quarantines are exactly the bytes boot found. The refusal +/// names the partition and its directory so the caller can fence THAT group +/// rather than abort the node's boot: the shapes it rejects are exactly what +/// a failed quarantine leaves behind, and one damaged local chain must not +/// take the whole node down. fn ensure_contiguous_chain( - recovered: &[RecoveredSegment], - partition_path: &str, - stream_id: usize, - topic_id: usize, - partition_id: usize, + identity: PartitionIdentity<'_>, + planned: &[PlannedSegment], ) -> Result<(), ServerError> { - let refused = |reason| { - Err(ServerError::PartitionChainRefused { - dir: PathBuf::from(partition_path), - stream_id, - topic_id, - partition_id, - reason, - }) - }; - for pair in recovered.windows(2) { + for pair in planned.windows(2) { let previous = &pair[0].segment; let next = &pair[1].segment; // A NON-tail empty segment can only be an orphan pairing: the torn- @@ -224,17 +448,19 @@ fn ensure_contiguous_chain( // more chain is exactly what a failed converge rebuild leaves behind. // Skipping it here was the guard's blind spot. if previous.size == IggyByteSize::default() { - return refused(PartitionChainRefusal::EmptyNonTailSegment { - empty_start: previous.start_offset, - next_start: next.start_offset, - }); + return Err( + identity.refusal(PartitionRecoveryRefusal::EmptyNonTailSegment { + empty_start: previous.start_offset, + next_start: next.start_offset, + }), + ); } if next.start_offset != previous.end_offset + 1 { - return refused(PartitionChainRefusal::Hole { + return Err(identity.refusal(PartitionRecoveryRefusal::Hole { previous_start: previous.start_offset, previous_end: previous.end_offset, next_start: next.start_offset, - }); + })); } } Ok(()) @@ -320,44 +546,362 @@ fn sweep_scratch_files_and_collect_offsets(partition_path: &str) -> Result u64 { - fs::metadata(path).map_or(0, |metadata| metadata.len()) +/// Byte length of a segment file, a missing file reading as empty. +/// +/// Any other stat failure is fail-stop, mirroring the `NotFound`-only leniency +/// of the directory listing above: recovery physically truncates files to the +/// bounds derived from these lengths, so folding a transient `EACCES` or +/// `EIO` into 0 would route a healthy segment into recover-as-empty, fencing +/// it out of service (worst route: an index stat error floors a healthy +/// sealed index to a 0-byte target while its entries still load). +fn file_len(path: &str) -> Result { + match fs::metadata(path) { + Ok(metadata) => Ok(metadata.len()), + Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(0), + Err(source) => { + error!( + path, + error = %source, + "failed to stat a segment file during recovery" + ); + Err(IggyError::CannotReadFileMetadata.into()) + } + } } -/// Derives `(start_timestamp, end_timestamp, end_offset)` from a segment's -/// 24-byte sparse index. `None` when the index holds no whole entry (the -/// caller recovers the segment as empty). The last entry's `position` is only -/// the last flushed batch's START byte, so the batch header is read back from -/// the messages file to prove the batch also ENDS inside it -- without -/// `enforce_fsync` there is no ordering barrier between the message write and -/// the index write, and a tail torn mid-flush would otherwise pass while -/// `end_offset` claims offsets whose bytes are incomplete. -#[allow(clippy::too_many_lines)] -async fn recover_segment_bounds( +/// Physically truncates a segment file to its recovered byte length, so disk +/// and the seeded size counters agree before storage reopens: reopen verifies +/// the on-disk length against the recovered size and refuses a divergence, +/// and before that check existed a leftover tail silently resurrected through +/// the writers' re-stat of the raw length. Truncation also protects state +/// transfer: the sender sizes each artifact from `segment.size` and hashes +/// exactly `[0, segment.size)`, so resurrected garbage INSIDE that range +/// would poison every artifact a torn replica offers once it serves as +/// primary. +/// +/// The tail being discarded was proven dead by the bounds walk: nothing past +/// the recovered size decodes (the interior-damage probe refuses recovery +/// outright when something does), so polls could never serve those bytes. +/// +/// Stats the file fresh instead of trusting a length carried from pass A: the +/// whole chain was walked in between, and the mutation must key on what is on +/// disk now. Synchronous `std::fs` on purpose (see [`FileScanner`]). The +/// fsync bounds the crash window: a power cut right after `set_len` may +/// re-present the torn tail on the next boot, which only walks and truncates +/// again (idempotent), but the sync keeps the common case deterministic. +fn truncate_to(path: &str, target_size: u64) -> Result<(), ServerError> { + let current_size = file_len(path)?; + if current_size == target_size { + return Ok(()); + } + // Unreachable by construction (walked bounds never exceed the file they + // were walked from); extending would fabricate a zero-filled tail, and + // zero bytes decode as valid-looking index entries -- three bare + // little-endian u64s with no magic to reject them -- so fail stop. + if target_size > current_size { + error!( + path, + current_size, + target_size, + "recovered bounds exceed the file they were walked from; \ + refusing to extend a segment file" + ); + return Err(IggyError::CannotWriteToFile.into()); + } + warn!( + path, + current_size, + target_size, + "truncating a segment file to its recovered bounds; discarding \ + torn tail bytes" + ); + let file = fs::OpenOptions::new() + .write(true) + .open(path) + .map_err(|source| { + error!( + path, + error = %source, + "failed to open a segment file for truncation during recovery" + ); + ServerError::from(IggyError::CannotWriteToFile) + })?; + file.set_len(target_size).map_err(|source| { + error!( + path, + target_size, + error = %source, + "failed to truncate a segment file to its recovered bounds" + ); + ServerError::from(IggyError::CannotWriteToFile) + })?; + file.sync_all().map_err(|source| { + error!( + path, + error = %source, + "failed to fsync a segment file after truncation" + ); + ServerError::from(IggyError::CannotSyncFile) + })?; + Ok(()) +} + +/// Stages the index rebuilt by the index-less walk in a scratch file beside +/// its final name. Without a rebuild a SEALED segment -- which never flushes +/// again -- would keep an empty index forever and pay a full log scan on +/// every poll. +/// +/// Staged, not written in place: an in-place writeback can tear -- a crash +/// mid-write may persist a later page while an earlier one still reads +/// zeros, and 24-byte zero runs decode as valid non-monotone entries, so the +/// next boot would fence the whole partition over its own repair artifact. +/// The staging file is pure scratch until pass C renames it into place: the +/// boot sweep unlinks orphaned `*.staging` files, so a crash anywhere before +/// the rename costs nothing. +fn stage_rebuilt_index(index_path: &str, entries: &[u8]) -> Result { + let staging_path = format!("{index_path}{STAGING_SUFFIX}"); + let file = fs::OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .open(&staging_path) + .map_err(|source| { + error!( + path = %staging_path, + error = %source, + "failed to open a sparse index staging file during recovery" + ); + ServerError::from(IggyError::CannotWriteToFile) + })?; + file.write_all_at(entries, 0).map_err(|source| { + error!( + path = %staging_path, + error = %source, + "failed to write a rebuilt sparse index during recovery" + ); + ServerError::from(IggyError::CannotWriteToFile) + })?; + file.sync_all().map_err(|source| { + error!( + path = %staging_path, + error = %source, + "failed to fsync a rebuilt sparse index after recovery" + ); + ServerError::from(IggyError::CannotSyncFile) + })?; + Ok(staging_path) +} + +/// Installs a staged rebuilt index at its final name. The rename is the +/// atomic commit point: the on-disk index is either the old one holding no +/// whole entry (whose walk re-runs the rebuild) or the complete rebuilt one, +/// never a mix of pages from both. +fn install_rebuilt_index( + staging_path: &str, index_path: &str, + partition_path: &str, +) -> Result<(), ServerError> { + fs::rename(staging_path, index_path).map_err(|source| { + error!( + from = %staging_path, + to = %index_path, + error = %source, + "failed to rename a rebuilt sparse index into place during recovery" + ); + ServerError::from(IggyError::CannotWriteToFile) + })?; + fsync_dir(partition_path) +} + +/// Makes renames and new files in `dir` durable. Synchronous like every +/// other mutation in this module (see [`FileScanner`]). +fn fsync_dir(dir: &str) -> Result<(), ServerError> { + fs::File::open(dir) + .and_then(|handle| handle.sync_all()) + .map_err(|source| { + error!( + dir, + error = %source, + "failed to fsync a directory during recovery" + ); + ServerError::from(IggyError::CannotSyncFile) + }) +} + +/// Moves a segment pair that recovery proved unreadable into a fresh +/// `.fenced.` directory -- the naming the partition-level +/// quarantine uses, so operators grep one pattern -- and seeds empty files +/// at the original names for the empty recovery to open. The bytes prove +/// nothing, yet they are the only copy of whatever the crash tore, so the +/// one verdict that would otherwise destroy data keeps it instead. +/// +/// Index first, log second on the reseed: recovery keys on `.log` stems and +/// sweeps orphaned indexes, so a crash between the two creates leaves only +/// states a later boot already understands (segment absent, or one orphan +/// index). +fn fence_unrecoverable_segment_files( + identity: PartitionIdentity<'_>, messages_path: &str, + index_path: &str, start_offset: u64, - messages_size: u64, - stream_id: usize, - topic_id: usize, - partition_id: usize, -) -> Result, ServerError> { +) -> Result<(), ServerError> { + let log_bytes = file_len(messages_path)?; + let index_bytes = file_len(index_path)?; + if log_bytes == 0 && index_bytes == 0 { + return Ok(()); + } + let mut fenced_dir = None; + for attempt in 0..FENCED_DIR_PROBE_LIMIT { + let candidate = format!("{}.fenced.{attempt}", identity.partition_path); + // `create_dir`, not `create_dir_all`: success is the claim on this + // suffix, and merging into an existing fence would mix evidence from + // two incidents. + match fs::create_dir(&candidate) { + Ok(()) => { + fenced_dir = Some(candidate); + break; + } + Err(source) if source.kind() == io::ErrorKind::AlreadyExists => {} + Err(source) => { + error!( + path = %candidate, + error = %source, + "failed to create a fence directory during recovery" + ); + return Err(IggyError::CannotWriteToFile.into()); + } + } + } + let Some(fenced_dir) = fenced_dir else { + error!( + partition_path = identity.partition_path, + "every fence directory suffix is taken; refusing to merge into one" + ); + return Err(IggyError::CannotWriteToFile.into()); + }; + let fenced_log = fenced_target(&fenced_dir, messages_path)?; + let fenced_index = fenced_target(&fenced_dir, index_path)?; + rename_into_fence(messages_path, &fenced_log)?; + rename_into_fence(index_path, &fenced_index)?; + seed_empty_file(index_path)?; + seed_empty_file(messages_path)?; + // The fence directory's new dirents, the partition directory's renames + // plus fresh files, and the parent's new fence-directory dirent. + fsync_dir(&fenced_dir)?; + fsync_dir(identity.partition_path)?; + if let Some(parent) = Path::new(identity.partition_path) + .parent() + .and_then(Path::to_str) + { + fsync_dir(parent)?; + } + warn!( + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, + start_offset, + fenced_log = %fenced_log.display(), + fenced_index = %fenced_index.display(), + log_bytes, + index_bytes, + "segment holds bytes but nothing in it decodes; moved the whole \ + .log/.index pair into the fence directory and recovered the segment \ + empty over fresh files" + ); + Ok(()) +} + +/// Destination of one fenced file: the fence directory plus the file's own +/// name, so the fenced copy stays greppable by its segment stem. +fn fenced_target(fenced_dir: &str, source_path: &str) -> Result { + Path::new(source_path).file_name().map_or_else( + || { + error!( + source_path, + "segment file path has no final component; cannot fence it" + ); + Err(IggyError::CannotWriteToFile.into()) + }, + |name| Ok(Path::new(fenced_dir).join(name)), + ) +} + +fn rename_into_fence(source_path: &str, target: &Path) -> Result<(), ServerError> { + match fs::rename(source_path, target) { + Ok(()) => Ok(()), + // A missing index beside a present log has nothing to move. + Err(source) if source.kind() == io::ErrorKind::NotFound => Ok(()), + Err(source) => { + error!( + from = source_path, + to = %target.display(), + error = %source, + "failed to move an unreadable segment file into its fence directory" + ); + Err(IggyError::CannotWriteToFile.into()) + } + } +} + +fn seed_empty_file(path: &str) -> Result<(), ServerError> { + fs::File::create(path) + .and_then(|file| file.sync_all()) + .map_err(|source| { + error!( + path, + error = %source, + "failed to seed a fresh empty segment file after fencing" + ); + ServerError::from(IggyError::CannotWriteToFile) + }) +} + +/// Index anchors for one segment: `(entry_count, first, last)`. +/// +/// A `.log` with NO `.index` beside it reads exactly like a 0-byte one, and +/// both belong on the index-less walk. It is an ordinary shape: +/// `SegmentStorage::new` creates the log before the index, so a crash or a +/// failed open between the two leaves precisely that pair, as does any +/// operator restore that drops an index. The reader's open is bare +/// `read(true)` and folds ENOENT into `CannotReadFile`, which propagates as a +/// plain `ServerError::Iggy` -- not a `PartitionRecoveryRefused` the caller +/// can fence -- so it would abort the whole boot for a segment the walk +/// rebuilds. Stat through the `NotFound`-lenient [`file_len`] first; every +/// other stat failure still fails stop there. +async fn load_index_anchors( + identity: PartitionIdentity<'_>, + index_path: &str, +) -> Result<(u64, Option, Option), ServerError> { + if file_len(index_path)? == 0 { + return Ok((0, None, None)); + } let reader = IggyIndexReader::new(index_path).await.map_err(|source| { error!( - stream_id, - topic_id, - partition_id, + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, path = %index_path, error = %source, "failed to open sparse index during recovery" ); source })?; + let entry_count = reader.entry_count().await.map_err(|source| { + error!( + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, + path = %index_path, + error = %source, + "failed to size sparse index during recovery" + ); + source + })?; let first = reader.load_first().await.map_err(|source| { error!( - stream_id, - topic_id, - partition_id, + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, path = %index_path, error = %source, "failed to read first sparse index entry during recovery" @@ -366,67 +910,176 @@ async fn recover_segment_bounds( })?; let last = reader.load_last().await.map_err(|source| { error!( - stream_id, - topic_id, - partition_id, + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, path = %index_path, error = %source, "failed to read last sparse index entry during recovery" ); source })?; + Ok((entry_count, first, last)) +} + +/// Derives a segment's readable bounds. `None` when the log holds no whole +/// batch at all (the caller recovers the segment as empty). +/// +/// With a whole index entry present, the last entry's `position` is only the +/// last flushed chunk's START byte, so the batch chain is walked from there to +/// prove where the segment really ends -- without `enforce_fsync` there is no +/// ordering barrier between the message write and the index write, and a tail +/// torn mid-flush would otherwise pass while `end_offset` claims offsets whose +/// bytes are incomplete. Without one, the log itself is walked from byte 0 and +/// the index is rebuilt from the batches found. Either way, bytes left past +/// the walked prefix go through the damage probe: a torn tail truncates, but +/// damage with intact batches after it -- or residue the probe cannot +/// classify within its limits -- refuses recovery. +#[allow(clippy::too_many_lines)] +async fn recover_segment_bounds( + identity: PartitionIdentity<'_>, + index_path: &str, + messages_path: &str, + start_offset: u64, + messages_size: u64, + scratch: &mut ScanScratch, +) -> Result, ServerError> { + let (entry_count, first, last) = load_index_anchors(identity, index_path).await?; match (first, last) { (Some(first), Some(last)) => { + // Interior entries were never validated before: a mis-strided + // index can decode to garbage entries that binary searches then + // trust. Monotonicity plus the walk's own anchor bound them: the + // walk below only accepts bounds when a whole batch decodes at + // the LAST entry's position, so ascending positions keep every + // surviving entry inside the truncated log. + validate_index_entries(identity, index_path, start_offset, entry_count, scratch)?; + + let messages = open_messages_file(identity, messages_path)?; + let mut scanner = FileScanner::new(&messages, messages_size, scratch); // The sparse index holds ONE entry per flushed chunk, pointing // at the chunk's FIRST batch -- `last.offset` is where the last // chunk STARTS, not where the segment ends (a whole journal // flushed as one chunk indexes only its first offset). Walk the // batch chain from that position to the file end to recover the - // true end offset; a header that no longer decodes marks a torn - // tail, which truncates the readable range to the last whole - // batch so the next append overwrites the torn bytes. - // Opened ONCE for the walk: the helper used to open the file per - // batch, which is an open + pread + close for every batch in the - // segment, synchronously, at boot. A failure to open a file that - // just stat'd walks nothing, which lands on the divergence refusal - // below rather than recovering an indexed segment as empty. - let messages = fs::File::open(messages_path).ok(); + // true end offset. let mut position = last.position; let mut end_offset = last.offset; let mut end_timestamp = last.timestamp; + let mut expected_offset = last.offset; let mut walked_any = false; - while let Some(messages) = messages.as_ref() - && position < messages_size - { - let Some(header) = read_batch_header(messages, position, messages_size) else { - break; + // TODO(hubcio): for batches that continue the chain exactly this + // indexed walk trusts the header decode alone (gap-opening + // batches checksum-verify below), so a torn flush that persisted + // the header page but zeroed the body is absorbed silently; the + // index-less walk below checksums every batch. Decide whether the + // indexed arm should checksum too (boot cost) or leave body rot + // to protocol-aware repair. + while position < messages_size { + let header = match scanner.peek_header(position) { + Ok(Some(header)) => header, + Ok(None) => break, + Err(source) => { + return Err(scan_read_failure(identity, messages_path, &source)); + } }; let extent = position.saturating_add(header.total_size() as u64); if extent > messages_size { break; } + // The anchor entry names the offset its chunk starts at, and + // batches inside one segment are contiguous from there -- + // except in one direction the server mints itself. Refuse + // only a REGRESSION: a lower offset re-adopted here regresses + // the partition's offset counter at bootstrap (re-minting + // already-served offsets on the next append) and one below + // the segment start underflows the recovered message count. + // A FORWARD gap can be ordinary boot output: a crash can + // persist the offset frontier ahead of the unsynced log, and + // the next boot then stamps `base_offset = frontier` into + // this same tail segment. But `base_offset` is covered by the + // batch checksum and an upward bit flip in an unverified + // header wears the same shape, so the gap-opening batch must + // checksum-verify before its offset is adopted (the legit + // frontier stamp is server-minted and checksums clean). + // Absorb a verified gap and keep serving; every offset + // adopted is one the primary durably promised. + if header.base_offset < expected_offset { + return Err( + identity.refusal(PartitionRecoveryRefusal::OffsetDiscontinuity { + start_offset, + expected_offset, + found_offset: header.base_offset, + position, + }), + ); + } + if header.base_offset > expected_offset { + let verifies = scanner + .slice_at(position, header.total_size()) + .map_err(|source| scan_read_failure(identity, messages_path, &source))? + .is_some_and(|batch| decode_batch_slice(batch).is_ok()); + if !verifies { + // Damage, not a frontier stamp: break and let the + // probe below classify the residue (a torn tail + // truncates, a verifying batch past it refuses). + break; + } + warn!( + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, + start_offset, + expected_offset, + found_offset = header.base_offset, + position, + "segment holds a forward offset gap in a byte-clean \ + chain; absorbing it (a restart most likely stamped \ + the restored offset frontier into this tail segment)" + ); + } if header.message_count > 0 { end_offset = header .base_offset .saturating_add(u64::from(header.message_count) - 1); end_timestamp = header.base_timestamp; + expected_offset = end_offset.saturating_add(1); } walked_any = true; position = extent; + if scanner.take_refilled() { + yield_to_reactor().await; + } } if !walked_any { - return Err(ServerError::RecoveredSegmentSizeDivergence { - stream_id, - topic_id, - partition_id, - start_offset, - end_offset: last.offset, - messages_size_bytes: messages_size, - indexed_size_bytes: last.position, - }); + return Err( + identity.refusal(PartitionRecoveryRefusal::IndexLogDivergence { + start_offset, + end_offset: last.offset, + messages_size_bytes: messages_size, + indexed_size_bytes: last.position, + }), + ); } - Ok(Some((first.timestamp, end_timestamp, end_offset, position))) + refuse_if_survivor_past_damage( + identity, + &mut scanner, + messages_path, + position, + messages_size, + Some(end_offset), + start_offset, + ) + .await?; + Ok(Some(WalkedBounds { + start_timestamp: first.timestamp, + end_timestamp, + end_offset, + messages_size: position, + index_size: entry_count * SPARSE_INDEX_ENTRY_SIZE as u64, + rebuilt_index: None, + })) } // No whole index entry, but the log holds bytes: recover the bounds by // WALKING the log from byte 0 instead of declaring the segment empty. @@ -438,42 +1091,57 @@ async fn recover_segment_bounds( // MID-CHAIN segment too, not just the tail. Recovering that as empty // then trips the contiguity guard and refuses the whole partition: // total serve loss (and offset reuse from 0) for a chain whose bytes - // are all present. The walk stops at the first header that does not - // decode or does not fit, which keeps the torn-tail truncation the - // indexed path performs. + // are all present. The walk keeps the torn-tail truncation the indexed + // path performs, and rebuilds the index from the batches it proves so + // a sealed segment does not pay a full-scan poll penalty forever. _ if messages_size > 0 => { - // Opened once, as above. Nothing walked means no whole batch, - // which is the `Ok(None)` the tail of this arm already returns. - let messages = fs::File::open(messages_path).ok(); + let messages = open_messages_file(identity, messages_path)?; + let mut scanner = FileScanner::new(&messages, messages_size, scratch); let mut position = 0u64; let mut start_timestamp = None; let mut end_offset = start_offset; let mut end_timestamp = 0; let mut expected_offset = start_offset; - let mut scratch = Vec::new(); - while let Some(messages) = messages.as_ref() - && position < messages_size - { - let Some(header) = read_batch_header(messages, position, messages_size) else { - break; + let mut rebuilt_index = Vec::new(); + let mut last_indexed_position: Option = None; + while position < messages_size { + let header = match scanner.peek_header(position) { + Ok(Some(header)) => header, + Ok(None) => break, + Err(source) => { + return Err(scan_read_failure(identity, messages_path, &source)); + } }; let extent = position.saturating_add(header.total_size() as u64); if extent > messages_size { break; } - // The FILENAME is the only trustworthy anchor once the index is - // gone, and `read_batch_header` checks a length, not a checksum. - // A torn header claiming an offset below `start_offset` would - // underflow the message count the caller derives; one claiming a - // jump above becomes this partition's counter, and the next - // prepare stamps a `base_offset` diverged from every peer. So - // the chain has to be contiguous from the filename onward, and - // the batch has to verify before its header is believed. - if header.base_offset != expected_offset - || !batch_verifies(messages, position, &header, &mut scratch) - { + // The FILENAME is the only trustworthy anchor once the index + // is gone, and the header decode checks a length, not a + // checksum. So the batch has to verify before its header is + // believed, and the chain has to be contiguous from the + // filename onward. + let verifies = scanner + .slice_at(position, header.total_size()) + .map_err(|source| scan_read_failure(identity, messages_path, &source))? + .is_some_and(|batch| decode_batch_slice(batch).is_ok()); + if !verifies { break; } + if header.base_offset != expected_offset { + // A batch that VERIFIES but does not continue the chain is + // durable data past a hole (or a duplicated range): the + // offsets in between are exactly what a truncation here + // would silently erase, so refuse instead. + return Err( + identity.refusal(PartitionRecoveryRefusal::OffsetDiscontinuity { + start_offset, + expected_offset, + found_offset: header.base_offset, + position, + }), + ); + } if header.message_count > 0 { end_offset = header .base_offset @@ -481,63 +1149,1577 @@ async fn recover_segment_bounds( end_timestamp = header.base_timestamp; start_timestamp.get_or_insert(header.base_timestamp); expected_offset = end_offset.saturating_add(1); + if last_indexed_position.is_none_or(|indexed| { + position.saturating_sub(indexed) >= REBUILT_INDEX_STRIDE_BYTES + }) { + push_index_entry( + &mut rebuilt_index, + header.base_offset, + header.base_timestamp, + position, + ); + last_indexed_position = Some(position); + } } position = extent; + if scanner.take_refilled() { + yield_to_reactor().await; + } } + refuse_if_survivor_past_damage( + identity, + &mut scanner, + messages_path, + position, + messages_size, + start_timestamp.map(|_| end_offset), + start_offset, + ) + .await?; let Some(start_timestamp) = start_timestamp else { - // Not one whole batch either: the bytes really are unusable, so + // Not one whole batch, and the probe above proved nothing + // decodable follows either: the bytes really are unusable, so // the caller's empty recovery is right after all. return Ok(None); }; warn!( - stream_id, - topic_id, - partition_id, + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, start_offset, messages_size, walked_size = position, - "sparse index holds no whole entry; recovered segment bounds by \ - walking the log instead of discarding it (the index repopulates \ - on the next flush, and polls take the index-less fallback until \ - then)" + rebuilt_entries = rebuilt_index.len() / SPARSE_INDEX_ENTRY_SIZE, + "sparse index holds no whole entry; recovered segment bounds \ + by walking the log and rebuilding its index from the walked \ + batches" ); - Ok(Some((start_timestamp, end_timestamp, end_offset, position))) + Ok(Some(WalkedBounds { + start_timestamp, + end_timestamp, + end_offset, + messages_size: position, + index_size: rebuilt_index.len() as u64, + rebuilt_index: Some(rebuilt_index), + })) } _ => Ok(None), } } -/// The batch command header at `position` in the messages file, or `None` -/// when the header does not fit / decode (`position` past the file, header -/// truncated, or garbage bytes). -/// Whether the batch at `position` decodes and passes its own `batch_checksum`. +/// Validates every whole index entry: the first must not claim an offset +/// below the segment's own start, and offsets and positions must strictly +/// ascend (the writer appends one entry per flushed chunk over a growing +/// log, and every chunk covers at least one message and one byte). /// -/// The index-less recovery walk trusts nothing else: without an index the only -/// anchors are the filename and the payload's self-description, and a torn -/// header is exactly what that walk exists to survive. -fn batch_verifies( - messages: &fs::File, - position: u64, - header: &BatchHeader, - scratch: &mut Vec, -) -> bool { - scratch.clear(); - scratch.resize(header.total_size(), 0); - if messages.read_exact_at(scratch, position).is_err() { - return false; - } - decode_batch_slice(scratch).is_ok() -} - -fn read_batch_header( - messages: &fs::File, - position: u64, - messages_size: u64, -) -> Option { - if position.checked_add(COMMAND_HEADER_SIZE as u64)? > messages_size { - return None; +/// Timestamps are deliberately NOT validated: a primary clock rewind across a +/// restart can legitimately regress persisted `base_timestamp` today, and the +/// lower-bound searches degrade gracefully on a non-monotone run, so refusing +/// would trade availability for nothing. +fn validate_index_entries( + identity: PartitionIdentity<'_>, + index_path: &str, + start_offset: u64, + entry_count: u64, + scratch: &mut ScanScratch, +) -> Result<(), ServerError> { + let file = fs::File::open(index_path).map_err(|source| { + error!( + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, + path = %index_path, + error = %source, + "failed to open sparse index for validation during recovery" + ); + ServerError::from(IggyError::CannotReadFile) + })?; + let window = &mut scratch.window; + let per_chunk_entries = SCAN_WINDOW_CAPACITY / SPARSE_INDEX_ENTRY_SIZE; + let mut previous: Option<(u64, u64)> = None; + let mut entry_index = 0u64; + let mut byte_position = 0u64; + while entry_index < entry_count { + let chunk_entries = (entry_count - entry_index).min(per_chunk_entries as u64); + // Bounded by the window capacity, so the try_from cannot fail. + let chunk_bytes = + usize::try_from(chunk_entries).unwrap_or(per_chunk_entries) * SPARSE_INDEX_ENTRY_SIZE; + window.resize(chunk_bytes, 0); + file.read_exact_at(&mut window[..], byte_position) + .map_err(|source| { + error!( + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, + path = %index_path, + error = %source, + "failed to read sparse index entries for validation during recovery" + ); + ServerError::from(IggyError::CannotReadFile) + })?; + for entry in window.chunks_exact(SPARSE_INDEX_ENTRY_SIZE) { + let entry_offset = read_u64_le(entry, 0); + let entry_position = read_u64_le(entry, 16); + if let Some((previous_offset, previous_position)) = previous + && (entry_offset <= previous_offset || entry_position <= previous_position) + { + return Err( + identity.refusal(PartitionRecoveryRefusal::IndexEntriesNotMonotone { + start_offset, + entry_index, + }), + ); + } + if previous.is_none() && entry_offset < start_offset { + return Err(identity.refusal( + PartitionRecoveryRefusal::IndexEntryBeforeSegmentStart { + start_offset, + first_entry_offset: entry_offset, + }, + )); + } + previous = Some((entry_offset, entry_position)); + entry_index += 1; + } + byte_position += chunk_bytes as u64; + } + Ok(()) +} + +/// Opens a segment's messages file for the recovery walk. Fail-stop on any +/// failure, mirroring `file_len`: recovery truncates to the bounds the walk +/// produces, so folding an open failure into "walked nothing" would route a +/// healthy indexed segment into a divergence refusal -- or an index-less one +/// into recover-as-empty, fencing the whole log out of service. +fn open_messages_file( + identity: PartitionIdentity<'_>, + messages_path: &str, +) -> Result { + fs::File::open(messages_path).map_err(|source| { + error!( + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, + path = %messages_path, + error = %source, + "failed to open a segment messages file during recovery" + ); + ServerError::from(IggyError::CannotReadFile) + }) +} + +/// A read failure inside the walk or probe is transient I/O, not evidence +/// about the bytes: fail stop rather than classify it as a torn tail, which +/// would truncate a healthy segment on an `EIO`. +fn scan_read_failure( + identity: PartitionIdentity<'_>, + path: &str, + source: &io::Error, +) -> ServerError { + error!( + stream_id = identity.stream_id, + topic_id = identity.topic_id, + partition_id = identity.partition_id, + path = %path, + error = %source, + "failed to read a segment file during the recovery walk" + ); + ServerError::from(IggyError::CannotReadFile) +} + +/// Classifies bytes left past the walked prefix, porting the WAL repair's +/// rule: truncation is sound only for a torn tail, and the question that +/// decides it is whether a complete entry follows the damage. A batch that +/// decodes, checksums, and plausibly extends the chain is durable data -- it +/// can only exist because an append completed after the damaged region -- so +/// discarding it would hide real loss behind a silent boot-time repair. +/// +/// The residue is deliberately NOT width-gated: a torn flush chunk is +/// bounded by the CHUNK, not by one record, and with `enforce_fsync = false` +/// delayed allocation routinely extends a file far past its written-back +/// pages, leaving hundreds of MiB of zeros behind one crash. That is the +/// canonical torn tail this module exists to truncate, so every residue is +/// probed whole. What bounds the probe instead is the per-candidate work +/// budget, whose exhaustion REFUSES and keeps the bytes rather than +/// truncating: past the limit the probe has proven nothing, and the cheapest +/// input to construct must never earn the destructive verdict. +async fn refuse_if_survivor_past_damage( + identity: PartitionIdentity<'_>, + scanner: &mut FileScanner<'_>, + messages_path: &str, + damage_position: u64, + messages_size: u64, + chain_end_offset: Option, + start_offset: u64, +) -> Result<(), ServerError> { + if damage_position >= messages_size { + // The walk consumed the whole file: nothing to classify. + return Ok(()); + } + let residue_bytes = messages_size - damage_position; + scanner.budget.grow_for_residue(residue_bytes); + match scanner + .probe_for_survivor(damage_position, chain_end_offset, start_offset) + .await + .map_err(|source| scan_read_failure(identity, messages_path, &source))? + { + ProbeOutcome::Survivor { position } => { + Err(identity.refusal(PartitionRecoveryRefusal::InteriorDamage { + start_offset, + damage_position, + survivor_position: position, + })) + } + ProbeOutcome::BudgetExhausted => Err(identity.refusal( + PartitionRecoveryRefusal::UnverifiedResidue { + start_offset, + damage_position, + residue_bytes, + candidates_examined: scanner.budget.spent_units, + budget_units: scanner.budget.limit_units, + }, + )), + ProbeOutcome::NoSurvivor => Ok(()), + } +} + +/// Verdict of the damage probe over the residue past the walked prefix. +/// `NoSurvivor` is the only verdict that permits truncation; running out of +/// budget is deliberately NOT folded into it, so a residue that is expensive +/// to scan refuses (keeping the bytes) instead of earning the destructive +/// outcome. +enum ProbeOutcome { + /// A complete, checksum-verifying batch starts at this position. + Survivor { position: u64 }, + /// The whole residue was scanned and nothing in it verifies. + NoSurvivor, + /// The scan budget ran out before the residue was classified. + BudgetExhausted, +} + +/// Forward-only buffered reads over one segment file for the recovery walk +/// and the damage probe. Parsing and checksumming happen against an in-memory +/// window so neither pays a syscall per batch -- the probe advances its +/// candidate one byte at a time, and per-candidate preads would turn one +/// damaged multi-GiB segment into a boot-length stall. +/// +/// Synchronous `std::fs` on purpose, like every mutation in this module: the +/// boot path's runtime sizes its blocking pool at zero and recovery must not +/// depend on `io_uring` opcode coverage. Only the sparse-index bound reads go +/// through the async `IggyIndexReader`. +struct FileScanner<'scan> { + file: &'scan fs::File, + file_len: u64, + window: &'scan mut Vec, + window_start: u64, + spill: &'scan mut Vec, + budget: &'scan mut ProbeBudget, + refilled: bool, +} + +impl<'scan> FileScanner<'scan> { + fn new(file: &'scan fs::File, file_len: u64, scratch: &'scan mut ScanScratch) -> Self { + let ScanScratch { + window, + spill, + probe_budget, + } = scratch; + window.clear(); + Self { + file, + file_len, + window, + window_start: 0, + spill, + budget: probe_budget, + refilled: false, + } + } + + /// True when the scanner hit disk since the last call. The async scan + /// loops yield to the reactor once per window of work on it: recovery + /// runs in front of the bootstrap barrier with the blocking pool sized + /// at zero, so an unyielding walk over a damaged multi-GiB chain would + /// pin the shard core -- signal handling included -- until it finishes. + fn take_refilled(&mut self) -> bool { + std::mem::take(&mut self.refilled) + } + + /// Bytes `[position, position + len)`, or `None` when they run past the + /// end of the file. + fn slice_at(&mut self, position: u64, len: usize) -> io::Result> { + let Some(end) = position.checked_add(len as u64) else { + return Ok(None); + }; + if end > self.file_len { + return Ok(None); + } + if len > SCAN_WINDOW_CAPACITY { + // A batch larger than the window: one direct read, no windowing. + // Callers only pass lengths from headers that already passed the + // plausibility cap, which is what bounds this resize. + self.spill.resize(len, 0); + self.file.read_exact_at(&mut self.spill[..], position)?; + self.refilled = true; + return Ok(Some(&self.spill[..])); + } + let window_end = self.window_start + self.window.len() as u64; + if position < self.window_start || end > window_end { + let fill = usize::try_from((self.file_len - position).min(SCAN_WINDOW_CAPACITY as u64)) + .unwrap_or(SCAN_WINDOW_CAPACITY); + self.window.resize(fill, 0); + self.file.read_exact_at(&mut self.window[..], position)?; + self.window_start = position; + self.refilled = true; + } + // In-window by the branch above, and the window is capacity-bounded, + // so the try_from cannot fail. + let start = usize::try_from(position - self.window_start).unwrap_or(0); + Ok(Some(&self.window[start..start + len])) + } + + /// The batch command header at `position`, or `None` when it does not fit + /// the file, does not decode (torn header, garbage bytes), or claims a + /// size no legal batch can reach. The size check runs BEFORE any caller + /// slices the claimed extent: an oversized claim cannot be a real batch, + /// so treating the header as undecodable is verdict-identical to reading + /// the claimed bytes and failing the verify, and it keeps one bit-flipped + /// length field from driving a claimed-size allocation and read. + fn peek_header(&mut self, position: u64) -> io::Result> { + let Some(bytes) = self.slice_at(position, COMMAND_HEADER_SIZE)? else { + return Ok(None); + }; + Ok(BatchHeader::decode(bytes) + .ok() + .filter(|header| header.total_size() as u64 <= MAX_RECOVERABLE_BATCH_BYTES)) + } + + /// Probes the residue for the first complete, checksum-verifying batch + /// starting after `damage_position`. + /// + /// Batch starts are byte-aligned (appends write exact-sized records with + /// no padding) and the damaged region's own lengths cannot be trusted, so + /// every byte offset is a candidate. Candidates are scanned inside the + /// loaded window and the window advances sequentially -- refilled at the + /// first candidate whose header no longer fits, re-reading at most one + /// header of overlap -- so each residue byte is read O(1) times instead + /// of once per candidate. The header decode pre-filters candidates + /// cheaply (204 reserved bytes must be zero), and offset sanity plus + /// length bounds run before a verify is paid, so the checksum only runs + /// on byte positions that already look like a plausible chain + /// continuation. + /// + /// Each candidate examined is charged one unit against the shared probe + /// budget -- examined, not verified: examining is flat-cost (zeros bail + /// on the undersized length, garbage on the first nonzero reserved + /// byte), so with the budget sized per residue byte an honest + /// front-to-back scan always fits, at any residue width, and total probe + /// work stays linear in the residue by construction. Window refills and + /// verify slices are deliberately not charged; they are already bounded + /// by the strictly-forward window advance and the plausibility cap on + /// claimed sizes. Exhaustion returns + /// [`ProbeOutcome::BudgetExhausted`], never `NoSurvivor`. + async fn probe_for_survivor( + &mut self, + damage_position: u64, + chain_end_offset: Option, + start_offset: u64, + ) -> io::Result { + let header_len = COMMAND_HEADER_SIZE as u64; + // The bytes AT the damage already failed to decode or verify, so the + // first candidate starts one past them. + let mut candidate = damage_position.saturating_add(1); + while candidate.saturating_add(header_len) <= self.file_len { + self.fill_window_at(candidate)?; + let window_end = self.window_start + self.window.len() as u64; + while candidate.saturating_add(header_len) <= window_end { + if !self.budget.charge_candidate() { + return Ok(ProbeOutcome::BudgetExhausted); + } + // In-window by the loop bound, and the window is + // capacity-bounded, so the try_from cannot fail. + let at = usize::try_from(candidate - self.window_start).unwrap_or(0); + if let Ok(header) = BatchHeader::decode(&self.window[at..at + COMMAND_HEADER_SIZE]) + { + let advances_chain = chain_end_offset + .map_or(header.base_offset >= start_offset, |chain_end| { + header.base_offset > chain_end + }); + let total_size = header.total_size(); + // The plausibility cap, not just the file length: with no + // width gate on the residue, this is what keeps one + // corrupted-upward length claim from driving a + // claimed-size spill allocation and read. + let fits = total_size as u64 <= MAX_RECOVERABLE_BATCH_BYTES + && candidate.saturating_add(total_size as u64) <= self.file_len; + if advances_chain && fits && header.message_count > 0 { + let batch = self.verify_slice(candidate, total_size)?; + if decode_batch_slice(batch).is_ok() { + return Ok(ProbeOutcome::Survivor { + position: candidate, + }); + } + } + } + candidate += 1; + } + if self.take_refilled() { + yield_to_reactor().await; + } + } + Ok(ProbeOutcome::NoSurvivor) + } + + /// Anchors the window at `position` unless the header there already sits + /// inside it. The probe's outer loop refills through this, so its + /// windows advance strictly forward. + fn fill_window_at(&mut self, position: u64) -> io::Result<()> { + let window_end = self.window_start + self.window.len() as u64; + if position >= self.window_start + && position.saturating_add(COMMAND_HEADER_SIZE as u64) <= window_end + { + return Ok(()); + } + let fill = usize::try_from((self.file_len - position).min(SCAN_WINDOW_CAPACITY as u64)) + .unwrap_or(SCAN_WINDOW_CAPACITY); + self.window.resize(fill, 0); + self.file.read_exact_at(&mut self.window[..], position)?; + self.window_start = position; + self.refilled = true; + Ok(()) + } + + /// Bytes `[position, position + len)` for one probe verification without + /// moving the scan window: an in-window slice costs no read, anything + /// else is one direct read into the spill buffer. The caller bounds + /// `len` against the file and the plausibility cap before calling, which + /// is what bounds the spill's growth. + fn verify_slice(&mut self, position: u64, len: usize) -> io::Result<&[u8]> { + let window_end = self.window_start + self.window.len() as u64; + let end = position.saturating_add(len as u64); + if position >= self.window_start && end <= window_end { + // In-window by the branch above, and the window is + // capacity-bounded, so the try_from cannot fail. + let at = usize::try_from(position - self.window_start).unwrap_or(0); + return Ok(&self.window[at..at + len]); + } + self.spill.resize(len, 0); + self.file.read_exact_at(&mut self.spill[..], position)?; + self.refilled = true; + Ok(&self.spill[..]) + } +} + +fn push_index_entry(rebuilt_index: &mut Vec, offset: u64, timestamp: u64, position: u64) { + rebuilt_index.extend_from_slice(&offset.to_le_bytes()); + rebuilt_index.extend_from_slice(×tamp.to_le_bytes()); + rebuilt_index.extend_from_slice(&position.to_le_bytes()); +} + +fn read_u64_le(bytes: &[u8], at: usize) -> u64 { + let mut raw = [0u8; 8]; + raw.copy_from_slice(&bytes[at..at + 8]); + u64::from_le_bytes(raw) +} + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + use configs::server::ServerSystemConfig; + use server_common::send_messages::{ + IggyMessage, IggyMessageHeader, IggyMessages, SendMessagesOwned, calculate_batch_checksum, + }; + use server_common::sharding::IggyNamespace; + use std::os::unix::fs::symlink; + use std::sync::Arc; + use tempfile::{TempDir, tempdir}; + + const STREAM_ID: usize = 1; + const TOPIC_ID: usize = 1; + const PARTITION_ID: usize = 1; + const SEGMENT_MAX_SIZE: u64 = 16 * 1024 * 1024; + const FIXTURE_TIMESTAMP: u64 = 1_700_000_000_000_000; + // Longer than one batch header and nonzero in the header's reserved + // region, so no prefix of it decodes as a batch. + const GARBAGE: [u8; 384] = [0xAB; 384]; + + fn test_config(tmp: &TempDir) -> ServerConfig { + let mut config = ServerConfig::default(); + // `ServerSystemConfig` is not `Clone`; build a fresh value and swap + // the whole `Arc`. + let system = ServerSystemConfig { + path: tmp.path().to_string_lossy().into_owned(), + ..ServerSystemConfig::default() + }; + config.system = Arc::new(system); + config + } + + fn prepare_partition_dir(config: &ServerConfig) -> String { + let partition_path = config + .system + .get_partition_path(STREAM_ID, TOPIC_ID, PARTITION_ID); + fs::create_dir_all(&partition_path).expect("create partition dir"); + partition_path + } + + // Batch header wire offsets the zero-padded fixture plants values at. + const HEADER_BATCH_LENGTH_OFFSET: usize = 32; + const HEADER_MESSAGE_COUNT_OFFSET: usize = 48; + + /// One fixed-width zero-padded record of the shape foreign storage + /// formats emit: a monotone u64 where the batch header keeps + /// `batch_length`, a nonzero u32 where it keeps `message_count`, zeros + /// everywhere else -- so its header decodes without any of it being a + /// batch. + fn zero_padded_record(claimed_batch_length: u64, sequence: u32) -> Vec { + let mut record = vec![0u8; COMMAND_HEADER_SIZE]; + record[HEADER_BATCH_LENGTH_OFFSET..HEADER_BATCH_LENGTH_OFFSET + 8] + .copy_from_slice(&claimed_batch_length.to_le_bytes()); + record[HEADER_MESSAGE_COUNT_OFFSET..HEADER_MESSAGE_COUNT_OFFSET + 4] + .copy_from_slice(&sequence.to_le_bytes()); + record + } + + /// One valid on-disk batch record: real message frames with their + /// per-message checksums, and the server-owned header fields stamped the + /// way persistence stamps them. + fn encoded_batch(base_offset: u64, message_count: usize) -> Vec { + encoded_batch_with_payload( + base_offset, + message_count, + &Bytes::from_static(b"segment-recovery-fixture"), + ) + } + + fn encoded_batch_with_payload( + base_offset: u64, + message_count: usize, + payload: &Bytes, + ) -> Vec { + let mut messages = IggyMessages::with_capacity(message_count); + for _ in 0..message_count { + messages.push(IggyMessage { + header: IggyMessageHeader { + origin_timestamp: FIXTURE_TIMESTAMP, + ..IggyMessageHeader::default() + }, + payload: payload.clone(), + user_headers: None, + }); + } + let namespace = IggyNamespace::new(STREAM_ID, TOPIC_ID, PARTITION_ID); + let SendMessagesOwned { mut header, blob } = + SendMessagesOwned::from_messages(namespace, &messages).expect("encode fixture batch"); + header.base_offset = base_offset; + header.base_timestamp = FIXTURE_TIMESTAMP; + header.batch_checksum = calculate_batch_checksum(&header, &blob); + let mut record = vec![0u8; header.total_size()]; + header.encode_into(&mut record[..COMMAND_HEADER_SIZE]); + record[COMMAND_HEADER_SIZE..].copy_from_slice(&blob); + record + } + + /// One sparse index entry, mirroring the `IggyIndexWriter` layout the + /// recovery reader expects. + fn index_entry(offset: u64, position: u64) -> Vec { + let mut entry = Vec::new(); + entry.extend_from_slice(&offset.to_le_bytes()); + entry.extend_from_slice(&FIXTURE_TIMESTAMP.to_le_bytes()); + entry.extend_from_slice(&position.to_le_bytes()); + entry + } + + /// Writes a segment's `.log` and `.index` fixtures and returns their + /// paths as `(messages_path, index_path)`. + fn write_segment( + config: &ServerConfig, + start_offset: u64, + log: &[u8], + index: &[u8], + ) -> (String, String) { + let messages_path = + config + .system + .get_messages_file_path(STREAM_ID, TOPIC_ID, PARTITION_ID, start_offset); + let index_path = + config + .system + .get_index_path(STREAM_ID, TOPIC_ID, PARTITION_ID, start_offset); + fs::write(&messages_path, log).expect("write log fixture"); + fs::write(&index_path, index).expect("write index fixture"); + (messages_path, index_path) + } + + fn len_of(path: &str) -> u64 { + fs::metadata(path).expect("stat fixture file").len() + } + + fn bytes_of(path: &str) -> Vec { + fs::read(path).expect("read fixture file") + } + + async fn recover(config: &ServerConfig) -> Result, ServerError> { + load_persisted_segments( + config, + STREAM_ID, + TOPIC_ID, + PARTITION_ID, + IggyByteSize::from(SEGMENT_MAX_SIZE), + &PartitionStats::default(), + ) + .await + } + + #[compio::test] + async fn given_torn_log_tail_when_recovering_should_truncate_files_to_walked_bounds() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let mut log = encoded_batch(0, 3); + let valid_len = log.len() as u64; + log.extend_from_slice(&GARBAGE); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index_entry(0, 0)); + + let recovered = recover(&config).await.expect("recover torn-tail segment"); + + assert_eq!(recovered.len(), 1); + let segment = &recovered[0].segment; + assert_eq!(segment.end_offset, 2); + assert_eq!(segment.size, IggyByteSize::from(valid_len)); + assert_eq!(segment.current_position, valid_len); + assert_eq!( + len_of(&messages_path), + valid_len, + "torn tail bytes must be gone from disk" + ); + assert_eq!(len_of(&index_path), SPARSE_INDEX_ENTRY_SIZE as u64); + } + + #[compio::test] + async fn given_torn_index_tail_when_recovering_should_floor_index_to_whole_entries() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let log = encoded_batch(0, 2); + let mut index = index_entry(0, 0); + index.extend_from_slice(&GARBAGE[..10]); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index); + + let recovered = recover(&config).await.expect("recover torn-index segment"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].segment.end_offset, 1); + assert_eq!( + len_of(&index_path), + SPARSE_INDEX_ENTRY_SIZE as u64, + "partial index entry must be gone from disk" + ); + assert_eq!(len_of(&messages_path), log.len() as u64); + } + + #[compio::test] + async fn given_no_recoverable_bytes_when_recovering_should_fence_files_and_seed_empty() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + let partition_path = prepare_partition_dir(&config); + let (messages_path, index_path) = write_segment(&config, 0, &GARBAGE, &GARBAGE[..10]); + + let recovered = recover(&config).await.expect("recover segment as empty"); + + assert_eq!(recovered.len(), 1); + let segment = &recovered[0].segment; + assert_eq!(segment.size, IggyByteSize::default()); + assert_eq!(segment.end_offset, 0); + assert_eq!(len_of(&messages_path), 0, "the served log must be empty"); + assert_eq!(len_of(&index_path), 0, "the served index must be empty"); + let fenced_dir = format!("{partition_path}.fenced.0"); + let fenced = |original: &str| { + Path::new(&fenced_dir).join(Path::new(original).file_name().expect("fixture file name")) + }; + assert_eq!( + fs::read(fenced(&messages_path)).expect("read fenced log"), + GARBAGE, + "the unreadable log bytes must survive in the fence directory" + ); + assert_eq!( + fs::read(fenced(&index_path)).expect("read fenced index"), + &GARBAGE[..10], + "the unreadable index bytes must survive in the fence directory" + ); + } + + #[compio::test] + async fn given_recovered_partition_when_recovering_again_should_change_nothing() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let mut log = encoded_batch(0, 3); + log.extend_from_slice(&GARBAGE); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index_entry(0, 0)); + + let first = recover(&config).await.expect("first recovery"); + let sizes_after_first = (len_of(&messages_path), len_of(&index_path)); + let bounds_after_first = (first[0].segment.end_offset, first[0].segment.size); + drop(first); + + let second = recover(&config).await.expect("second recovery"); + + assert_eq!( + (len_of(&messages_path), len_of(&index_path)), + sizes_after_first, + "a second recovery must not move the files" + ); + assert_eq!( + (second[0].segment.end_offset, second[0].segment.size), + bounds_after_first + ); + } + + #[compio::test] + async fn given_unopenable_index_when_recovering_should_fail_stop_without_truncating() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let mut log = encoded_batch(0, 1); + log.extend_from_slice(&GARBAGE); + let messages_path = + config + .system + .get_messages_file_path(STREAM_ID, TOPIC_ID, PARTITION_ID, 0); + let index_path = config + .system + .get_index_path(STREAM_ID, TOPIC_ID, PARTITION_ID, 0); + fs::write(&messages_path, &log).expect("write log fixture"); + // Self-referential symlink: every open or stat that follows it fails + // with ELOOP, root or not (unlike permission bits, which root + // bypasses). + symlink(&index_path, &index_path).expect("create self-referential index symlink"); + + // Recovery stats the index before opening it (a missing one routes to + // the index-less walk), so an ELOOP surfaces from the stat. + let error = recover(&config) + .await + .err() + .expect("an unstattable index must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::Iggy(inner) if matches!(**inner, IggyError::CannotReadFileMetadata) + ), + "expected CannotReadFileMetadata, got {error:?}" + ); + assert_eq!( + bytes_of(&messages_path), + log, + "fail-stop must leave the log untouched" + ); + } + + #[compio::test] + async fn given_unopenable_log_when_recovering_index_less_should_fail_stop_without_truncating() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let messages_path = + config + .system + .get_messages_file_path(STREAM_ID, TOPIC_ID, PARTITION_ID, 0); + let index_path = config + .system + .get_index_path(STREAM_ID, TOPIC_ID, PARTITION_ID, 0); + fs::write(&index_path, &GARBAGE[..10]).expect("write torn index fixture"); + // See the index variant above; the log stem is still collected by the + // directory sweep, so recovery reaches the stat and must fail stop + // there instead of recovering the segment as empty. + symlink(&messages_path, &messages_path).expect("create self-referential log symlink"); + + let error = recover(&config) + .await + .err() + .expect("an unstattable log must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::Iggy(inner) if matches!(**inner, IggyError::CannotReadFileMetadata) + ), + "expected CannotReadFileMetadata, got {error:?}" + ); + assert_eq!( + len_of(&index_path), + 10, + "fail-stop must leave the torn index untouched" + ); + assert!( + fs::symlink_metadata(&messages_path) + .expect("lstat log symlink") + .file_type() + .is_symlink(), + "fail-stop must leave the log symlink in place" + ); + } + + #[compio::test] + async fn given_clean_segment_when_recovering_should_leave_files_untouched() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let log = encoded_batch(0, 4); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index_entry(0, 0)); + + let recovered = recover(&config).await.expect("recover clean segment"); + + assert_eq!(recovered.len(), 1); + let segment = &recovered[0].segment; + assert_eq!(segment.end_offset, 3); + assert!(!segment.sealed, "the tail segment must accept writes"); + assert_eq!(len_of(&messages_path), log.len() as u64); + assert_eq!(len_of(&index_path), SPARSE_INDEX_ENTRY_SIZE as u64); + } + + #[compio::test] + async fn given_torn_mid_chain_segment_when_recovering_should_truncate_it_too() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + // Sealed segment holding offsets 0..=2 plus a garbage tail, then the + // tail segment holding offsets 3..=4. + let mut sealed_log = encoded_batch(0, 3); + let sealed_valid_len = sealed_log.len() as u64; + sealed_log.extend_from_slice(&GARBAGE); + let (sealed_messages_path, _sealed_index_path) = + write_segment(&config, 0, &sealed_log, &index_entry(0, 0)); + let tail_log = encoded_batch(3, 2); + let (tail_messages_path, _tail_index_path) = + write_segment(&config, 3, &tail_log, &index_entry(3, 0)); + + let recovered = recover(&config).await.expect("recover two-segment chain"); + + assert_eq!(recovered.len(), 2); + assert!(recovered[0].segment.sealed); + assert_eq!(recovered[0].segment.end_offset, 2); + assert!(!recovered[1].segment.sealed); + assert_eq!(recovered[1].segment.end_offset, 4); + assert_eq!( + len_of(&sealed_messages_path), + sealed_valid_len, + "a mid-chain torn tail must be truncated too" + ); + assert_eq!(len_of(&tail_messages_path), tail_log.len() as u64); + } + + #[compio::test] + async fn given_valid_batch_after_damage_when_recovering_should_refuse_and_preserve_files() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let mut log = encoded_batch(0, 3); + log.extend_from_slice(&GARBAGE); + log.extend_from_slice(&encoded_batch(3, 1)); + let index = index_entry(0, 0); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index); + + let error = recover(&config) + .await + .err() + .expect("a surviving batch past damage must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::InteriorDamage { .. }, + .. + } + ), + "expected an interior-damage refusal, got {error:?}" + ); + assert_eq!( + bytes_of(&messages_path), + log, + "a refusal must leave the log byte-identical" + ); + assert_eq!( + bytes_of(&index_path), + index, + "a refusal must leave the index byte-identical" + ); + } + + #[compio::test] + async fn given_garbage_head_with_valid_batch_later_when_recovering_should_refuse() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let mut log = GARBAGE.to_vec(); + log.extend_from_slice(&encoded_batch(5, 1)); + let (messages_path, index_path) = write_segment(&config, 0, &log, &GARBAGE[..10]); + + let error = recover(&config) + .await + .err() + .expect("a lost head with valid batches later must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::InteriorDamage { .. }, + .. + } + ), + "expected an interior-damage refusal, got {error:?}" + ); + assert_eq!(bytes_of(&messages_path), log); + assert_eq!(bytes_of(&index_path), &GARBAGE[..10]); + } + + #[compio::test] + async fn given_offset_gap_after_valid_batches_when_recovering_index_less_should_refuse() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let mut log = encoded_batch(0, 2); + log.extend_from_slice(&encoded_batch(5, 1)); + let (messages_path, _index_path) = write_segment(&config, 0, &log, &GARBAGE[..10]); + + let error = recover(&config) + .await + .err() + .expect("an offset gap inside one segment must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::OffsetDiscontinuity { + expected_offset: 2, + found_offset: 5, + .. + }, + .. + } + ), + "expected an offset-discontinuity refusal, got {error:?}" + ); + assert_eq!(bytes_of(&messages_path), log); + } + + #[compio::test] + async fn given_multi_batch_torn_tail_when_recovering_index_less_should_truncate_at_break() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let mut log = encoded_batch(0, 2); + log.extend_from_slice(&encoded_batch(2, 2)); + let valid_len = log.len() as u64; + log.extend_from_slice(&GARBAGE); + let (messages_path, index_path) = write_segment(&config, 0, &log, &GARBAGE[..10]); + + let recovered = recover(&config).await.expect("recover multi-batch tail"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].segment.end_offset, 3); + assert_eq!( + len_of(&messages_path), + valid_len, + "the walk must keep every whole batch before the tear" + ); + assert_eq!( + bytes_of(&index_path), + index_entry(0, 0), + "the index must be rebuilt from the walked batches" + ); + assert!( + fs::metadata(format!("{index_path}{STAGING_SUFFIX}")).is_err(), + "the staged rebuild must be renamed into place, not copied" + ); + + let sizes_after_first = (len_of(&messages_path), len_of(&index_path)); + drop(recovered); + let second = recover(&config).await.expect("second recovery"); + assert_eq!(second[0].segment.end_offset, 3); + assert_eq!( + (len_of(&messages_path), len_of(&index_path)), + sizes_after_first, + "recovering over a rebuilt index must be a no-op" + ); + } + + #[compio::test] + async fn given_holed_chain_when_recovering_should_leave_every_segment_untouched() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + // First segment carries a torn tail that WOULD truncate; the hole to + // the next segment must refuse the chain before that happens. + let mut first_log = encoded_batch(0, 3); + first_log.extend_from_slice(&GARBAGE); + let first_index = index_entry(0, 0); + let (first_messages_path, first_index_path) = + write_segment(&config, 0, &first_log, &first_index); + let next_log = encoded_batch(10, 1); + let next_index = index_entry(10, 0); + let (next_messages_path, next_index_path) = + write_segment(&config, 10, &next_log, &next_index); + + let error = recover(&config) + .await + .err() + .expect("a holed chain must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::Hole { .. }, + .. + } + ), + "expected a hole refusal, got {error:?}" + ); + assert_eq!( + bytes_of(&first_messages_path), + first_log, + "a refused chain must leave even truncation candidates byte-identical" + ); + assert_eq!(bytes_of(&first_index_path), first_index); + assert_eq!(bytes_of(&next_messages_path), next_log); + assert_eq!(bytes_of(&next_index_path), next_index); + } + + #[compio::test] + async fn given_non_monotone_index_entries_when_recovering_should_refuse() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let batch0 = encoded_batch(0, 1); + let batch1 = encoded_batch(1, 1); + let batch2 = encoded_batch(2, 1); + let mut log = batch0.clone(); + log.extend_from_slice(&batch1); + log.extend_from_slice(&batch2); + let last_position = (batch0.len() + batch1.len()) as u64; + // Interior garbage entry: ascending against its predecessor, so only + // the offset regression to the (valid) last entry exposes it. + let mut index = index_entry(0, 0); + index.extend_from_slice(&index_entry(50, 100)); + index.extend_from_slice(&index_entry(2, last_position)); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index); + + let error = recover(&config) + .await + .err() + .expect("a non-monotone index must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::IndexEntriesNotMonotone { + entry_index: 2, + .. + }, + .. + } + ), + "expected a non-monotone index refusal, got {error:?}" + ); + assert_eq!(bytes_of(&messages_path), log); + assert_eq!(bytes_of(&index_path), index); + } + + #[compio::test] + async fn given_index_entry_below_segment_start_when_recovering_should_refuse() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let log = encoded_batch(5, 1); + let index = index_entry(3, 0); + let (messages_path, index_path) = write_segment(&config, 5, &log, &index); + + let error = recover(&config) + .await + .err() + .expect("an index claiming offsets below the segment start must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::IndexEntryBeforeSegmentStart { + first_entry_offset: 3, + .. + }, + .. + } + ), + "expected a below-start index refusal, got {error:?}" + ); + assert_eq!(bytes_of(&messages_path), log); + assert_eq!(bytes_of(&index_path), index); + } + + #[compio::test] + async fn given_index_less_wide_batches_when_recovering_should_rebuild_strided_index() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + // First batch alone crosses the rebuild stride, so the second batch + // must get its own entry at the first batch's total size. + let wide = encoded_batch_with_payload(0, 1, &Bytes::from(vec![0x42u8; 70 * 1024])); + let narrow = encoded_batch(1, 1); + let mut log = wide.clone(); + log.extend_from_slice(&narrow); + let (_messages_path, index_path) = write_segment(&config, 0, &log, &GARBAGE[..10]); + + let recovered = recover(&config).await.expect("recover wide-batch segment"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].segment.end_offset, 1); + let rebuilt = bytes_of(&index_path); + assert_eq!(rebuilt.len(), 2 * SPARSE_INDEX_ENTRY_SIZE); + let entry = |index: usize| { + let at = index * SPARSE_INDEX_ENTRY_SIZE; + ( + read_u64_le(&rebuilt, at), + read_u64_le(&rebuilt, at + 8), + read_u64_le(&rebuilt, at + 16), + ) + }; + assert_eq!(entry(0), (0, FIXTURE_TIMESTAMP, 0)); + assert_eq!(entry(1), (1, FIXTURE_TIMESTAMP, wide.len() as u64)); + } + + #[compio::test] + async fn given_zero_padded_records_when_probing_should_scan_whole_residue_and_recover_empty() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + let partition_path = prepare_partition_dir(&config); + // Torn index forces the index-less walk, and the garbage head keeps + // it from decoding anything, so the whole file is probe residue. + // Each record's header decodes and claims an 8 KiB batch that fits, + // so aligned candidates pay a (fast-failing) verify -- and none of + // that exhausts the residue-sized budget, so the probe classifies + // the whole file as survivor-free and the empty recovery fences it. + let mut log = GARBAGE.to_vec(); + for record in 0..128u32 { + log.extend_from_slice(&zero_padded_record( + 8 * 1024 + u64::from(record), + record + 1, + )); + } + let (messages_path, _index_path) = write_segment(&config, 0, &log, &GARBAGE[..10]); + + let recovered = recover(&config) + .await + .expect("a survivor-free residue must recover as empty, not refuse"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].segment.size, IggyByteSize::default()); + assert_eq!(len_of(&messages_path), 0, "the served log must be empty"); + let fenced_log = Path::new(&format!("{partition_path}.fenced.0")).join( + Path::new(&messages_path) + .file_name() + .expect("log file name"), + ); + assert_eq!( + fs::read(fenced_log).expect("read fenced log"), + log, + "the unclassifiable bytes must survive in the fence directory" + ); + } + + #[compio::test] + async fn given_wide_zeros_residue_when_recovering_should_truncate_at_break() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + // The canonical torn flush chunk: a crash under `enforce_fsync = + // false` leaves the file extended far past its written-back pages, + // reading as zeros -- residue bounded by the CHUNK (up to a whole + // segment), not by one record. No survivor decodes anywhere in it, + // so recovery must truncate to the walked prefix, at any residue + // width and regardless of any configured message size. + let mut log = encoded_batch(0, 3); + let valid_len = log.len() as u64; + log.resize(log.len() + 512 * 1024, 0); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index_entry(0, 0)); + + let recovered = recover(&config) + .await + .expect("a zero-filled torn flush chunk must truncate, not refuse"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].segment.end_offset, 2); + assert_eq!( + len_of(&messages_path), + valid_len, + "the zero-filled residue must be gone from disk" + ); + assert_eq!(len_of(&index_path), SPARSE_INDEX_ENTRY_SIZE as u64); + } + + #[test] + fn probe_budget_charges_per_candidate_across_probes() { + let mut budget = ProbeBudget::default(); + budget.grow_for_residue(4); + for _ in 0..8 { + assert!(budget.charge_candidate(), "honest scans fit the budget"); + } + assert!( + !budget.charge_candidate(), + "the ninth candidate against a 4-byte residue must exhaust" + ); + // A later probe in the same load widens the shared limit; spent + // units carry over rather than resetting per probe. + budget.grow_for_residue(2); + assert!(budget.charge_candidate()); + } + + #[compio::test] + async fn given_exhausted_probe_budget_when_classifying_residue_should_refuse() { + let tmp = tempdir().expect("tempdir"); + let mut log = encoded_batch(0, 2); + let valid_len = log.len() as u64; + log.extend_from_slice(&GARBAGE); + let messages_path = tmp.path().join("00000000000000000000.log"); + fs::write(&messages_path, &log).expect("write log fixture"); + let file = fs::File::open(&messages_path).expect("open log fixture"); + let partition_path = tmp.path().to_string_lossy().into_owned(); + let identity = PartitionIdentity { + partition_path: &partition_path, + stream_id: STREAM_ID, + topic_id: TOPIC_ID, + partition_id: PARTITION_ID, + }; + // No once-through scan can exhaust a residue-sized budget, so the + // exhaustion path is a tripwire for probe defects that re-examine + // candidates. Simulate one by pre-spending the shared budget past + // anything this residue can grow it by. + let mut scratch = ScanScratch::default(); + scratch.probe_budget.spent_units = u64::MAX / 2; + let mut scanner = FileScanner::new(&file, log.len() as u64, &mut scratch); + + let error = refuse_if_survivor_past_damage( + identity, + &mut scanner, + &messages_path.to_string_lossy(), + valid_len, + log.len() as u64, + Some(1), + 0, + ) + .await + .expect_err("an exhausted budget must refuse instead of truncating"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::UnverifiedResidue { + damage_position, + residue_bytes, + candidates_examined, + budget_units, + .. + }, + .. + } if *damage_position == valid_len + && *residue_bytes == GARBAGE.len() as u64 + && candidates_examined > budget_units + ), + "expected a budget-exhausted refusal, got {error:?}" + ); + assert_eq!( + bytes_of(&messages_path.to_string_lossy()), + log, + "a refusal must leave the log byte-identical" + ); + } + + #[compio::test] + async fn given_indexed_offset_regression_when_recovering_should_refuse_without_panicking() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + // A decodable batch claiming offsets below the segment start: absorbed, + // it would regress the recovered end offset below the start and + // underflow the message-count arithmetic. + let mut log = encoded_batch(100, 1); + log.extend_from_slice(&encoded_batch(5, 1)); + let index = index_entry(100, 0); + let (messages_path, index_path) = write_segment(&config, 100, &log, &index); + + let error = recover(&config) + .await + .err() + .expect("an indexed walk hitting a regressed offset must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::OffsetDiscontinuity { + expected_offset: 101, + found_offset: 5, + .. + }, + .. + } + ), + "expected an offset-discontinuity refusal, got {error:?}" + ); + assert_eq!(bytes_of(&messages_path), log); + assert_eq!(bytes_of(&index_path), index); + } + + #[compio::test] + async fn given_indexed_forward_offset_gap_when_recovering_should_absorb_and_serve() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + // The shape the server's own boot path mints: a crash persists the + // offset frontier ahead of the unsynced log, and the next boot + // appends `base_offset = frontier` into the existing tail segment. + // Byte-clean, fully decodable, index intact -- must serve, not + // refuse. + let mut log = encoded_batch(0, 2); + log.extend_from_slice(&encoded_batch(5, 1)); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index_entry(0, 0)); + + let recovered = recover(&config) + .await + .expect("a forward offset gap in a byte-clean indexed chain must recover"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].segment.end_offset, 5); + assert_eq!( + bytes_of(&messages_path), + log, + "absorbing the gap must leave the log byte-identical" + ); + assert_eq!(bytes_of(&index_path), index_entry(0, 0)); + } + + #[compio::test] + async fn given_indexed_forward_gap_with_failing_checksum_when_recovering_should_truncate_at_break() + { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + // Wears the frontier-stamp shape (header decodes, base_offset ahead + // of the chain) but fails the batch checksum: a real frontier stamp + // is server-minted and checksums clean, so this is damage and must + // never be adopted as the new offset frontier. + let mut log = encoded_batch(0, 2); + let valid_len = log.len() as u64; + let mut corrupt = encoded_batch(5, 1); + corrupt[COMMAND_HEADER_SIZE + 4] ^= 0xFF; + log.extend_from_slice(&corrupt); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index_entry(0, 0)); + + let recovered = recover(&config) + .await + .expect("a failing gap batch with nothing verifying past it must truncate"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].segment.end_offset, 1); + assert_eq!( + len_of(&messages_path), + valid_len, + "the unverified gap batch must be gone from disk" + ); + assert_eq!(len_of(&index_path), SPARSE_INDEX_ENTRY_SIZE as u64); + } + + #[compio::test] + async fn given_indexed_forward_gap_with_failing_checksum_before_valid_batch_when_recovering_should_refuse() + { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + // A verifying batch past the failing gap batch is durable data: + // truncating there would erase it, so recovery must refuse and keep + // every byte. + let mut log = encoded_batch(0, 2); + let mut corrupt = encoded_batch(5, 1); + corrupt[COMMAND_HEADER_SIZE + 4] ^= 0xFF; + log.extend_from_slice(&corrupt); + log.extend_from_slice(&encoded_batch(6, 1)); + let index = index_entry(0, 0); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index); + + let error = recover(&config) + .await + .err() + .expect("a verifying batch past the failing gap batch must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::InteriorDamage { .. }, + .. + } + ), + "expected an interior-damage refusal, got {error:?}" + ); + assert_eq!(bytes_of(&messages_path), log); + assert_eq!(bytes_of(&index_path), index); + } + + #[compio::test] + async fn given_bit_flipped_batch_length_when_recovering_should_truncate_at_break() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + // A corrupted-upward length claim (the classic all-ones flip) is not + // a plausible batch: the walk must break at the header itself, never + // sizing an allocation or a read by what the header claims. + let mut log = encoded_batch(0, 2); + let valid_len = log.len() as u64; + log.extend_from_slice(&zero_padded_record(0xFFFF_FFFF, 1)); + let (messages_path, _index_path) = write_segment(&config, 0, &log, &GARBAGE[..10]); + + let recovered = recover(&config) + .await + .expect("an implausible length claim in the tail must truncate, not refuse"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].segment.end_offset, 1); + assert_eq!( + len_of(&messages_path), + valid_len, + "the walk must break at the implausible header and truncate there" + ); + } + + #[compio::test] + async fn given_bit_flipped_batch_length_before_valid_batch_when_recovering_should_refuse() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let mut log = encoded_batch(0, 2); + let valid_len = log.len() as u64; + log.extend_from_slice(&zero_padded_record(0xFFFF_FFFF, 1)); + log.extend_from_slice(&encoded_batch(2, 1)); + let (messages_path, _index_path) = write_segment(&config, 0, &log, &GARBAGE[..10]); + + let error = recover(&config) + .await + .err() + .expect("a surviving batch past an implausible header must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::InteriorDamage { + damage_position, + survivor_position, + .. + }, + .. + } if *damage_position == valid_len + && *survivor_position == valid_len + COMMAND_HEADER_SIZE as u64 + ), + "expected an interior-damage refusal, got {error:?}" + ); + assert_eq!(bytes_of(&messages_path), log); + } + + #[compio::test] + async fn given_refused_chain_when_index_rebuilt_should_stage_without_touching_final() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + // The index-less first segment wants a rebuild; the hole to the next + // segment refuses the chain in pass B, before any install. + let first_log = encoded_batch(0, 2); + let first_index = GARBAGE[..10].to_vec(); + let (first_messages_path, first_index_path) = + write_segment(&config, 0, &first_log, &first_index); + write_segment(&config, 10, &encoded_batch(10, 1), &index_entry(10, 0)); + + let error = recover(&config) + .await + .err() + .expect("a holed chain must refuse recovery"); + + assert!( + matches!( + &error, + ServerError::PartitionRecoveryRefused { + reason: PartitionRecoveryRefusal::Hole { .. }, + .. + } + ), + "expected a hole refusal, got {error:?}" + ); + assert_eq!( + bytes_of(&format!("{first_index_path}{STAGING_SUFFIX}")), + index_entry(0, 0), + "pass A must stage the rebuilt index beside the final one" + ); + assert_eq!( + bytes_of(&first_index_path), + first_index, + "a refusal must leave the final index byte-identical" + ); + assert_eq!(bytes_of(&first_messages_path), first_log); + } + + #[compio::test] + async fn given_orphaned_index_staging_when_recovering_should_sweep_it() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let log = encoded_batch(0, 2); + let (messages_path, index_path) = write_segment(&config, 0, &log, &index_entry(0, 0)); + let staging_path = format!("{index_path}{STAGING_SUFFIX}"); + fs::write(&staging_path, GARBAGE).expect("write orphaned staging fixture"); + + let recovered = recover(&config).await.expect("recover clean segment"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].segment.end_offset, 1); + assert!( + fs::metadata(&staging_path).is_err(), + "an orphaned staging file must be swept at boot" + ); + assert_eq!(len_of(&messages_path), log.len() as u64); + assert_eq!(len_of(&index_path), SPARSE_INDEX_ENTRY_SIZE as u64); + } + #[compio::test] + async fn given_absent_index_when_recovering_should_walk_index_less_and_rebuild() { + let tmp = tempdir().expect("tempdir"); + let config = test_config(&tmp); + prepare_partition_dir(&config); + let log = encoded_batch(0, 4); + let messages_path = + config + .system + .get_messages_file_path(STREAM_ID, TOPIC_ID, PARTITION_ID, 0); + let index_path = config + .system + .get_index_path(STREAM_ID, TOPIC_ID, PARTITION_ID, 0); + fs::write(&messages_path, &log).expect("write log fixture"); + + let recovered = recover(&config) + .await + .expect("a log with no index beside it must recover, not abort the boot"); + + assert_eq!(recovered.len(), 1); + assert_eq!(recovered[0].segment.end_offset, 3); + assert_eq!(len_of(&messages_path), log.len() as u64); + assert_eq!( + len_of(&index_path), + SPARSE_INDEX_ENTRY_SIZE as u64, + "the walk must install a rebuilt index over the missing one" + ); } - let mut header_bytes = [0u8; COMMAND_HEADER_SIZE]; - messages.read_exact_at(&mut header_bytes, position).ok()?; - BatchHeader::decode(&header_bytes).ok() } diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs index 34c5616326..f026193c43 100644 --- a/core/server/src/server_error.rs +++ b/core/server/src/server_error.rs @@ -162,21 +162,27 @@ pub enum ServerError { expected: u128, found: u128, }, - // Per-partition, not fatal: the boot path fences this one group (quarantines - // its segment files and materialises it fresh) instead of taking the node - // down for one damaged local chain. The shapes it reports are exactly what a - // failed state-transfer quarantine leaves behind, and the rebuild recovers - // the data from a peer. + // Per-partition, not fatal: the boot path fences this one group instead of + // taking the node down for one damaged local chain. Only STRUCTURAL + // refusals route here -- shapes where the local files contradict + // themselves, so a retried boot cannot help. Transient recovery I/O + // failures (stat, open, read, truncate, fsync) stay node-fatal on purpose: + // a retried boot can still serve that partition, while fencing it would + // quarantine healthy data. #[error( - "partition {stream_id}/{topic_id}/{partition_id} at {dir} recovered an \ - unusable segment chain: {reason}" + "partition {stream_id}/{topic_id}/{partition_id} at {dir} refused segment \ + recovery: {reason}. Boot moves the partition's segment files into a \ + sibling `.fenced.N` directory and keeps them; with peer \ + replicas the partition is rebuilt empty and refilled by state transfer, \ + while with replica_count = 1 (or when the quarantine itself fails) it \ + is tombstoned and not served" )] - PartitionChainRefused { + PartitionRecoveryRefused { dir: PathBuf, stream_id: usize, topic_id: usize, partition_id: usize, - reason: PartitionChainRefusal, + reason: PartitionRecoveryRefusal, }, #[error( "shard {shard_id} aborted while waiting for shard-0 to broadcast the metadata \ @@ -243,18 +249,6 @@ pub enum ServerError { #[source] source: std::io::Error, }, - #[error( - "recovered segment for stream {stream_id}, topic {topic_id}, partition {partition_id} at start_offset {start_offset} has message/index divergence (messages_size={messages_size_bytes}, indexed_size={indexed_size_bytes}, end_offset={end_offset}); recovery aborted before opening listeners. Restore the partition from a healthy replica or snapshot, or move the segment aside for offline repair before restarting." - )] - RecoveredSegmentSizeDivergence { - stream_id: usize, - topic_id: usize, - partition_id: usize, - start_offset: u64, - end_offset: u64, - messages_size_bytes: u64, - indexed_size_bytes: u64, - }, #[error( "failed to load persisted {consumer_kind} offsets for stream {stream_id}, topic {topic_id}, partition {partition_id} from {path}" )] @@ -281,14 +275,18 @@ pub enum ServerError { ShardJoinFailures { failures: Vec }, } -/// Why a recovered segment chain cannot be served. +/// Why a partition's recovered segments cannot be served. /// -/// Both shapes mean the same thing operationally -- the local files do not form -/// a chain this replica can serve -- but they are distinguished because they -/// point at different causes: an empty non-tail segment is a failed rebuild's -/// orphan pairing, a hole is a stray or half-unlinked file. +/// Every shape here is structural -- the local files contradict themselves or +/// each other -- but they are distinguished because they point at different +/// causes, and not all of them are at-rest corruption: an empty non-tail +/// segment is a failed rebuild's orphan pairing, a hole is a stray or +/// half-unlinked file, interior damage is bit rot (or a resurrected tail +/// appended over), a divergent index is a mis-strided or foreign write, and +/// offsets that do not continue the chain can be minted into byte-clean files +/// by an upstream crash window as well as by damage. #[derive(Debug)] -pub enum PartitionChainRefusal { +pub enum PartitionRecoveryRefusal { EmptyNonTailSegment { empty_start: u64, next_start: u64, @@ -298,9 +296,69 @@ pub enum PartitionChainRefusal { previous_end: u64, next_start: u64, }, + /// The index holds entries but no whole batch decodes where its last + /// entry points, so index and log describe different files. + IndexLogDivergence { + start_offset: u64, + end_offset: u64, + messages_size_bytes: u64, + indexed_size_bytes: u64, + }, + /// A complete, checksum-verifying batch survives PAST bytes that do not + /// decode. A torn tail has nothing after it, so this is interior damage, + /// and truncating at it would silently discard the surviving batches. + InteriorDamage { + start_offset: u64, + damage_position: u64, + survivor_position: u64, + }, + /// Bytes past the walked prefix that the damage probe could not + /// classify: it ran out of work budget before proving or disproving a + /// survivor. The budget is sized so a front-to-back scan of every + /// residue in the load always fits, so exhaustion means candidate + /// offsets were re-examined -- a probe defect, not an at-rest shape. + /// Truncation is only ever sound for a proven torn tail, so giving up + /// keeps the bytes. The residue width is diagnostic only; it is not a + /// gate. + UnverifiedResidue { + start_offset: u64, + damage_position: u64, + residue_bytes: u64, + candidates_examined: u64, + budget_units: u64, + }, + /// A batch does not continue the offset chain, so offsets are not + /// contiguous inside one segment file. The cause is not necessarily + /// at-rest damage: a crash window that leaves the durable offset + /// frontier past the recovered end offset stamps the same shape into + /// byte-clean files. + OffsetDiscontinuity { + start_offset: u64, + expected_offset: u64, + found_offset: u64, + position: u64, + }, + /// Index entries must ascend in offset and position (they are appended, + /// one per flushed chunk, over a growing log); a regression means the + /// file was written mis-strided or over foreign bytes. + IndexEntriesNotMonotone { + start_offset: u64, + entry_index: u64, + }, + IndexEntryBeforeSegmentStart { + start_offset: u64, + first_entry_offset: u64, + }, + /// A writer reopening over recovered bounds found the on-disk length + /// diverging from the size recovery just validated and truncated to. + StorageSizeMismatch { + start_offset: u64, + on_disk_bytes: u64, + expected_bytes: u64, + }, } -impl std::fmt::Display for PartitionChainRefusal { +impl std::fmt::Display for PartitionRecoveryRefusal { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::EmptyNonTailSegment { @@ -320,6 +378,77 @@ impl std::fmt::Display for PartitionChainRefusal { "segment {previous_start} ends at offset {previous_end} but the next \ starts at {next_start}, leaving a hole" ), + Self::IndexLogDivergence { + start_offset, + end_offset, + messages_size_bytes, + indexed_size_bytes, + } => write!( + f, + "segment {start_offset} has message/index divergence: the index ends \ + at offset {end_offset}, byte {indexed_size_bytes}, where the \ + {messages_size_bytes}-byte log holds no whole batch" + ), + Self::InteriorDamage { + start_offset, + damage_position, + survivor_position, + } => write!( + f, + "segment {start_offset} holds undecodable bytes at {damage_position} \ + with a complete verifying batch after them at {survivor_position}; \ + not a torn tail, and truncating would discard durable batches" + ), + Self::UnverifiedResidue { + start_offset, + damage_position, + residue_bytes, + candidates_examined, + budget_units, + } => write!( + f, + "segment {start_offset} holds {residue_bytes} bytes past the walked \ + prefix at {damage_position} that the damage probe could not \ + classify before exhausting its work budget ({candidates_examined} \ + candidate offsets examined of {budget_units} allowed); truncating \ + unproven bytes could destroy durable batches" + ), + Self::OffsetDiscontinuity { + start_offset, + expected_offset, + found_offset, + position, + } => write!( + f, + "segment {start_offset} holds a batch at byte {position} whose \ + base offset {found_offset} does not continue the chain at \ + {expected_offset}" + ), + Self::IndexEntriesNotMonotone { + start_offset, + entry_index, + } => write!( + f, + "segment {start_offset} index entry {entry_index} regresses in \ + offset or position; the index was not appended over this log" + ), + Self::IndexEntryBeforeSegmentStart { + start_offset, + first_entry_offset, + } => write!( + f, + "segment {start_offset} index claims offset {first_entry_offset}, \ + below the segment's own start" + ), + Self::StorageSizeMismatch { + start_offset, + on_disk_bytes, + expected_bytes, + } => write!( + f, + "segment {start_offset} file length {on_disk_bytes} diverged from \ + its recovered size {expected_bytes} at writer open" + ), } } } diff --git a/core/server_common/src/lib.rs b/core/server_common/src/lib.rs index f7b7106586..16070aeed9 100644 --- a/core/server_common/src/lib.rs +++ b/core/server_common/src/lib.rs @@ -26,6 +26,7 @@ pub mod fs_utils; pub mod iobuf; pub mod log; mod memory_pool; +mod reactor_yield; mod segment_storage; pub mod send_messages; pub mod sharding; @@ -40,6 +41,7 @@ pub use consensus_message::{ }; pub use executor::create_shard_executor; pub use memory_pool::{MEMORY_POOL, MemoryPool, MemoryPoolSettings, memory_pool}; +pub use reactor_yield::yield_to_reactor; pub use segment_storage::{ IndexReader, IndexWriter, MessagesReader, MessagesWriter, SegmentStorage, }; diff --git a/core/server_common/src/reactor_yield.rs b/core/server_common/src/reactor_yield.rs new file mode 100644 index 0000000000..66e6683c68 --- /dev/null +++ b/core/server_common/src/reactor_yield.rs @@ -0,0 +1,119 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! A yield that is guaranteed to suspend. +//! +//! Long CPU passes (recovery walks, artifact hashing) hand the core back to +//! the reactor by awaiting a short timer. The runtime's timer wheel registers +//! a timer only when its deadline is still in the future at the wheel's OWN +//! clock re-read, and its sleep future completes on the first poll without +//! ever suspending when registration is refused -- so a fixed short duration +//! is a race against the code path between the two clock reads, and the +//! window is machine- and build-dependent (a debug-build cold path loses a +//! 1 us head start essentially always). No constant wins that race; the only +//! deterministic shape is to retry with a growing duration until one +//! registration wins. + +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::time::Duration; + +/// First attempted timer duration. The common case: on a warm path one +/// microsecond outlives the registration window and the first attempt wins. +const YIELD_FIRST_ATTEMPT: Duration = Duration::from_micros(1); + +/// Ceiling for the attempt doubling. Reaching it would mean a whole-second +/// deadline was already in the past by the time the wheel re-read the clock: +/// a broken or frozen clock, not a lost race. Registration is guaranteed +/// long before; the cap only keeps the retry loop's growth finite. +const YIELD_ATTEMPT_CAP: Duration = Duration::from_secs(1); + +/// Hands the core back to the reactor: the first poll ALWAYS returns +/// `Pending` with a real timer registered, on any machine, by construction. +/// +/// A registered timer with a near-now deadline fires on the reactor's next +/// turn, so the attempted duration does not throttle the caller; it only has +/// to be long enough to register. A bare self-waking yield is no +/// alternative: this runtime does not reliably re-poll a task that wakes +/// itself from inside its own poll, and a task parked that way may never +/// resume. +pub async fn yield_to_reactor() { + RegisteredYield { + registered: None, + attempt: YIELD_FIRST_ATTEMPT, + } + .await; +} + +struct RegisteredYield { + /// The timer that won registration; later polls delegate to it. + registered: Option>>>, + attempt: Duration, +} + +impl Future for RegisteredYield { + type Output = (); + + fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<()> { + let this = self.get_mut(); + if let Some(timer) = this.registered.as_mut() { + return timer.as_mut().poll(context); + } + loop { + // One allocation per attempt; at the callers' once-per-window + // cadence that is noise against the work being yielded from. + let mut timer: Pin>> = + Box::pin(compio::time::sleep(this.attempt)); + if timer.as_mut().poll(context).is_pending() { + this.registered = Some(timer); + return Poll::Pending; + } + debug_assert!( + this.attempt < YIELD_ATTEMPT_CAP, + "a {:?} timer deadline was already in the past at registration; \ + the runtime clock is broken or frozen", + this.attempt + ); + this.attempt = (this.attempt * 2).min(YIELD_ATTEMPT_CAP); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::task::Waker; + + // Pins the suspension point itself: a yield whose future is Ready on its + // first poll never hands the core back, and nothing else would notice + // (callers still finish their pass, just without ever suspending). + #[compio::test] + async fn given_yield_future_when_polled_once_should_be_pending() { + let mut future = std::pin::pin!(yield_to_reactor()); + let mut context = Context::from_waker(Waker::noop()); + assert!( + future.as_mut().poll(&mut context).is_pending(), + "the first poll must register a real timer instead of completing inline" + ); + } + + #[compio::test] + async fn given_yield_future_when_awaited_should_complete() { + yield_to_reactor().await; + } +} diff --git a/core/server_common/src/segment_storage/index_writer.rs b/core/server_common/src/segment_storage/index_writer.rs index f0603fb8a6..8740a8bb49 100644 --- a/core/server_common/src/segment_storage/index_writer.rs +++ b/core/server_common/src/segment_storage/index_writer.rs @@ -21,7 +21,7 @@ use err_trail::ErrContext; use iggy_common::IggyError; use std::rc::Rc; use std::sync::atomic::{AtomicU64, Ordering}; -use tracing::trace; +use tracing::{error, trace}; /// A dedicated struct for writing to the index file. #[derive(Debug)] @@ -54,10 +54,6 @@ impl IndexWriter { .map_err(|_| IggyError::CannotReadFile)?; if file_exists { - let _ = file.sync_all().await.error(|e: &std::io::Error| { - format!("Failed to fsync index file after creation: {file_path}. {e}",) - }); - let actual_index_size = file .metadata() .await @@ -67,7 +63,17 @@ impl IndexWriter { .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!( + "Index file size on disk: {actual_index_size} does not match expected size: {expected_index_size}, file: {file_path}" + ); + return Err(IggyError::SegmentSizeMismatchAtOpen( + actual_index_size, + expected_index_size, + )); + } } let size = index_size_bytes.load(Ordering::Relaxed); diff --git a/core/server_common/src/segment_storage/messages_writer.rs b/core/server_common/src/segment_storage/messages_writer.rs index 1d0094becb..002402b8bc 100644 --- a/core/server_common/src/segment_storage/messages_writer.rs +++ b/core/server_common/src/segment_storage/messages_writer.rs @@ -22,7 +22,7 @@ use std::{ rc::Rc, sync::atomic::{AtomicU64, Ordering}, }; -use tracing::trace; +use tracing::{error, trace}; /// A dedicated struct for writing to the messages file. #[derive(Debug)] @@ -62,10 +62,6 @@ impl MessagesWriter { .map_err(|_| IggyError::CannotReadFile)?; if file_exists { - let _ = file.sync_all().await.error(|e: &std::io::Error| { - format!("Failed to fsync messages file after creation: {file_path}, error: {e}") - }); - let actual_messages_size = file .metadata() .await @@ -75,7 +71,20 @@ impl MessagesWriter { .map_err(|_| IggyError::CannotReadFileMetadata)? .len(); - messages_size_bytes.store(actual_messages_size, Ordering::Relaxed); + // The caller seeds the size counter from recovered, validated bounds + // and recovery truncates the file to them. A divergent on-disk length + // means appending would resurrect or shear bytes those bounds + // exclude, so refuse the open. + let expected_messages_size = messages_size_bytes.load(Ordering::Relaxed); + if actual_messages_size != expected_messages_size { + error!( + "Messages file size on disk: {actual_messages_size} does not match expected size: {expected_messages_size}, file: {file_path}" + ); + return Err(IggyError::SegmentSizeMismatchAtOpen( + actual_messages_size, + expected_messages_size, + )); + } } trace!( diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 37de132e4c..1efaa598a3 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -2176,6 +2176,30 @@ where drop(partition); continue; } + // A standing tombstone is a damage verdict or an + // unfinished teardown, and it lifts only through + // `ConfirmRemove` below, i.e. proof the disk delete + // completed. Inserting would route the namespace while + // the verdict stands: the plane drops requests for + // tombstoned namespaces without replying, so clients + // would hang to their read timeout over data declared + // lost. The reconciler skips tombstoned namespaces + // before building, so reaching this means the op was + // staged before the fence landed. Damage control like + // the guard above: the build already planted its + // initial segment on disk. + if partitions.is_tombstoned(&namespace) { + tracing::error!( + shard = self_shard_id, + ns_raw = namespace.inner(), + epoch, + "discarding InsertOwned for a tombstoned namespace: the \ + tombstone lifts only via ConfirmRemove, never by routing a \ + fresh build over it" + ); + drop(partition); + continue; + } partitions.insert(namespace, *partition); self.shards_table.insert( namespace, diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 1662ea6a1c..8e971c7083 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -373,6 +373,10 @@ impl Simulator { /// # Panics /// Panics if a replica's shard count does not fit `u32` (impossible: /// mesh construction caps it at `u16`). + // TODO(hubcio): partitions created down this path are built via + // `IggyPartition::with_in_memory_storage` and rely on the writer-less + // persist branch in `IggyPartition`; give them first-class in-memory + // segment storage so that branch can be deleted. #[allow(clippy::cast_possible_truncation)] pub fn init_partition(&mut self, namespace: IggyNamespace) { for (i, replica) in self.replicas.iter().enumerate() { diff --git a/foreign/go/errors/errors.yaml b/foreign/go/errors/errors.yaml index b52fa9b163..c1f96ea4a9 100644 --- a/foreign/go/errors/errors.yaml +++ b/foreign/go/errors/errors.yaml @@ -1057,6 +1057,14 @@ fields: - name: Value type: uint64 +- name: SegmentSizeMismatchAtOpen + code: 4102 + format: "segment file size on disk: %d does not match expected size: %d" + fields: + - name: OnDisk + type: uint64 + - name: Expected + type: uint64 - name: ConsumerGroupIdNotFound code: 5000 format: "consumer group with id: %d for topic with id: %d was not found." diff --git a/foreign/go/errors/errors_gen.go b/foreign/go/errors/errors_gen.go index f69cfee3c9..ca981aa126 100644 --- a/foreign/go/errors/errors_gen.go +++ b/foreign/go/errors/errors_gen.go @@ -2172,6 +2172,20 @@ func (e InvalidReservedField) Is(target error) bool { return ok } +type SegmentSizeMismatchAtOpen struct { + OnDisk uint64 + Expected uint64 +} + +func (e SegmentSizeMismatchAtOpen) Error() string { + return fmt.Sprintf("segment file size on disk: %d does not match expected size: %d", e.OnDisk, e.Expected) +} +func (e SegmentSizeMismatchAtOpen) Code() Code { return 4102 } +func (e SegmentSizeMismatchAtOpen) Is(target error) bool { + _, ok := target.(SegmentSizeMismatchAtOpen) + return ok +} + type ConsumerGroupIdNotFound struct { GroupId uint32 TopicId uint32 @@ -2824,6 +2838,7 @@ var ( ErrProducerClosed = ProducerClosed{} ErrInvalidOffset = InvalidOffset{} ErrInvalidReservedField = InvalidReservedField{} + ErrSegmentSizeMismatchAtOpen = SegmentSizeMismatchAtOpen{} ErrConsumerGroupIdNotFound = ConsumerGroupIdNotFound{} ErrInvalidConsumerGroupId = InvalidConsumerGroupId{} ErrConsumerGroupNameNotFound = ConsumerGroupNameNotFound{} @@ -3066,6 +3081,7 @@ const ( ProducerClosedCode Code = 4057 InvalidOffsetCode Code = 4100 InvalidReservedFieldCode Code = 4101 + SegmentSizeMismatchAtOpenCode Code = 4102 ConsumerGroupIdNotFoundCode Code = 5000 InvalidConsumerGroupIdCode Code = 5002 ConsumerGroupNameNotFoundCode Code = 5003 @@ -3501,6 +3517,8 @@ func (c Code) String() string { return "InvalidOffset" case InvalidReservedFieldCode: return "InvalidReservedField" + case SegmentSizeMismatchAtOpenCode: + return "SegmentSizeMismatchAtOpen" case ConsumerGroupIdNotFoundCode: return "ConsumerGroupIdNotFound" case InvalidConsumerGroupIdCode: @@ -3982,6 +4000,8 @@ func FromCode(code Code) IggyError { return ErrInvalidOffset case InvalidReservedFieldCode: return ErrInvalidReservedField + case SegmentSizeMismatchAtOpenCode: + return ErrSegmentSizeMismatchAtOpen case ConsumerGroupIdNotFoundCode: return ErrConsumerGroupIdNotFound case InvalidConsumerGroupIdCode: diff --git a/foreign/node/src/wire/error.code.ts b/foreign/node/src/wire/error.code.ts index 8c542c1d96..d2973d22e0 100644 --- a/foreign/node/src/wire/error.code.ts +++ b/foreign/node/src/wire/error.code.ts @@ -222,6 +222,7 @@ export const translateErrorCode = (code: number): string => { case '4057': return "Producer closed"; case '4100': return "Invalid offset: {0}"; case '4101': return "Invalid reserved field value: {0}, expected: 0"; + case '4102': return "Segment file size on disk: {0} does not match expected size: {1}"; // CONSUMER GROUP case '5000': return "Consumer group with ID: {0} for topic with ID: {1} was not found.";