numinnex commented on code in PR #3946:
URL: https://github.com/apache/iggy/pull/3946#discussion_r3832417622
##########
core/server/src/segment_recovery.rs:
##########
@@ -438,106 +1012,1407 @@ 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,
probe_limits, 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;
+ 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<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()
-}
-
-fn read_batch_header(
- messages: &fs::File,
- position: u64,
+/// 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, 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)
+}
+
+/// 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.
+///
+/// What needs bounding is the torn RECORD, not the flush chunk: a flush
+/// chunk is unbounded, but every record inside it is capped by
+/// `message_bus.max_message_size`, and a residue holding no complete batch
+/// is about one record wide by construction -- any following whole batch
+/// verifies and ends the probe. Several holed near-max records can stack
+/// wider than the cap, which is exactly why cap and budget exhaustion REFUSE
+/// and keep the bytes rather than truncating: past the limits 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,
-) -> 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 residue_bytes = messages_size - damage_position;
+ let limits = scanner.limits;
+ if residue_bytes > limits.max_residue_bytes {
+ return Err(
+ identity.refusal(PartitionRecoveryRefusal::UnverifiedResidue {
+ start_offset,
+ damage_position,
+ residue_bytes,
+ scan_limit_bytes: limits.max_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,
+ scan_limit_bytes: limits.scan_budget_bytes,
+ },
+ )),
+ 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,
+ limits: ProbeLimits,
+ window: &'scan mut Vec<u8>,
+ window_start: u64,
+ spill: &'scan mut Vec<u8>,
+ refilled: bool,
+}
+
+impl<'scan> FileScanner<'scan> {
+ fn new(
+ file: &'scan fs::File,
+ file_len: u64,
+ limits: ProbeLimits,
+ scratch: &'scan mut ScanScratch,
+ ) -> Self {
+ let ScanScratch { window, spill } = scratch;
+ window.clear();
+ Self {
+ file,
+ file_len,
+ limits,
+ window,
+ window_start: 0,
+ spill,
+ 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<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)?;
+ 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 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())
+ }
+
+ /// 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.
+ ///
+ /// Every byte read from disk and every byte handed to verification is
+ /// charged against the scan budget. Charging the handed slice whole --
+ /// even when the verify bails early or the bytes were already windowed --
+ /// is deliberate: verification cost is what a crafted residue can inflate
+ /// without adding reads, and a pessimistic charge keeps the bound
+ /// deterministic. Exhaustion returns [`ProbeOutcome::BudgetExhausted`],
+ /// never `NoSurvivor`.
+ async fn probe_for_survivor(
+ &mut self,
+ damage_position: u64,
+ chain_end_offset: Option<u64>,
+ start_offset: u64,
+ ) -> io::Result<ProbeOutcome> {
+ let header_len = COMMAND_HEADER_SIZE as u64;
+ let mut spent_bytes = 0u64;
+ // 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 {
+ spent_bytes =
spent_bytes.saturating_add(self.fill_window_at(candidate)?);
+ let window_end = self.window_start + self.window.len() as u64;
+ while candidate.saturating_add(header_len) <= window_end {
+ // 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();
+ let fits = candidate.saturating_add(total_size as u64) <=
self.file_len;
+ if advances_chain && fits && header.message_count > 0 {
+ let (batch, read_bytes) = self.verify_slice(candidate,
total_size)?;
+ if decode_batch_slice(batch).is_ok() {
+ return Ok(ProbeOutcome::Survivor {
+ position: candidate,
+ });
+ }
+ spent_bytes = spent_bytes
+ .saturating_add(read_bytes)
+ .saturating_add(total_size as u64);
+ }
+ }
+ candidate += 1;
+ if spent_bytes > self.limits.scan_budget_bytes {
+ return Ok(ProbeOutcome::BudgetExhausted);
+ }
+ }
+ if self.take_refilled() {
+ yield_to_reactor().await;
+ }
+ }
+ Ok(ProbeOutcome::NoSurvivor)
+ }
+
+ /// Anchors the window at `position` unless the header there already sits
+ /// inside it; returns the bytes read (0 on a hit). The probe's outer
+ /// loop refills through this, so its windows advance strictly forward.
+ fn fill_window_at(&mut self, position: u64) -> io::Result<u64> {
+ 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(0);
+ }
+ 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(fill as u64)
+ }
+
+ /// 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. Returns the slice and
+ /// the disk bytes it cost. The caller bounds `len` against the file
+ /// before calling.
+ fn verify_slice(&mut self, position: u64, len: usize) ->
io::Result<(&[u8], u64)> {
+ 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], 0));
+ }
+ self.spill.resize(len, 0);
+ self.file.read_exact_at(&mut self.spill[..], position)?;
+ self.refilled = true;
+ Ok((&self.spill[..], len as u64))
+ }
+}
+
+/// Hands the shard core back to the reactor between scan windows. A
+/// zero-duration timer, NOT a bare self-waking yield: this runtime does not
+/// reliably re-poll a task that wakes itself from inside its own poll, and a
+/// boot task parked that way would never resume.
+async fn yield_to_reactor() {
Review Comment:
**Blocker: this never yields, so the mitigation added for the probe-bound
review comment is not actually present.**
`compio::time::sleep(Duration::ZERO)` resolves to
`sleep_until(Instant::now())`. `TimerRuntime::insert` returns `None` when
`deadline <= Instant::now()` (compio-runtime 0.12.3 — the version `Cargo.lock`
pins — `src/time.rs:273-276`), and `create_timer` (`:67-72`) awaits a
`TimerFuture` only `if let Some(key) = key`, with no else branch. Since
`insert` re-reads the clock, a deadline of "now captured one call earlier"
always satisfies `<=`, so the future completes on its first poll and no
suspension point is ever constructed. `Duration::ZERO` is precisely the value
that can never register.
Measured three independent ways on this branch:
- 3,000,000 calls, sibling task gained **0** ticks (control `sleep(5ms)`
gained 95, `sleep(1us)` gained 983/2000)
- first poll of `sleep(0ns)` = `Ready`, `sleep(1ns)` = `Ready`, `sleep(1us)`
= `Pending`
- polling this function directly with a noop waker: `yield_to_reactor()`
first poll `Ready`, `sleep(1us)` first poll `Pending`
So all three call sites (`:973`, `:1087`, `:1485`) are no-ops, the walk and
probe still pin the shard core through `BootstrapBarrier`, and the doc at
`:1531-1534` asserts the opposite of what happens.
Fix: `Duration::from_micros(1)` — measured 13.3 µs/call and it does yield.
Do **not** use `from_nanos(1)` (measured 0 yields / 2000) or `from_nanos(100)`
(46/2000, racy). At the `take_refilled()` cadence that is ≤512 yields/GiB ≈
3.4-6.8 ms/GiB against a measured 90-102 ms/GiB walk, i.e. 3-7%.
Worth fixing `core/partitions/src/state_transfer.rs:2813` in the same
commit: it is the same body with the same rationale and two live callers
(`:748`, `:2995`), and it has been dead too. A test asserting `Pending` on
first poll would pin this — nothing currently would notice it regressing.
##########
core/server/src/segment_recovery.rs:
##########
@@ -438,106 +1012,1407 @@ 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,
probe_limits, 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;
+ 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<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()
-}
-
-fn read_batch_header(
- messages: &fs::File,
- position: u64,
+/// 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, 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)
+}
+
+/// 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.
+///
+/// What needs bounding is the torn RECORD, not the flush chunk: a flush
+/// chunk is unbounded, but every record inside it is capped by
+/// `message_bus.max_message_size`, and a residue holding no complete batch
+/// is about one record wide by construction -- any following whole batch
+/// verifies and ends the probe. Several holed near-max records can stack
+/// wider than the cap, which is exactly why cap and budget exhaustion REFUSE
+/// and keep the bytes rather than truncating: past the limits 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,
-) -> 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 residue_bytes = messages_size - damage_position;
+ let limits = scanner.limits;
+ if residue_bytes > limits.max_residue_bytes {
Review Comment:
**Blocker: this cap refuses the canonical torn tail — the exact case this PR
exists to truncate — and at `replica_count = 1` that is now silent data loss.**
`max_residue_bytes` is `config.message_bus.max_message_size` (`:377`), 64
MiB by default. But what a torn append can leave holed is bounded by the *flush
chunk*, not by one record: `iggy_partition.rs:2978` mints one index entry per
flush and the chunk loop breaks only at `file_position >= max_segment_size`
(`:2997`, 1 GiB default). With `enforce_fsync = false` the whole chunk is a
single unordered writeback, so the ordinary ext4 delayed-allocation artifact —
`i_size` extended, tail pages never written back, reading as zeros — leaves a
residue of hundreds of MiB with no survivor in it. That is a torn tail, and it
now refuses.
Reproduced on this branch: valid prefix + 4096 zero bytes with the cap at 1
KiB yields `"segment 0 holds 4096 bytes past the walked prefix at 400 that the
damage probe could not classify within its 1024-byte limit"`. `a8240cafd`
truncated and served the prefix. `UnverifiedResidue` is not in the
`rebuild_for_rejoin` carve-out (`bootstrap.rs:1934`), so at `replica_count = 1`
this tombstones.
The doc at `:1256-1262` cites the WAL as precedent, but the WAL's own
justification does not transfer: `prepare_journal.rs:206` reasons "one entry
per `append` + fsync means a torn tail is at most one entry wide". Segments
fsync per flush chunk at best, never per record. And
`prepare_journal.rs:225-239` probes *first* and keeps the size check "as a
second refusal, not as the classifier" — this code inverted that order.
Two further problems with the same constant:
1. **It is read from a live knob** whose validator is floor-only
(`configs/src/server_config/message_bus.rs:144`, `validators.rs:229-235`).
Lowering `max_message_size` between boots permanently refuses partitions
written under the old value, and at `replica_count > 1` every replica applies
the same lowered value, so all of them refuse, all take `rebuild_for_rejoin`,
and the group returns empty and healthy-looking. `prepare_journal.rs:37` is a
compile-time `const` for exactly this reason.
2. **It is currently the only thing bounding the probe's cost**, because the
per-candidate `BatchHeader::decode` is charged nothing. Measured with every
candidate decoding but `fits` failing: 64 MiB → 63.7 ms, 256 MiB → 244 ms, 1
GiB → 977 ms, all verdict `Ok` (truncate), budget never fired. So raising the
cap without also charging candidates re-opens an unbounded scan.
Suggested shape, and it wants to land as **one** atomic change — splitting
it is how the live-knob problem appeared in the first place:
- **Delete the width gate.** Once candidates are charged the budget bounds
total work regardless of file size, which is immune to the knob by
construction. Keep the residue width as a diagnostic field in
`UnverifiedResidue` rather than a gate. (Note the value of this constant has
now been mis-derived three times — `max_message_size` here, and a frozen
`SEGMENT_MAX_SIZE_BYTES + SEGMENT_SIZE_OVERSHOOT_BYTES` is also wrong, since
`partition.rs:65-74` documents the overshoot as *tracking the shipped default*,
not bounding the knob, so a legal `max_message_size = 256 MiB` produces legal
segments the frozen cap would refuse.)
- **Charge per candidate examined**, not per candidate verified. One unit
per candidate: decode cost is flat, not proportional to header size — zeros
bail on the `batch_length < 256` compare (`batch.rs:107-112`), garbage on the
first nonzero reserved byte, and measured spread across zeros / 0xAB /
fully-decoding content is 0.95-1.19 s/GiB, within 25%. Charging
`COMMAND_HEADER_SIZE` against a byte budget instead exhausts after ~8.5 MiB of
residue and refuses honest tails narrower than today's cap.
- **Scope the budget to the whole `load_persisted_segments` call.**
`spent_bytes` is currently a local reset per `probe_for_survivor` (`:1450`)
while `ProbeLimits` is per-load, and pass A probes every segment before pass B
can refuse, so the cost is `O(S × limit)`.
- **Derive the limit from a compile-time constant**, which needs an upper
validator on `message_bus.max_message_size` — there is none today, so no
compile-time bound on the widest legal sealed segment is currently derivable.
- Size the multiplier against the honest worst case with the composition you
pick: dropping the window-fill charge leaves candidates ≈ 1× residue (so 2× is
ample); keeping it makes spend ≈ 2.00006× residue, which overruns a 2× budget
and needs 4×.
For scale, on the current code with the cap lifted the crafted rows cost
122-514 ms and the canonical honest tail 1189 ms with verdict `Ok` — so the
correct behaviour is affordable; it is the gate, not the cost, that is refusing.
##########
core/server/src/segment_recovery.rs:
##########
@@ -378,55 +903,104 @@ async fn recover_segment_bounds(
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,
probe_limits, 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): this indexed walk trusts the header decode alone,
+ // 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. A
+ // decodable header that breaks the chain is not a later
+ // flush of this segment: absorbing it would adopt offsets
+ // the chain never proved -- a lower one regresses the
+ // partition's offset counter at bootstrap and re-mints
+ // already-served offsets on the next append, and one below
+ // the segment start underflows the recovered message count.
+ // Refuse, mirroring the index-less walk.
+ if header.base_offset != expected_offset {
Review Comment:
**Blocker: this refuses byte-clean data that the server's own boot path
creates, and it is strictly worse here than both master and `a8240cafd`.**
Only `found_offset < expected_offset` can underflow the message count at
`:314`. This condition also refuses `found_offset > expected_offset` — a
forward gap — and a forward gap in a byte-clean tail segment is minted by
ordinary recovery, with no damage and no rare timing:
1. `iggy_partition.rs:654,680-703` stamps `offset_frontier = offset + 1`
into the fsynced superblock.
2. Segment writes are not fsynced at the shipped `enforce_fsync = false`,
and `iggy_partition.rs:3648-3663` writes messages *before* indexes, so a crash
can leave the frontier ahead of the log.
3. On boot `bootstrap.rs:2675-2691` seeds `offset` from the max segment end,
then `restore_offset_frontier` (`iggy_partition.rs:761-780`) raises it to
`frontier - 1`.
4. `ensure_initial_segment` returns early because segments exist
(`partition_helpers.rs:304-313`), so the next append stamps `base_offset =
frontier` into the **existing tail segment** — an offset hole with no byte hole.
5. That segment's index is intact, so the next boot walks the *indexed* arm
and hits this check.
Reproduced: log `encoded_batch(0,2) ++ encoded_batch(5,1)`, index
`index_entry(0,0)` gives `"segment 0 holds a batch at byte 400 whose base
offset 5 does not continue the chain at 2"`. Both master and `a8240cafd`
absorbed this and served the data — `git show
master:core/server/src/segment_recovery.rs` shows the indexed walk had no
continuity check, and `end_offset` came out correct.
The consequence is worse than unavailability. `OffsetDiscontinuity` is not
in the `rebuild_for_rejoin` carve-out (`bootstrap.rs:1934`), so at
`replica_count = 1` the partition is tombstoned, and `untombstone` has exactly
one caller — `ConfirmRemove` (`shard/src/lib.rs:2207`) — so there is no
operator remedy short of deleting the partition. See my note on the tombstone
branch for why that then degrades to serving empty.
Worth noting this file already documents the shape:
`server_error.rs:315-320`, added in this same change, says "a crash window that
leaves the durable offset frontier past the recovered end offset stamps the
same shape into byte-clean files."
Fix: refuse only `found_offset < expected_offset`; absorb a forward gap and
`warn!`. That keeps the `:314` arithmetic safe by construction — which is the
property the guard was added for — without refusing a chain whose bytes are all
present and all decodable. The upstream defect (boot minting a hole inside an
existing segment) is worth its own fix, ideally sealing and rotating when
`restore_offset_frontier` raises the counter above the recovered end, but that
is separate from this refusal.
##########
core/server/src/bootstrap.rs:
##########
@@ -1963,8 +1981,22 @@ async fn build_shard_for_thread(
continue;
}
}
- // The refused load already folded its segment counts in.
+ // 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 {
+ error!(
+ stream_id,
+ topic_id,
+ partition_id = partition_metadata.id,
+ partition_dir,
+ "no peer replica holds this partition's data;
tombstoning it \
+ instead of serving it empty"
+ );
+ partitions.tombstone(namespace);
Review Comment:
**Blocker: this tombstone does not hold, so the `replica_count = 1`
protection added for the previous review round is not in effect. It is defeated
twice, and the end state is worse than before the change.**
**Within this same boot.** The `continue` here runs *before*
`partitions.insert` (`:2042`), so `partitions.contains(&ns)` is false — and
`reconcile_additions` consults `is_tombstoned` only *inside* `if
partitions.contains(&ns)` (`partition_reconciler.rs:540`, `:559`). So the
namespace falls through the owner check and `has_staged_insert_owned` to
`build_partition_fresh` (`:646`), which calls `ensure_initial_segment`
(`partition_helpers.rs:668-669`) and plants a `.log`/`.index` pair on disk at
`offset_frontier()`. `ReconcileOp::InsertOwned` apply
(`shard/src/lib.rs:2146-2183`) then inserts into `partitions` *and*
`shards_table` without calling `untombstone`, so the namespace ends up routed
while still tombstoned: `on_request` returns with no reply
(`iggy_partitions.rs:533-541`), so clients hang to their read timeout rather
than getting an error or an empty result. Every later reconciler pass then hits
`contains && is_tombstoned && !has_pending_delete_failure` → `deferred; conti
nue` forever (`:552-562`), because no delete was ever attempted.
**On the next boot.** `quarantine_segment_files` moves only `.log` /
`.index` / `.staging` (`state_transfer.rs:1235-1247`); the superblock slots are
`superblock.a` / `superblock.b` (`journal/src/superblock.rs:95-96`) and
survive, by design. So the next boot's
`sweep_scratch_files_and_collect_offsets` finds no `.log` stem,
`load_persisted_segments` returns empty, `restore_offset_frontier` re-seeds
from the surviving superblock, `ensure_initial_segment` plants
segment-at-frontier, and the partition is inserted and **served empty — with no
refusal logged at all.** Iteration 1 at least re-logged the refusal on every
boot; this is quieter.
Reproduced: real segment plus a `superblock.a`/`.b` pair, called
`quarantine_segment_files`, asserted the superblock survives and both segment
files moved out, then `load_persisted_segments` → `Ok` with 0 segments.
The asymmetry against the precedent this arm was modelled on: the superblock
tombstone at `:2037` deliberately does *not* quarantine (`:2013-2020`, "The
segment files stay exactly where they are ... nothing to quarantine"), so its
trigger survives and it re-derives the same verdict every boot. This arm
quarantines first and so destroys its own evidence. A tombstone is only durable
if its cause is.
Suggested fix, and **the order matters because reversing it destroys data**:
1. **First**, gate `reconcile_additions` on `is_tombstoned` regardless of
`contains`, and either clear the tombstone in `InsertOwned` apply or refuse to
build. Safe to land on its own, and required under any tombstone shape.
2. **Then** stop quarantining when the verdict is a tombstone. Leaving the
chain in place makes the verdict re-derivable and re-logged every boot, with
the bytes at their real paths, and mints no `.fenced.N`.
Not the other order: with the files left in place but the reconciler gate
absent, `ensure_initial_segment` opens `SegmentStorage::new(.., 0, 0, false)`
and `file_exists = false` **truncates both files** (its own comment at
`partition_helpers.rs:340-344` says so). When the frontier is absent or 0 —
which `build_partition_fresh`'s comment at `:652-664` calls normal — that
truncates `00000000000000000000.log`, the oldest segment of the refused chain,
with no fenced copy anywhere. Quarantining is currently what makes the
reconciler fall-through survivable.
Two things to carry with it: `StorageSizeMismatch` is the one refusal whose
cause genuinely is not durable (it arrives from pass C after pass C already
truncated, so it self-heals rather than re-tombstoning — correct for a "the
filesystem lied" assertion, but worth stating), and the `.fenced.N` description
in `config.toml` needs updating, since the tombstone branch would then leave
files in the partition directory rather than in `.fenced.N` as `:478` currently
claims.
##########
core/server/src/segment_recovery.rs:
##########
@@ -438,106 +1012,1407 @@ 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,
probe_limits, 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;
+ 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<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()
-}
-
-fn read_batch_header(
- messages: &fs::File,
- position: u64,
+/// 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, 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)
+}
+
+/// 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.
+///
+/// What needs bounding is the torn RECORD, not the flush chunk: a flush
+/// chunk is unbounded, but every record inside it is capped by
+/// `message_bus.max_message_size`, and a residue holding no complete batch
+/// is about one record wide by construction -- any following whole batch
+/// verifies and ends the probe. Several holed near-max records can stack
+/// wider than the cap, which is exactly why cap and budget exhaustion REFUSE
+/// and keep the bytes rather than truncating: past the limits 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,
-) -> 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 residue_bytes = messages_size - damage_position;
+ let limits = scanner.limits;
+ if residue_bytes > limits.max_residue_bytes {
+ return Err(
+ identity.refusal(PartitionRecoveryRefusal::UnverifiedResidue {
+ start_offset,
+ damage_position,
+ residue_bytes,
+ scan_limit_bytes: limits.max_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,
+ scan_limit_bytes: limits.scan_budget_bytes,
+ },
+ )),
+ 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,
+ limits: ProbeLimits,
+ window: &'scan mut Vec<u8>,
+ window_start: u64,
+ spill: &'scan mut Vec<u8>,
+ refilled: bool,
+}
+
+impl<'scan> FileScanner<'scan> {
+ fn new(
+ file: &'scan fs::File,
+ file_len: u64,
+ limits: ProbeLimits,
+ scratch: &'scan mut ScanScratch,
+ ) -> Self {
+ let ScanScratch { window, spill } = scratch;
+ window.clear();
+ Self {
+ file,
+ file_len,
+ limits,
+ window,
+ window_start: 0,
+ spill,
+ 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<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);
Review Comment:
**Blocker: one bit-flipped `batch_length` allocates and reads up to a whole
segment here, on the boot path, with no yield.**
This is the index-less walk's per-batch verify path. `len` is
`header.total_size()`, and the only guard before it is `extent <=
messages_size` — the *file* length, not any cap. So a `batch_length` field
corrupted upward (the classic `0xFFFF_FFFF`-style flip) drives
`spill.resize(len, 0)` to `max_segment_size`, 1 GiB by default.
Measured: cold `Vec::resize(1 GiB, 0)` is **42.8 ms** of memset (23-25 GB/s
including page faults), then a 1 GiB `pread` overwrites it, then
`decode_batch_slice` fails and the walk breaks at that position anyway. And
`spill` never shrinks, so **1 GiB of RSS stays pinned for the rest of the
partition load**. Shards load concurrently, one per core, so the ceiling is
`nproc × max_segment_size` — 32 GiB on a 32-core box. With `yield_to_reactor`
currently being a no-op (separate comment), none of it is preemptible.
This is verbatim the failure mode the WAL already guards against —
`prepare_journal.rs:37`: *"prevents a bit-flipped size field (e.g.
`0xFFFF_FFFF`) from causing a multi-GiB allocation during the WAL scan."* The
probe's `verify_slice` at `:1524` is incidentally bounded, because `fits` is
checked against `file_len` with `candidate > damage_position` so `total_size <
residue`; the walk has no such relation. So the bound got ported to one of the
two arms.
Fix: reject a header whose `total_size()` exceeds the largest record a
segment can legally hold, in `peek_header` before slicing. One `u64`
comparison, ~64k comparisons per GiB of walk at the measured batch density,
under 1 µs/GiB against a 90-102 ms/GiB walk. Verdict-identical: an oversized
header cannot be a real batch, and today's code reaches the same reject after
paying the giant read. Drops the ceiling from `nproc × 1 GiB` to `nproc × 64
MiB`.
Apply the same guard in the probe's `fits` test as well, not just here — if
the residue width gate goes away (see my note on `max_residue_bytes`), the
probe's spill loses the bound it currently gets from that cap, and this becomes
the only thing holding 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]
