numinnex commented on code in PR #3946:
URL: https://github.com/apache/iggy/pull/3946#discussion_r3830600590
##########
core/server/src/segment_recovery.rs:
##########
@@ -438,106 +739,1026 @@ 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<u64> = 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
.saturating_add(u64::from(header.message_count) - 1);
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;
}
+ refuse_if_survivor_past_damage(
+ identity,
+ &mut scanner,
+ messages_path,
+ position,
+ messages_size,
+ start_timestamp.map(|_| end_offset),
+ start_offset,
+ )?;
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<u8>,
-) -> 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()
+/// 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, truncating the whole log to zero.
+fn open_messages_file(
+ identity: PartitionIdentity<'_>,
+ messages_path: &str,
+) -> Result<fs::File, ServerError> {
+ 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)
}
-fn read_batch_header(
- messages: &fs::File,
- position: u64,
+/// 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.
+/// Unlike the WAL there is NO width cap on the damage: a segment flush chunk
+/// is unbounded, so any amount of trailing garbage can still be one torn
+/// write.
+fn refuse_if_survivor_past_damage(
+ identity: PartitionIdentity<'_>,
+ scanner: &mut FileScanner<'_>,
+ messages_path: &str,
+ damage_position: u64,
messages_size: u64,
-) -> Option<BatchHeader> {
- if position.checked_add(COMMAND_HEADER_SIZE as u64)? > messages_size {
- return None;
+ chain_end_offset: Option<u64>,
+ start_offset: u64,
+) -> Result<(), ServerError> {
+ if damage_position >= messages_size {
+ // The walk consumed the whole file: nothing to classify.
+ return Ok(());
+ }
+ let survivor = scanner
+ .probe_for_survivor(damage_position, chain_end_offset, start_offset)
+ .map_err(|source| scan_read_failure(identity, messages_path,
&source))?;
+ if let Some(survivor_position) = survivor {
+ return Err(identity.refusal(PartitionRecoveryRefusal::InteriorDamage {
+ start_offset,
+ damage_position,
+ survivor_position,
+ }));
+ }
+ Ok(())
+}
+
+/// 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<u8>,
+ window_start: u64,
+ spill: &'scan mut Vec<u8>,
+}
+
+impl<'scan> FileScanner<'scan> {
+ fn new(file: &'scan fs::File, file_len: u64, scratch: &'scan mut
ScanScratch) -> Self {
+ let ScanScratch { window, spill } = scratch;
+ window.clear();
+ Self {
+ file,
+ file_len,
+ window,
+ window_start: 0,
+ spill,
+ }
+ }
+
+ /// 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<Option<&[u8]>> {
+ 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.
+ self.spill.resize(len, 0);
+ self.file.read_exact_at(&mut self.spill[..], position)?;
+ 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;
+ }
+ // 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 or does not decode (torn header, garbage bytes).
+ fn peek_header(&mut self, position: u64) ->
io::Result<Option<BatchHeader>> {
+ let Some(bytes) = self.slice_at(position, COMMAND_HEADER_SIZE)? else {
+ return Ok(None);
+ };
+ Ok(BatchHeader::decode(bytes).ok())
+ }
+
+ /// Position of the first complete, checksum-verifying batch starting
+ /// after `damage_position`, or `None` when the residue holds none.
+ ///
+ /// 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 offset is a candidate. The header decode pre-filters candidates
+ /// cheaply -- 204 reserved bytes must be zero -- and offset sanity plus
+ /// length bounds run before a checksum is paid, so the full verify only
+ /// runs on byte positions that already look like a plausible chain
+ /// continuation.
+ fn probe_for_survivor(
Review Comment:
**Blocker: this probe is unbounded and can wedge boot.**
The scan advances one byte at a time to EOF, paying a full `slice_at` +
`decode_batch_slice` per candidate, with no cap on `header.total_size()` and no
work budget. Cost is `O(R) + O(Q·L)` (R = residue, Q = qualifying candidates, L
= claimed length). `slice_at` re-anchors its single 4 MiB window at each
candidate, so the refill interval is `(cap − L)` and read amplification is
`cap/(cap − L)` inside the window and `L/256` past it. Those meet at `cap/256`,
which is why the spill boundary is only a ~1.9x step — the knee is *inside* the
window path, so anchoring the spill alone would not help.
It is synchronous with zero yields, in front of `BootstrapBarrier`, so boot
never completes, Ctrl-C is dead, and shard 0 never binds listeners.
What worries me most is that this needs neither corrupt nor crafted data. An
ordinary zero-padded fixed-width record decodes as a valid `BatchHeader` off
its own padding:
```
[32..40] = monotone u64 id, little-endian -> read as batch_length
[48..52] = nonzero u32 tag -> read as message_count
[52..256] = 0 (the record's own padding) -> the 204 reserved bytes
```
`BatchHeader::decode` accepts it (`batch_length >= 256`, reserved region
zero), and on a partition's first segment `advances_chain` collapses to `0 >=
0` because `chain_end_offset` is `None` and `start_offset` is 0. So any
monotone u64 in the low-megabyte range at offset 32 of a zero-padded record
qualifies: a byte offset, a row count, a file size, a nanosecond delta.
Measured against this branch: 8 MiB residue = 1.8 s, 16 MiB = 8.7 s, 32 MiB =
23 s, and 90 s at a larger id magnitude — all *accepted and truncated*, i.e.
the ordinary torn-tail path, not a refusal.
Suggested fix, in order of necessity:
1. Cumulative byte budget counting **bytes read plus bytes verified**. A
verify-only budget does not fire here: the zero-padded blob bails at frame 0
after ~40 bytes hashed (`verify_and_recompute_batch_checksum` returns on the
first bad per-message checksum), so the 90 s run verified ~12 MiB total while
reading orders of magnitude more. Read amplification is the entire cost in that
regime.
2. Residue width cap at `message_bus.max_message_size` — see my note on the
"unbounded flush chunk" comment below.
3. Per-chunk yield, using `compio::time::sleep(Duration::ZERO)` rather than
a bare yield: `state_transfer.rs:2801-2812` documents that a task waking itself
from inside its own poll is not reliably re-polled here.
`recover_segment_bounds` is already `async fn`, so making the probe async keeps
every read on sync `std::fs` and does not touch the `thread_pool_limit(0)`
constraint. `FileScanner`'s borrows hold across the await since compio futures
need not be `Send`.
One constraint I would treat as load-bearing: exhausting either the cap or
the budget must **refuse** with files preserved, never fall through to "no
survivor found" and truncate. A budget is by construction reachable by cheap
input, so a fall-through makes the cheapest input the destructive one. Distinct
reason variants would let operators separate "gave up" from "proved interior
damage".
##########
core/server/src/segment_recovery.rs:
##########
@@ -438,106 +739,1026 @@ 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<u64> = 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
.saturating_add(u64::from(header.message_count) - 1);
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;
}
+ refuse_if_survivor_past_damage(
+ identity,
+ &mut scanner,
+ messages_path,
+ position,
+ messages_size,
+ start_timestamp.map(|_| end_offset),
+ start_offset,
+ )?;
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<u8>,
-) -> 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()
+/// 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, truncating the whole log to zero.
+fn open_messages_file(
+ identity: PartitionIdentity<'_>,
+ messages_path: &str,
+) -> Result<fs::File, ServerError> {
+ 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)
}
-fn read_batch_header(
- messages: &fs::File,
- position: u64,
+/// 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.
+/// Unlike the WAL there is NO width cap on the damage: a segment flush chunk
Review Comment:
This premise is not correct, and it is what justifies omitting the bound
that makes `probe_for_survivor` unbounded.
What needs bounding is the torn **record**, not the torn **chunk**. A record
is capped by `message_bus.max_message_size` (64 MiB; see
`core/configs/src/server_config/partition.rs:69-75`, whose own doc says it
"actually bounds an appendable batch") — the same order as
`prepare_journal.rs:37`'s `MAX_ENTRY_SIZE`. So the WAL's width cap and its
refusal at `prepare_journal.rs:230-239` port over directly, contrary to what
this comment says.
There is also a tighter argument available: a residue holding no complete
batch is bounded by roughly one batch by construction, since any whole batch
that follows the torn record verifies and the probe returns immediately. That
is not a hard invariant (with near-64 MiB batches and non-ascending writeback,
several records can each be holed), but it does not need to be — since
exhaustion refuses, an under-tight cap only fences a partition a wider cap
might have truncated, which errs toward preserving bytes.
Worth deleting alongside the probe fix, since leaving it tells the next
reader the bound is impossible.
##########
core/server/src/segment_recovery.rs:
##########
@@ -147,46 +235,91 @@ 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 means disk diverged
+ // from the size this pass just truncated to: structural, and the
+ // heal path for data directories an earlier size-counter bug left
+ // with resurrected tails. 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);
Review Comment:
**Blocker: u64 underflow.** This underflows whenever a walked header carries
a `base_offset` below the segment's start.
Reproduced: segment named 100, log = `encoded_batch(100, 1) ||
encoded_batch(5, 1)`, index = one entry `(100, 0)` gives
```
panicked at core/server/src/segment_recovery.rs:263:44: attempt to subtract
with overflow
```
Release has no `overflow-checks` (root `Cargo.toml:372`), so it wraps to
roughly `u64::MAX - 93` into `PartitionStats::messages_count`. That propagates
partition → topic → stream, and both `/stats` and `/metrics` fold streams with
`saturating_add`, so one wrapped partition saturates the node total. The web UI
types `messagesCount` as `number` (`web/src/lib/domain/Partition.ts:26`,
`Stream.ts:23`), so it silently rounds past `Number.MAX_SAFE_INTEGER`.
The root cause looks like an asymmetry this PR introduces rather than the
arithmetic itself: the index-less arm refuses `OffsetDiscontinuity` and index
entries get `IndexEntryBeforeSegmentStart`, but walked headers in the indexed
arm get neither a continuity check nor a checksum (per the TODO above the
indexed walk).
`saturating_sub` alone would not be enough — the same bogus `end_offset`
also drives `current_offset` below the active segment's own start, so later
appends mint offsets that segment routing cannot resolve. I think the walked
header needs to be refused, which also makes the arithmetic safe by
construction.
##########
core/server/src/segment_recovery.rs:
##########
@@ -320,44 +445,196 @@ fn
sweep_scratch_files_and_collect_offsets(partition_path: &str) -> Result<Vec<u
Ok(start_offsets)
}
-fn file_len(path: &str) -> 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 and
+/// truncate it to nothing (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<u64, ServerError> {
+ 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())
+ }
+ }
+}
+
+/// 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(())
+}
+
+/// Persists the index rebuilt by the index-less walk, replacing whatever
+/// partial or stale bytes the crash left. Without this a SEALED segment --
+/// which never flushes again -- would keep an empty index forever and pay a
+/// full log scan on every poll.
+///
+/// Written straight to the final path: a crash mid-write leaves a shorter
+/// index whose whole entries are a valid prefix of this same rebuild, and the
+/// next boot walks and rewrites it again -- recovery is itself the repair
+/// path for a torn index, so no rename dance is needed.
+fn write_rebuilt_index(path: &str, entries: &[u8]) -> Result<(), ServerError> {
Review Comment:
**Blocker: this repair path can fabricate the corruption its own validator
refuses.**
The rebuilt index is written in place: `.create(true).truncate(true)` sets
the length to 0, then `write_all_at` extends. The doc just above says a crash
mid-write leaves "a shorter index whose whole entries are a valid prefix of
this same rebuild", but a single buffered pwrite gives no prefix guarantee
before the fsync — writeback can persist a later page while an earlier one
still reads as zeros. And zeros decode as valid 24-byte entries, which
`truncate_to`'s own doc states a few lines up ("zero bytes decode as
valid-looking index entries -- three bare little-endian u64s with no magic to
reject them").
Confirmed the outcome directly: an `.index` of 48 zero bytes over a valid
log yields `PartitionRecoveryRefused { IndexEntriesNotMonotone { entry_index: 1
} }`. So the repair path can produce exactly the shape `validate_index_entries`
fences on, and at `replica_count = 1` that partition comes back empty.
The staging + rename + dir fsync pattern already exists in tree at
`state_transfer.rs:1247`. Writing the rebuilt index to `{stem}.index.staging`
during pass A and renaming in pass C would also bound the pass-A peak, which
currently holds every segment's rebuilt index in memory simultaneously (~384
KiB per segment at the 64 KiB stride).
`sweep_scratch_files_and_collect_offsets` already unlinks `*.staging`, so the
fix composes with what is here.
Minor, same function: `.create(true)` is dead — this arm is only reachable
after `IggyIndexReader::new` succeeded, so the file exists.
##########
core/server/src/segment_recovery.rs:
##########
@@ -438,106 +739,1026 @@ 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<u64> = 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
.saturating_add(u64::from(header.message_count) - 1);
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;
}
+ refuse_if_survivor_past_damage(
+ identity,
+ &mut scanner,
+ messages_path,
+ position,
+ messages_size,
+ start_timestamp.map(|_| end_offset),
+ start_offset,
+ )?;
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<u8>,
-) -> 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()
+/// 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(
Review Comment:
**Blocker (adjacent):** this new validator is what converts a pre-existing
writer bug into a partition-destroying boot refusal, so I think the writer fix
belongs in this PR.
In `core/partitions/src/iggy_index_writer.rs`: `write_all_at` at `:113-117`,
`fetch_add` at `:119-120`, then `if self.fsync { self.fsync().await? }` at
`:122-124`. On an fsync error the `?` returns with the counter already
advanced. `iggy_partition.rs:3663-3673` rewinds only the messages cursor, and
the retry's `flush_index` triple is unchanged because `segment.size` only
advances after a successful persist (`:3688-3690`). So the retry appends a
byte-identical triple at index byte I+24 while I already holds it, and
`validate_index_entries` refuses it here as `IndexEntriesNotMonotone`. Master
had no monotonicity validation and booted through this.
Gated on `enforce_fsync = true`, which is not the default — but it is what
durability-minded operators choose, and this PR's own integration test sets it
(`core/integration/tests/cluster/crash_recovery_corruption.rs:71`). Worth
noting Linux reports an fsync error once per fd and then clears it, so on a
degrading disk the first fsync fails and the retry's succeeds, which is
precisely the sequence that plants the duplicate.
Moving `fetch_add` after the fsync makes the retry overwrite at I instead of
appending at I+24, so boot's file-length-derived `entry_count` sees no
duplicate. One line.
Separately on this function (not blocking): it reads every index file whole
at boot, ungated, even for a clean sealed segment, while
`poll_plan.rs:59`/`:699` refuses exactly that read above
`SEALED_INDEX_RESIDENT_MAX_BYTES` (512 KiB) and `IggyIndexReader::load_all`'s
own rustdoc tells callers that cannot afford an unbounded read to gate on
`entry_count`. Benign at the default flush cadence (~24 KiB/segment), but
`messages_required_to_save = 1` with small messages puts it in the tens of MiB
per segment. Gating the interior scan on `entry_count` while keeping first+last
mandatory would resolve it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
