hubcio commented on code in PR #3946:
URL: https://github.com/apache/iggy/pull/3946#discussion_r3832847418
##########
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:
fixed in cfd2265e1, in your order. step 1: reconcile_additions checks the
tombstone for unmaterialised namespaces too, and InsertOwned apply refuses to
route a tombstoned one - refuse over clear, since the tombstone lifts only via
ConfirmRemove (proof the delete completed) and a second untombstone path would
re-open the serve-empty hole. step 2: a tombstone verdict no longer quarantines
- the chain stays put, so the verdict re-derives and re-logs every boot.
StorageSizeMismatch's non-durable cause is called out in the arm, and
config.toml now says the rc=1 files stay at their original paths. one
consequence: delete + recreate of a boot-tombstoned partition stays dark until
the refused files are cleared by hand, since recreate reuses the namespace and
nothing tears down what was never routed.
##########
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:
taken in 0929f697b, one deviation. gate deleted; budget charges one unit per
candidate examined, scoped to the whole partition load. for the limit i didn't
use a compile-time constant: it grows 2x per residue byte the load actually
presents - knob-immune by construction, and there's no bound left to
mis-derive. exhaustion still refuses; residue width stays as a diagnostic
field, and your wide-zeros shape is now a test asserting truncate. one residual
the flat charge shares with your proposal: a residue with forged-valid
per-message checksums can make each candidate's verify run long - that needs
arbitrary local writes, at which point planting valid batches is easier than
slowing boot, so i left it outside the threat model. the 256 MiB upper
validator on max_message_size landed too (frozen const in iggy_common).
##########
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:
fixed in 0929f697b - the indexed arm now refuses only found < expected; a
forward gap absorbs with a warn naming the frontier-restore cause. your repro
is a test (recovers, end offset 5). the index-less arm still refuses both
directions, since a break there is a truncation point. one note: absorbing the
gap leaves messages_count overcounting by the hole width - master did the same,
left alone.
##########
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:
fixed in 0929f697b - peek_header rejects any total_size above a frozen
MAX_RECOVERABLE_BATCH_BYTES (the new 256 MiB max_message_size ceiling plus the
header envelope), and the same cap sits in the probe's fits test, which bounds
the spill there now that the width gate is gone. tests cover the 0xFFFF_FFFF
claim both as a tail (truncates at break) and with a valid batch after it
(refuses).
##########
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:
you were right that sleep(0) never registers, but from_micros(1) turned out
machine-dependent too - on my box in a debug build the path between sleep's
deadline mint and insert's clock re-read exceeds 1us essentially always, so the
first-poll-Pending test failed 4/4 here. replaced in 0929f697b with a shared
server_common::yield_to_reactor: it polls a sleep once and, if it completed
inline, doubles the duration and retries inside the same poll - first poll
suspends by construction, no race to win. state_transfer's helper is folded
into the same fn and the Pending test now pins it deterministically.
--
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]
