numinnex commented on code in PR #3946:
URL: https://github.com/apache/iggy/pull/3946#discussion_r3835217768
##########
core/server/src/segment_recovery.rs:
##########
@@ -378,55 +925,122 @@ 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,
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 --
+ // except in one direction the server mints itself. Refuse
+ // only a REGRESSION: a lower offset re-adopted here regresses
+ // the partition's offset counter at bootstrap (re-minting
+ // already-served offsets on the next append) and one below
+ // the segment start underflows the recovered message count.
+ // A FORWARD gap in a byte-clean, fully decodable tail is
+ // ordinary boot output, not damage: a crash can persist the
+ // offset frontier ahead of the unsynced log, and the next
+ // boot then stamps `base_offset = frontier` into this same
+ // tail segment. Absorb it and keep serving; every offset
+ // adopted is one the primary durably promised.
+ if header.base_offset < expected_offset {
+ return Err(
+
identity.refusal(PartitionRecoveryRefusal::OffsetDiscontinuity {
+ start_offset,
+ expected_offset,
+ found_offset: header.base_offset,
+ position,
+ }),
+ );
+ }
+ if header.base_offset > expected_offset {
Review Comment:
> **Corrected after reviewing `7a9c4b3f0` and `c727a1cc9`.** My original
recommendation here was to revert this branch to `header.base_offset !=
expected_offset`. **That was wrong** — a plain revert is an availability
regression, and the checksum gate you added is what makes a revert safe.
Corrected recommendation below; the rest of the original finding stands,
re-scoped.
**The checksum gate narrows this finding; it does not close it.**
Confirmed at `c727a1cc9` that a bit flip in `base_offset` now fails
`decode_batch_slice` and truncates cleanly — `end_offset = 1`, log cut to the
400-byte valid prefix. `write_batch_header_fields` (`batch.rs:452-458`) does
hash `base_offset`, so that route is genuinely shut.
But the flip was the cheapest *witness* of the finding, not the finding. Ran
against `c727a1cc9`: appending a **clean** `encoded_batch(2^63+2, 1)` after
`encoded_batch(0, 2)` under `index_entry(0, 0)` still recovers `end_offset =
9223372036854775810` — the identical value, with no checksum corrupted
anywhere. `calculate_batch_checksum` binds `base_offset` to the batch's own
bytes, never to its position in this log, so it can prove self-consistency and
cannot prove placement. Any whole intact record that lands in the tail from
somewhere else — block recycle, misdirected write, a replay from elsewhere in
the partition, an operator copy — verifies and is adopted. And the permanence
chain is untouched: that value seeds `current_offset` (`bootstrap.rs:2672-2679`
→ `:2708-2710`), and `write_superblock`'s advance-only max
(`iggy_partition.rs:680-682`) makes it survive an operator deleting the
offending segment.
**Concretely, and independent of the absorption: the walk never compares
`header.partition_id` to the partition it is recovering.** A partition-7 batch
appended to partition 1's tail recovers `end_offset = 9000002`. It is adopted
with **no gap** too — `end_offset = 3` for a foreign batch that continues the
chain exactly, in the indexed arm (header-trusted, never verified) *and* in the
index-less arm (verifies its own checksum, continues the chain). So this does
not live in the gap branch. The produce path checks exactly this thirty lines
from its decode (`server_common/src/send_messages.rs:471`,
`batch.header.partition_id != namespace.partition_id() as u64` →
`InvalidCommand`); recovery — the path that truncates and re-seeds the offset
space — is the only reader that skips it. One comparison per arm, on every
batch, immediately after `peek_header` and before the offset split, with a
distinct refusal rather than `break`: a foreign record mid-log is not a torn
tail, and pres
erving evidence is this module's policy for that class. Note it authenticates
the partition component alone, not the packed namespace, so it catches
partition-7-into-partition-1 but not stream-2/topic-1/partition-1 into
stream-1/topic-1/partition-1. Necessary and cheap, not a complete placement
authenticator.
**The regression arm still refuses without verifying, so one bit gets
opposite verdicts by direction.** `base_offset` 101 → 97 gives
`OffsetDiscontinuity { expected_offset: 101, found_offset: 97, position: 7528
}`, a refusal that never lifts at `replica_count = 1`; the same bit upward
truncates cleanly and the node boots. Reproduced independently by two reviewers
with matching numbers. The verify you just added is exactly the tool that
classifies this — it is applied to one branch only.
**The new `break` also has a third outcome its comment does not name.** When
the failing gap batch is the first thing past the last index entry,
`walked_any` stays false and the `IndexLogDivergence` return at `:1069` fires
*before* `refuse_if_survivor_past_damage` ever runs — so neither claimed
outcome happens: not the torn-tail truncate, and not the `InteriorDamage`
refusal even with a verifying batch sitting past the damage. Both sub-cases
were run; both give `IndexLogDivergence { end_offset: 0, indexed_size_bytes: 0
}`, whose operator text at `server_error.rs:388-391` reads "the {N}-byte log
holds no whole batch" about a log that holds a whole batch whose header decoded
and whose checksum failed. That wants its own refusal variant. The verdict
itself is correct and non-destructive — bytes preserved, nothing truncated —
but it is a behaviour change: under `cfd2265e1` those bytes were absorbed and
served.
## Suggested shape — one change rather than four
Hoist the verify above **both** branches, keep `break` on failure so the
probe classifies the damage, and refuse only a *verified* discontinuity in
either direction:
```rust
if header.base_offset != expected_offset {
let verifies = /* slice_at + decode_batch_slice, once, above both
branches */;
if !verifies {
break; // damage -> probe
classifies
}
return
Err(identity.refusal(PartitionRecoveryRefusal::OffsetDiscontinuity { .. }));
}
```
That removes the adoption, removes the direction asymmetry, and leaves the
indexed arm's policy identical to the index-less arm's (`:1122`
break-on-`!verifies`, `:1131` refuse-on-`!=`) — which closes the
two-arms-disagree finding as a side effect. The only newly-truncatable bytes
are bytes that failed their own checksum *and* have nothing verifying after
them, which is precisely the torn tail this module exists to truncate. No
verified byte becomes truncatable. And a verified regression means genuinely
duplicated committed offsets, so that refusal stays and becomes meaningful
instead of a coin flip on flip direction.
**Why not a plain revert, which is what I originally suggested:** refusing
any `!=` sends that upward single-bit flip from "truncates cleanly, node boots"
to a permanent `replica_count = 1` tombstone. On the shipped `enforce_fsync =
false` that is a routine torn tail. A plain revert is strictly worse than the
current head on this class — the verify is what makes the revert safe, and
hoisting it means the revert keeps the useful half of this commit instead of
discarding it.
## Two smaller notes
**Neither new fixture exercises the property the commit body names.** Both
flip `corrupt[COMMAND_HEADER_SIZE + 4]`, i.e. blob byte 4, which sits inside
frame 0's stored per-message checksum field (`send_messages.rs:142-147` writes
id at `8..24`, deltas at `24..32`, lengths at `32..40`, leaving `0..8` for the
checksum). So `verify_and_recompute_batch_checksum` returns
`InvalidMessageChecksum` before `header.batch_checksum` is ever compared —
measured `InvalidMessageChecksum(17987071420115200071, 17987072171734476871,
5)`, where a real `base_offset` flip gives `InvalidBatchChecksum(...)`.
Deleting `base_offset` from `write_batch_header_fields` would leave both new
tests green. `base_offset` is header bytes `8..16`, so flipping inside
`corrupt[8..16]` after `encoded_batch` stamps runs the per-message pass clean
and then trips the batch comparison on `base_offset` specifically. Two fixtures
are needed, since `corrupt[8] ^= 0x04` is the downward case and the absorb path
needs an upwa
rd one.
**`:325`'s `end_offset - start_offset + 1` and `:461`'s `previous.end_offset
+ 1` are plain adds behind this branch.** An absorbed `encoded_batch(u64::MAX,
1)` panics `attempt to add with overflow` at `:325:44` in debug and wraps to a
zero message count in release. That one is no longer flip-reachable — the gate
did close it — but it wants `checked_add` regardless.
## What stands verbatim from the original finding
The span count at `:325` still over-counts across a *verified* absorbed gap
(`end_offset = 1003`, `messages_count = 1004`, for four real messages), with a
second site at `state_transfer.rs:2563-2568`. And `walk_segment_payload`
(`state_transfer.rs:762-767`) still returns `NonContiguous` for **any** gap,
forward included, so a segment carrying an absorbed one can never be installed
by a peer — `spill_transfer_segment:1723` and `adopt_staged_segment:1789` fail
on it on every attempt, forever. That is the part tightening the gate cannot
fix: a more selective gate fires less often, but its output is still a segment
shape the rest of the system refuses to handle and has no repair route for.
Also for the record: at `replica_count = 1` this branch has no legitimate
input at all. The frontier-stamp shape it exists to serve cannot be minted
there — all three `ProbeAsBackup` sites gate on `replica_count > 1`
(`bootstrap.rs:2410`, `bootstrap.rs:2607`, `partition_helpers.rs:594` via
`restarted` at `:547`), and the only advancing frontier writers are
`state_transfer.rs:2044`/`:2047`. Mid-chain, an absorbed gap is at least caught
loudly by the chain guard (`Hole { previous_end: 1003, next_start: 2 }`); it is
the tail segment that reaches service, and the tail is where `end_offset` seeds
`current_offset` under an advance-only persist.
One route I checked and can rule out, so it does not get chased: superblock
rot in `offset_frontier` is **not** an entry point. The record carries a
trailing `XxHash3_64` verified on read (`journal/src/superblock.rs:408-415`)
with ping-pong fallback to the partner slot.
##########
core/server/src/segment_recovery.rs:
##########
@@ -438,106 +1052,1538 @@ async fn recover_segment_bounds(
// MID-CHAIN segment too, not just the tail. Recovering that as empty
// then trips the contiguity guard and refuses the whole partition:
// total serve loss (and offset reuse from 0) for a chain whose bytes
- // are all present. The walk stops at the first header that does not
- // decode or does not fit, which keeps the torn-tail truncation the
- // indexed path performs.
+ // are all present. The walk keeps the torn-tail truncation the indexed
+ // path performs, and rebuilds the index from the batches it proves so
+ // a sealed segment does not pay a full-scan poll penalty forever.
_ if messages_size > 0 => {
- // Opened once, as above. Nothing walked means no whole batch,
- // which is the `Ok(None)` the tail of this arm already returns.
- let messages = fs::File::open(messages_path).ok();
+ let messages = open_messages_file(identity, messages_path)?;
+ let mut scanner = FileScanner::new(&messages, messages_size,
scratch);
let mut position = 0u64;
let mut start_timestamp = None;
let mut end_offset = start_offset;
let mut end_timestamp = 0;
let mut expected_offset = start_offset;
- let mut scratch = Vec::new();
- while let Some(messages) = messages.as_ref()
- && position < messages_size
- {
- let Some(header) = read_batch_header(messages, position,
messages_size) else {
- break;
+ let mut rebuilt_index = Vec::new();
+ let mut last_indexed_position: Option<u64> = None;
+ while position < messages_size {
+ let header = match scanner.peek_header(position) {
+ Ok(Some(header)) => header,
+ Ok(None) => break,
+ Err(source) => {
+ return Err(scan_read_failure(identity, messages_path,
&source));
+ }
};
let extent = position.saturating_add(header.total_size() as
u64);
if extent > messages_size {
break;
}
- // The FILENAME is the only trustworthy anchor once the index
is
- // gone, and `read_batch_header` checks a length, not a
checksum.
- // A torn header claiming an offset below `start_offset` would
- // underflow the message count the caller derives; one
claiming a
- // jump above becomes this partition's counter, and the next
- // prepare stamps a `base_offset` diverged from every peer. So
- // the chain has to be contiguous from the filename onward, and
- // the batch has to verify before its header is believed.
- if header.base_offset != expected_offset
- || !batch_verifies(messages, position, &header, &mut
scratch)
- {
+ // The FILENAME is the only trustworthy anchor once the index
+ // is gone, and the header decode checks a length, not a
+ // checksum. So the batch has to verify before its header is
+ // believed, and the chain has to be contiguous from the
+ // filename onward.
+ let verifies = scanner
+ .slice_at(position, header.total_size())
+ .map_err(|source| scan_read_failure(identity,
messages_path, &source))?
+ .is_some_and(|batch| decode_batch_slice(batch).is_ok());
+ if !verifies {
break;
}
+ if header.base_offset != expected_offset {
+ // A batch that VERIFIES but does not continue the chain is
+ // durable data past a hole (or a duplicated range): the
+ // offsets in between are exactly what a truncation here
+ // would silently erase, so refuse instead.
+ return Err(
+
identity.refusal(PartitionRecoveryRefusal::OffsetDiscontinuity {
+ start_offset,
+ expected_offset,
+ found_offset: header.base_offset,
+ position,
+ }),
+ );
+ }
if header.message_count > 0 {
end_offset = header
.base_offset
.saturating_add(u64::from(header.message_count) - 1);
end_timestamp = header.base_timestamp;
start_timestamp.get_or_insert(header.base_timestamp);
expected_offset = end_offset.saturating_add(1);
+ if last_indexed_position.is_none_or(|indexed| {
+ position.saturating_sub(indexed) >=
REBUILT_INDEX_STRIDE_BYTES
+ }) {
+ push_index_entry(
+ &mut rebuilt_index,
+ header.base_offset,
+ header.base_timestamp,
+ position,
+ );
+ last_indexed_position = Some(position);
+ }
}
position = extent;
+ 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()
+/// 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(())
}
-fn read_batch_header(
- messages: &fs::File,
- position: u64,
+/// 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.
+///
+/// The residue is deliberately NOT width-gated: a torn flush chunk is
+/// bounded by the CHUNK, not by one record, and with `enforce_fsync = false`
+/// delayed allocation routinely extends a file far past its written-back
+/// pages, leaving hundreds of MiB of zeros behind one crash. That is the
+/// canonical torn tail this module exists to truncate, so every residue is
+/// probed whole. What bounds the probe instead is the per-candidate work
+/// budget, whose exhaustion REFUSES and keeps the bytes rather than
+/// truncating: past the limit the probe has proven nothing, and the cheapest
+/// input to construct must never earn the destructive verdict.
+async fn refuse_if_survivor_past_damage(
+ identity: PartitionIdentity<'_>,
+ scanner: &mut FileScanner<'_>,
+ messages_path: &str,
+ damage_position: u64,
messages_size: u64,
-) -> 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;
+ scanner.budget.grow_for_residue(residue_bytes);
+ match scanner
+ .probe_for_survivor(damage_position, chain_end_offset, start_offset)
+ .await
+ .map_err(|source| scan_read_failure(identity, messages_path, &source))?
+ {
+ ProbeOutcome::Survivor { position } => {
+ Err(identity.refusal(PartitionRecoveryRefusal::InteriorDamage {
+ start_offset,
+ damage_position,
+ survivor_position: position,
+ }))
+ }
+ ProbeOutcome::BudgetExhausted => Err(identity.refusal(
+ PartitionRecoveryRefusal::UnverifiedResidue {
+ start_offset,
+ damage_position,
+ residue_bytes,
+ candidates_examined: scanner.budget.spent_units,
+ budget_units: scanner.budget.limit_units,
+ },
+ )),
+ ProbeOutcome::NoSurvivor => Ok(()),
+ }
+}
+
+/// Verdict of the damage probe over the residue past the walked prefix.
+/// `NoSurvivor` is the only verdict that permits truncation; running out of
+/// budget is deliberately NOT folded into it, so a residue that is expensive
+/// to scan refuses (keeping the bytes) instead of earning the destructive
+/// outcome.
+enum ProbeOutcome {
+ /// A complete, checksum-verifying batch starts at this position.
+ Survivor { position: u64 },
+ /// The whole residue was scanned and nothing in it verifies.
+ NoSurvivor,
+ /// The scan budget ran out before the residue was classified.
+ BudgetExhausted,
+}
+
+/// Forward-only buffered reads over one segment file for the recovery walk
+/// and the damage probe. Parsing and checksumming happen against an in-memory
+/// window so neither pays a syscall per batch -- the probe advances its
+/// candidate one byte at a time, and per-candidate preads would turn one
+/// damaged multi-GiB segment into a boot-length stall.
+///
+/// Synchronous `std::fs` on purpose, like every mutation in this module: the
+/// boot path's runtime sizes its blocking pool at zero and recovery must not
+/// depend on `io_uring` opcode coverage. Only the sparse-index bound reads go
+/// through the async `IggyIndexReader`.
+struct FileScanner<'scan> {
+ file: &'scan fs::File,
+ file_len: u64,
+ window: &'scan mut Vec<u8>,
+ window_start: u64,
+ spill: &'scan mut Vec<u8>,
+ budget: &'scan mut ProbeBudget,
+ refilled: bool,
+}
+
+impl<'scan> FileScanner<'scan> {
+ fn new(file: &'scan fs::File, file_len: u64, scratch: &'scan mut
ScanScratch) -> Self {
+ let ScanScratch {
+ window,
+ spill,
+ probe_budget,
+ } = scratch;
+ window.clear();
+ Self {
+ file,
+ file_len,
+ window,
+ window_start: 0,
+ spill,
+ budget: probe_budget,
+ refilled: false,
+ }
+ }
+
+ /// True when the scanner hit disk since the last call. The async scan
+ /// loops yield to the reactor once per window of work on it: recovery
+ /// runs in front of the bootstrap barrier with the blocking pool sized
+ /// at zero, so an unyielding walk over a damaged multi-GiB chain would
+ /// pin the shard core -- signal handling included -- until it finishes.
+ fn take_refilled(&mut self) -> bool {
+ std::mem::take(&mut self.refilled)
+ }
+
+ /// Bytes `[position, position + len)`, or `None` when they run past the
+ /// end of the file.
+ fn slice_at(&mut self, position: u64, len: usize) ->
io::Result<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.
+ // Callers only pass lengths from headers that already passed the
+ // plausibility cap, which is what bounds this resize.
+ self.spill.resize(len, 0);
+ self.file.read_exact_at(&mut self.spill[..], position)?;
+ self.refilled = true;
+ return Ok(Some(&self.spill[..]));
+ }
+ let window_end = self.window_start + self.window.len() as u64;
+ if position < self.window_start || end > window_end {
+ let fill = usize::try_from((self.file_len -
position).min(SCAN_WINDOW_CAPACITY as u64))
+ .unwrap_or(SCAN_WINDOW_CAPACITY);
+ self.window.resize(fill, 0);
+ self.file.read_exact_at(&mut self.window[..], position)?;
+ self.window_start = position;
+ self.refilled = true;
+ }
+ // In-window by the branch above, and the window is capacity-bounded,
+ // so the try_from cannot fail.
+ let start = usize::try_from(position - self.window_start).unwrap_or(0);
+ Ok(Some(&self.window[start..start + len]))
+ }
+
+ /// The batch command header at `position`, or `None` when it does not fit
+ /// the file, does not decode (torn header, garbage bytes), or claims a
+ /// size no legal batch can reach. The size check runs BEFORE any caller
+ /// slices the claimed extent: an oversized claim cannot be a real batch,
+ /// so treating the header as undecodable is verdict-identical to reading
+ /// the claimed bytes and failing the verify, and it keeps one bit-flipped
+ /// length field from driving a claimed-size allocation and read.
+ fn peek_header(&mut self, position: u64) ->
io::Result<Option<BatchHeader>> {
+ let Some(bytes) = self.slice_at(position, COMMAND_HEADER_SIZE)? else {
+ return Ok(None);
+ };
+ Ok(BatchHeader::decode(bytes)
+ .ok()
+ .filter(|header| header.total_size() as u64 <=
MAX_RECOVERABLE_BATCH_BYTES))
+ }
+
+ /// Probes the residue for the first complete, checksum-verifying batch
+ /// starting after `damage_position`.
+ ///
+ /// Batch starts are byte-aligned (appends write exact-sized records with
+ /// no padding) and the damaged region's own lengths cannot be trusted, so
+ /// every byte offset is a candidate. Candidates are scanned inside the
+ /// loaded window and the window advances sequentially -- refilled at the
+ /// first candidate whose header no longer fits, re-reading at most one
+ /// header of overlap -- so each residue byte is read O(1) times instead
+ /// of once per candidate. The header decode pre-filters candidates
+ /// cheaply (204 reserved bytes must be zero), and offset sanity plus
+ /// length bounds run before a verify is paid, so the checksum only runs
+ /// on byte positions that already look like a plausible chain
+ /// continuation.
+ ///
+ /// Each candidate examined is charged one unit against the shared probe
+ /// budget -- examined, not verified: examining is flat-cost (zeros bail
+ /// on the undersized length, garbage on the first nonzero reserved
+ /// byte), so with the budget sized per residue byte an honest
+ /// front-to-back scan always fits, at any residue width, and total probe
+ /// work stays linear in the residue by construction. Window refills and
+ /// verify slices are deliberately not charged; they are already bounded
+ /// by the strictly-forward window advance and the plausibility cap on
+ /// claimed sizes. Exhaustion returns
+ /// [`ProbeOutcome::BudgetExhausted`], never `NoSurvivor`.
+ async fn probe_for_survivor(
+ &mut self,
+ damage_position: u64,
+ chain_end_offset: Option<u64>,
+ start_offset: u64,
+ ) -> io::Result<ProbeOutcome> {
+ let header_len = COMMAND_HEADER_SIZE as u64;
+ // The bytes AT the damage already failed to decode or verify, so the
+ // first candidate starts one past them.
+ let mut candidate = damage_position.saturating_add(1);
+ while candidate.saturating_add(header_len) <= self.file_len {
+ self.fill_window_at(candidate)?;
+ let window_end = self.window_start + self.window.len() as u64;
+ while candidate.saturating_add(header_len) <= window_end {
+ if !self.budget.charge_candidate() {
+ return Ok(ProbeOutcome::BudgetExhausted);
+ }
+ // In-window by the loop bound, and the window is
+ // capacity-bounded, so the try_from cannot fail.
+ let at = usize::try_from(candidate -
self.window_start).unwrap_or(0);
+ if let Ok(header) = BatchHeader::decode(&self.window[at..at +
COMMAND_HEADER_SIZE])
+ {
+ let advances_chain = chain_end_offset
+ .map_or(header.base_offset >= start_offset,
|chain_end| {
+ header.base_offset > chain_end
+ });
+ let total_size = header.total_size();
+ // The plausibility cap, not just the file length: with no
+ // width gate on the residue, this is what keeps one
+ // corrupted-upward length claim from driving a
+ // claimed-size spill allocation and read.
+ let fits = total_size as u64 <= MAX_RECOVERABLE_BATCH_BYTES
+ && candidate.saturating_add(total_size as u64) <=
self.file_len;
+ if advances_chain && fits && header.message_count > 0 {
+ let batch = self.verify_slice(candidate, total_size)?;
+ if decode_batch_slice(batch).is_ok() {
+ return Ok(ProbeOutcome::Survivor {
+ position: candidate,
+ });
+ }
+ }
+ }
+ candidate += 1;
+ }
+ if self.take_refilled() {
Review Comment:
**Blocker: the yield is outside the candidate loop, so every verify spill
read in a window lands in one un-preemptible stretch.**
`take_refilled()` is polled here, after the inner candidate `while` closes.
`verify_slice` sets `refilled = true` (`:1564`) but nothing reads it until this
point, so a window that performs N spill reads yields once, after all N.
Modeled maximum un-yielded synchronous `pread64`: **256 GiB** (bait every 4
KiB, 256 MiB claims), **4 TiB** at the 256 B pitch.
Recovery runs in front of `BootstrapBarrier` with the blocking pool sized at
zero (`server_common/src/executor.rs:88`), so that stretch pins the shard core
with signals unserviced — Ctrl-C included. Which means the "preemptible" half
of `0929f697b`'s own title does not hold, independently of the volume problem
in my comment on `charge_candidate` above.
Worth noting both walk arms get this right — `:1000` and `:1126` check per
batch. Only the probe defers to the window boundary, which reads as an
oversight rather than a policy.
Suggested fix: also yield inside the candidate loop after a verify that hit
the spill path. `take_refilled` is a `mem::take`, so the existing outer-loop
call and a new inner one cannot double-yield on the same read, and the outer
one still covers windows that scanned without verifying.
Two things this fix cannot do, so they are worth separating out:
- **Accounting will never close the residual.** A single large synchronous
`pread` is inherently un-preemptible, and this module uses synchronous
`std::fs` deliberately. So a bounded budget shrinks how many such reads happen
but not how long one takes. The end state that fixes it properly is a chunked
verify — stream the checksum through the 4 MiB window instead of spilling —
which also drops the per-shard spill high-water from 256 MiB to 4 MiB with no
change to which shapes are classifiable. That needs a
`verify_and_recompute_batch_checksum` that accepts non-contiguous input, so it
is a `binary_protocol` follow-up rather than part of this fix.
- **Separately, `OFFER_HASH_CHUNK_LEN` is now mis-tuned.** The two yields in
`state_transfer.rs` (`:748`, `:2979`) were dead before this PR
(`sleep(Duration::ZERO)`, 32 ns) and are now real (11.5–13.0 µs measured). At a
1 MiB chunk that is 1024 yields/GiB against a 20.8 µs hash per chunk — **+55%
to +63%** on `verify_state_artifact_yielding`, and +39% to +59% on the sender's
segment walk. That is a runtime path, not boot-once. Raising it to 4 MiB brings
the overhead to +14–16%, keeps the un-yielded stretch a bounded 82 µs CPU pass,
and matches `SCAN_WINDOW_CAPACITY`. One constant, in a file this PR already
touches.
---
> **Addendum after `c727a1cc9`.** The new `!verifies` break in the indexed
arm's gap branch gives this finding one more entry route: shapes that
previously absorbed a header-decodable forward gap, and so never entered
`refuse_if_survivor_past_damage` at all, now hand it a residue starting at the
gap batch. The finding itself is unchanged — same cost, same fix — but the
probe is now reachable from the indexed arm too, not only from a walk that ran
out of decodable batches.
>
> Also measured on the new head, so it is not confused with this finding:
the gap verify that `c727a1cc9` adds at `:1019-1022` is charged against no
budget either, but it **is** bounded by construction. Walk positions strictly
increase, each verify covers that batch's own extent, and `extent >
messages_size` breaks before the slice — so the extents are disjoint and total
verified bytes cannot exceed the walked span. That is the opposite of the
situation here: the probe advances `candidate` by one byte at a time, so its
verify ranges *overlap*, and the overlap is the whole defect. The two are not
the same fix, and the walk-side one needs no charge.
##########
core/server/config.toml:
##########
@@ -463,6 +463,26 @@ archive_expired = false
# Unsupported: setting this to `true` aborts boot.
recreate_missing_state = false
+# At boot, segment recovery walks each partition's segments: bytes after the
+# last decodable batch of a genuinely torn tail are physically truncated from
+# the .log/.index files, and the index is rebuilt when it was damaged. Only
+# the walk of a segment whose index was lost re-checksums batches; with an
+# intact index the walk trusts batch headers (decodable, contiguous offsets),
+# and bytes before the last index entry are not re-examined at boot at all --
+# at-rest damage there surfaces on the read path via validate_checksum.
+# Damage in the middle of a segment, or trailing bytes too large or costly to
Review Comment:
> **Updated after `7a9c4b3f0` and `c727a1cc9`.** Two things changed here.
`7a9c4b3f0` adds a second true rebuild trigger — an index that is *absent*
altogether — which makes the `:468` clause below more wrong, not less, and
introduces an operator-visible boot-cost change this block did not mention.
`c727a1cc9` keeps the forward-gap absorption behind a checksum, so **Swap A is
now the live wording rather than the alternative** — folded into the
replacement block, revised for the gate. Clause list and replacement block both
updated below.
**Blocker: four clauses in this paragraph describe behaviour the code does
not have. It is the only operator-facing description of a destructive boot path
and the repo has no `docs/`.**
1. **`:473-474` "trailing bytes too large or costly to prove torn, is never
silently truncated: the partition is refused"** — both halves are dead.
`0929f697b` deleted the width gate, and the commit's own new test
`given_wide_zeros_residue_when_recovering_should_truncate_at_break` asserts a
512 KiB residue *truncates*. "Costly to prove torn" is dead too, because
`BudgetExhausted` is unreachable by any on-disk shape (see my comment on
`charge_candidate`). Note this clause becomes live again if the budget fix
lands as a `SCAN_WINDOW_CAPACITY` refusal rather than a verify charge, so the
wording depends on which fix ships.
2. **`:470` "with an intact index the walk trusts batch headers (decodable,
contiguous offsets)"** — the indexed arm now absorbs forward gaps while the
index-less arm refuses any `!=`. The asymmetry is documented nowhere
operator-facing. Dies if the absorption is reverted.
3. **`:477-480` "the refused files stay at their original paths … re-derived
and re-logged on every boot"** — not true for `StorageSizeMismatch`, which is
raised in pass C *after* `truncate_to` already ran, and
`bootstrap.rs:1958-1972`'s own comment concedes the next boot accepts the
chain. That tombstone lasts one process, not until an operator intervenes.
4. **`:480` "single-replica refusals with no recoverable bytes at stake (a
hole from a stray file, an orphaned empty segment)"** — neither verdict
establishes that; see my comment on the reconciler gate. Both can fence a fully
populated chain and serve it empty.
Also missing entirely: **`.fenced.N` has a second producer on a partition
that recovers fine.** `fence_unrecoverable_segment_files`
(`segment_recovery.rs:742`) moves a tail segment whose bytes decode to nothing
into the same directory name and recovers that segment empty *while the
partition keeps serving* — one fresh directory per such segment, so
`.fenced.0`, `.fenced.1`, … accumulate under a healthy partition. That is the
one outcome in this feature that loses data and keeps serving, and the block
currently teaches the opposite reading of a fence directory. This was asked for
last iteration and is still not named.
**Plus, new at `c727a1cc9` / `7a9c4b3f0`:**
5. **`:468` "the index is rebuilt when it was damaged" is now more wrong.**
It was already wrong about *damaged* — a damaged-but-nonempty index is floored
to whole entries or refuses the partition; a rebuild happens only in the arm
reached when the index holds no whole entry. `7a9c4b3f0` adds a second genuine
trigger, *absent altogether*, which the sentence also does not name. So the set
of rebuild triggers grew and the one case the sentence names is still not one
of them.
6. **The block says nothing about which arm absorbs a forward gap**, and
after `c727a1cc9` that is the operator-visible part: whether you boot or
tombstone depends on whether the index survived the same crash. The index-less
walk refuses any offset gap; the indexed walk now refuses only a regression and
absorbs a checksum-verified forward gap.
7. **The missing-index boot cost is undocumented.** A restore that dropped
every `.index` now boots by re-reading and re-checksumming every byte of every
segment and rebuilding each index durably — two fsyncs per segment
(`stage_rebuilt_index`'s `sync_all` at `:684`, `install_rebuilt_index`'s
`fsync_dir` at `:712`), measured ~6.0 ms per barrier, so roughly 1-2 minutes of
fsync alone at 10k segments on top of a full checksummed walk at 46-114 ms/GiB.
Before, that shape refused the boot outright. Worth a line, because a boot that
looks hung after such a restore is doing exactly this work. (Follow-up for the
code rather than the doc: those `fsync_dir` calls all target the same
`partition_path`, so one barrier after all installs would halve them.)
## Replacement for `:466-484`
Written against the head as shipped — i.e. absorption kept behind the
checksum gate. If the crew's recommendation on the gap branch lands (verify
hoisted above both branches, adoption dropped), swap the third paragraph for
**Swap C** below.
```
# At boot, segment recovery walks each partition's segments: bytes after the
# last decodable batch of a genuinely torn tail are physically truncated from
# the .log/.index files. A LOST index is rebuilt from the batches the walk
# proves -- lost meaning absent altogether, or holding no whole 24-byte
# entry. An index that still holds whole entries is NEVER rebuilt, only
# floored to whole entries, or the partition is refused when its entries
# contradict the log. Only the walk of a segment whose index was lost
# re-checksums every batch; with an intact index the walk trusts batch
# headers, except that a batch opening a forward offset gap must pass its
# batch checksum before the walk adopts its offset. Bytes before the last
# index entry are not re-examined at boot at all -- at-rest damage there
# surfaces on the read path via validate_checksum.
#
# A restore that dropped every .index therefore boots by re-reading and
# re-checksumming every byte of every segment and rebuilding each index
# durably, two fsyncs per segment: budget minutes, not seconds, on a
# partition holding thousands of segments. A boot that looks hung after such
# a restore is doing that work. Before, that same shape refused the boot
# outright.
#
# The two walks disagree on a forward offset gap, and which one runs is
# decided by whether the index survived the same crash. The index-less walk
# refuses ANY offset gap. The indexed walk refuses only a gap that
# REGRESSES; a forward gap whose opening batch passes its batch checksum is
# absorbed, and the segment's end offset then covers offsets no batch in it
# holds. That checksum proves the batch was minted by a server and not
# altered since -- NOT that it belongs at this position, in this segment, or
# in this partition: the walk does not compare a batch's own partition_id
# against the partition it was found in, so an intact batch that lands here
# from anywhere is adopted along with its offsets. Reported message counts
# for that topic (/stats, GetTopic messages_count) are inflated by the width
# of the gap, and a peer can never install the segment, because state
# transfer's own walk still refuses any gap -- so a partition that boots
# clean with one "forward offset gap" warning is permanently unrepairable
# from a peer. A gap whose opening batch FAILS its checksum is damage: the
# walk stops there and the bytes from that point on go through the damage
# probe like any other residue.
#
# Damage in the middle of a segment is never silently truncated: the
# partition is refused. With peer replicas the refused files are moved to a
# .fenced.N directory beside the partition, which is rebuilt empty and
# refilled from a peer. With replica_count = 1 there is no peer, so nothing
# is moved and nothing is rebuilt: the refused files stay at their original
# paths, the partition is tombstoned, and the same refusal is re-derived and
# re-logged on every boot until an operator intervenes. A tombstoned
# partition is unrouted -- clients get the retriable TransientNotAccepted
# status, never an empty poll that would read as a healthy empty partition.
# One refusal is not durable: a length divergence found when a writer
# reopens a file recovery just truncated clears on the next boot, so that
# tombstone lasts only for the life of the process.
#
# .fenced.N has a second producer, on a partition that recovers FINE: a tail
# segment holding bytes that decode to nothing anywhere is moved there and
# the segment recovered empty, so the partition serves without those bytes.
# One fresh directory per such segment, so .fenced.0, .fenced.1, ... can
# accumulate under a healthy partition. .fenced.N alone therefore does not
# mean a partition was refused -- grep the boot log for "refusing the
# recovered segment chain", which names the directory and the reason.
#
# The metadata WAL truncates genuinely torn tails too; interior WAL damage
# or oversized trailing bytes refuse boot instead.
```
**Swap C** — if the gap branch is fixed as recommended (verify hoisted above
both branches, adoption dropped, `partition_id` checked), replace the whole
third paragraph with:
```
# Either walk refuses a batch whose base offset does not continue the chain,
# or whose partition_id is not this partition's. A mismatch whose batch
# fails its own checksum is damage rather than data: the walk stops there
# and the residue goes to the damage probe, so a bit flip in a tail batch's
# offset truncates like any other torn tail instead of refusing the
# partition.
```
**Swap B** from the previous round (the rc=1 rebuild allowlist) is unchanged
and still applies. The `max_message_size` items — the `:962` bound, the HTTP
cap via `http.max_request_size`, Go's hard 64 MiB const at
`foreign/go/internal/vsr/header.go:30`, and the stale advice at `:942` — are
untouched by both new commits and still open.
--
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]
