hubcio commented on code in PR #3946:
URL: https://github.com/apache/iggy/pull/3946#discussion_r3841490873
##########
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:
fixed in e5e8712c2, your shape: verify hoisted above both branches,
unverified mismatch in either direction breaks to the probe (the downward flip
now truncates like the upward one), verified mismatch refuses
`OffsetDiscontinuity` both ways, absorption gone.
`partition_id` is compared in both arms, with one deviation: the check sits
behind the same checksum gate as the offset split. `partition_id` is header
bytes 0..8 and hashed like `base_offset`, so refusing an unverified
foreign-looking header would recreate the one-bit-two-verdicts asymmetry - flip
in offset truncates, flip in partition stamp tombstones. a batch that verifies
with a foreign stamp refuses as the new `ForeignBatch`; chain-continuing
batches with the right stamp still pay no verify.
the `walked_any = false` route now probes before the `IndexLogDivergence`
verdict, so a verifying survivor past the anchor refuses as `InteriorDamage`,
and the divergence text says "no batch that decodes and verifies". both
sub-cases have tests.
fixtures now flip header bytes 8..16 (one upward, one downward), so dropping
`base_offset` from `write_batch_header_fields` fails them. the span arithmetic
in the stats seed and the chain guard is saturating/checked.
the "stands verbatim" items die with the absorption: either walk refuses any
gap now, so no gap-bearing segment reaches service, the span count only sees
contiguous chains, and state transfer is never handed a shape its own walk
refuses.
##########
core/common/src/lib.rs:
##########
@@ -40,6 +40,17 @@ pub use
consumer_group_client_state::ConsumerGroupClientState;
/// partition id (those are small, dense, zero-based), so it can't collide with
/// a genuine end-of-partition empty poll, which echoes the real partition id.
pub const RESYNC_REQUIRED_PARTITION_SENTINEL: u32 = u32::MAX;
+
+/// Frozen ceiling on `message_bus.max_message_size`, the knob that caps a
+/// single framed wire message and with it the widest batch record any
+/// admission path can persist. Frozen rather than knob-derived because
+/// boot-time segment recovery sizes fixed scan and allocation limits from the
+/// widest LEGAL record: a limit read from the live knob would change meaning
+/// between boots and refuse partitions written under an older value. Config
+/// validation rejects a knob above this at boot; raising it is a
+/// compatibility decision, not a tuning change, since segments written under
+/// a larger value would exceed what recovery on an older build accepts.
+pub const MAX_MESSAGE_SIZE_UPPER_BYTES: u64 = 256 * 1024 * 1024;
Review Comment:
fixed in 35b9cdb08: `http.max_request_size` validated against the ceiling
when http is enabled, and `message_bus.validate()` moved ahead of the
artifact-floor check so the ceiling error surfaces on the first boot, not the
second. the const doc names both validated knobs and carries the
drain-and-re-produce note for data above the ceiling; config.toml gained the
ceiling note at `max_message_size` (including go's hard 64 MiB frame constant)
and the body-cap note at `max_request_size`.
two deliberate omissions: no batch-total check in
`IggyMessagesBatch::validate` - published crate, client-side behaviour change -
say the word and i'll add it reusing `TooBigMessagePayload`. and
`max_request_size` between ~4/3x the bus cap and the ceiling can still admit a
batch no peer accepts a frame for at `replica_count > 1`; documented at the
knob rather than enforced, because base64 slack means a hard cross-check
refuses legal configs (an 80 MiB body can carry a record under a 64 MiB bus
cap). a boot-time warn for that band is easy if you want it.
##########
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() {
Review Comment:
fixed in 348b79924, two counters as suggested: `charge_candidate` stays at 1
unit / 2x residue with refills uncharged, and a new verification counter
charges `total_size` per handed slice - in-window included - checked before the
read, so exhaustion never pays for the slice that broke it.
went with 4x residue plus the degradation: `BudgetExhausted` with
`chain_end_offset` none folds into recover-as-empty (pair fenced whole, no
tombstone minted), with a walked prefix it refuses, and `UnverifiedResidue`
reports both budgets now.
the tripwire you asked for is
`given_overlapping_verify_claims_when_probing_should_refuse_on_verify_budget` -
bait residue over a walked prefix, pinning `verify_budget_bytes == 4 *
residue_bytes` - plus a no-prefix twin for the recover-empty convergence.
##########
core/server/src/server_error.rs:
##########
@@ -162,21 +162,27 @@ pub enum ServerError {
expected: u128,
found: u128,
},
- // Per-partition, not fatal: the boot path fences this one group
(quarantines
- // its segment files and materialises it fresh) instead of taking the node
- // down for one damaged local chain. The shapes it reports are exactly
what a
- // failed state-transfer quarantine leaves behind, and the rebuild recovers
- // the data from a peer.
+ // Per-partition, not fatal: the boot path fences this one group instead of
+ // taking the node down for one damaged local chain. Only STRUCTURAL
+ // refusals route here -- shapes where the local files contradict
+ // themselves, so a retried boot cannot help. Transient recovery I/O
+ // failures (stat, open, read, truncate, fsync) stay node-fatal on purpose:
+ // a retried boot can still serve that partition, while fencing it would
+ // quarantine healthy data.
#[error(
- "partition {stream_id}/{topic_id}/{partition_id} at {dir} recovered an
\
- unusable segment chain: {reason}"
+ "partition {stream_id}/{topic_id}/{partition_id} at {dir} refused
segment \
+ recovery: {reason}. Boot moves the partition's segment files into a \
+ sibling `<partition dir>.fenced.N` directory and keeps them; with
peer \
Review Comment:
fixed in b2265c348 - the disposition claim is gone from the `#[error]`
entirely; a comment on the variant says the bootstrap arms own disposition,
since only they know which branch ran. the tombstone `error!` carries `%reason`
now, deliberately repeated so the grep line stands alone.
##########
core/server/src/partition_reconciler.rs:
##########
@@ -598,6 +599,25 @@ async fn reconcile_additions(
continue;
}
+ // Tombstoned without ever being materialised: a boot-time damage
+ // verdict (a refused segment chain, an untrusted superblock) fenced
+ // the namespace before any partition existed, so no teardown ran and
+ // no `ConfirmRemove` is coming to lift the tombstone. Building fresh
+ // would plant segment 0 over the refused files, truncating the
+ // oldest one, and the partition would then serve empty, hiding
+ // exactly the loss the tombstone surfaces. Deliberately uncounted:
+ // nothing lifts this state short of a metadata commit, and a commit
+ // bumps `Streams::revision`, which forces the next pass past the
+ // fast-skip.
+ if partitions.is_tombstoned(&ns) {
Review Comment:
fixed in 8a8c50473, both parts in one commit so no intermediate state ships
with the allowlist narrowed before the exit exists.
exit: `reconcile_removals` sweeps namespaces that are tombstoned and absent
from both the map and the committed target through `tear_down_owned_partition`
- disk delete first (fires only once metadata says the partition is gone), and
only its success enqueues the `ConfirmRemove` that lifts the fence, exactly to
avoid the bare-untombstone window you described. while the namespace is still
committed the fence holds. tests cover the exit, the hold, and the recycled-id
recreate coming up clean.
allowlist: went with the evidence-based form - `Hole` and
`EmptyNonTailSegment` carry the planned chain's walked byte total now, and rc=1
rebuilds only at `recoverable_bytes == 0`; a populated chain behind either
verdict tombstones until the operator-driven exit above.
agreed the `TransientNotAccepted` terminal discriminant + counter is a
follow-up (six sdk mirrors).
##########
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:
fixed in 348b79924: a yield inside the candidate loop after every verify
that hit the spill path; `take_refilled` is a take, so the inner and outer
yields cannot double-fire on one read, and the outer still covers windows that
scanned without verifying.
`OFFER_HASH_CHUNK_LEN` raised to 4 MiB in c8107b2a6, with your overhead
numbers in the doc.
agreed on both carve-outs: the chunked verify needs
`verify_and_recompute_batch_checksum` over non-contiguous input, so it stays a
binary_protocol follow-up, and until it lands no accounting closes the
single-large-pread residual.
##########
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:
replaced in 689aa64cf with your block, swap C as the third paragraph since
the gap branch landed as recommended, re-worded where this round's other fixes
moved the target again:
- "costly to prove torn" is live again, phrased for the verify-budget fix
that shipped, with the recover-empty convergence for residue after a walk that
proved nothing.
- the tombstone paragraph names both exits that now exist: the
zero-recoverable-bytes rebuild and the operator delete clearing the fence via
the reconciler.
- the lost-index rebuild triggers (absent, or no whole entry), the
floor-never-rebuild rule, and the missing-index full-rescan boot cost paragraph
are in.
- the second .fenced.N producer is named, with the grep line that
disambiguates.
the max_message_size items landed in 35b9cdb08 in the same push. the
fsync_dir batching (one barrier after all index installs) is accepted as a code
follow-up.
##########
core/server_common/src/reactor_yield.rs:
##########
@@ -0,0 +1,119 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! A yield that is guaranteed to suspend.
+//!
+//! Long CPU passes (recovery walks, artifact hashing) hand the core back to
+//! the reactor by awaiting a short timer. The runtime's timer wheel registers
+//! a timer only when its deadline is still in the future at the wheel's OWN
+//! clock re-read, and its sleep future completes on the first poll without
+//! ever suspending when registration is refused -- so a fixed short duration
+//! is a race against the code path between the two clock reads, and the
+//! window is machine- and build-dependent (a debug-build cold path loses a
+//! 1 us head start essentially always). No constant wins that race; the only
+//! deterministic shape is to retry with a growing duration until one
+//! registration wins.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::time::Duration;
+
+/// First attempted timer duration. The common case: on a warm path one
+/// microsecond outlives the registration window and the first attempt wins.
+const YIELD_FIRST_ATTEMPT: Duration = Duration::from_micros(1);
+
+/// Ceiling for the attempt doubling. Reaching it would mean a whole-second
+/// deadline was already in the past by the time the wheel re-read the clock:
+/// a broken or frozen clock, not a lost race. Registration is guaranteed
+/// long before; the cap only keeps the retry loop's growth finite.
+const YIELD_ATTEMPT_CAP: Duration = Duration::from_secs(1);
+
+/// Hands the core back to the reactor: the first poll ALWAYS returns
+/// `Pending` with a real timer registered, on any machine, by construction.
+///
+/// A registered timer with a near-now deadline fires on the reactor's next
+/// turn, so the attempted duration does not throttle the caller; it only has
+/// to be long enough to register. A bare self-waking yield is no
+/// alternative: this runtime does not reliably re-poll a task that wakes
+/// itself from inside its own poll, and a task parked that way may never
+/// resume.
+pub async fn yield_to_reactor() {
+ RegisteredYield {
+ registered: None,
+ attempt: YIELD_FIRST_ATTEMPT,
+ }
+ .await;
+}
+
+struct RegisteredYield {
+ /// The timer that won registration; later polls delegate to it.
+ registered: Option<Pin<Box<dyn Future<Output = ()>>>>,
+ attempt: Duration,
+}
+
+impl Future for RegisteredYield {
+ type Output = ();
+
+ fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<()> {
+ let this = self.get_mut();
+ if let Some(timer) = this.registered.as_mut() {
+ return timer.as_mut().poll(context);
+ }
+ loop {
+ // One allocation per attempt; at the callers' once-per-window
+ // cadence that is noise against the work being yielded from.
+ let mut timer: Pin<Box<dyn Future<Output = ()>>> =
+ Box::pin(compio::time::sleep(this.attempt));
+ if timer.as_mut().poll(context).is_pending() {
+ this.registered = Some(timer);
+ return Poll::Pending;
+ }
+ debug_assert!(
+ this.attempt < YIELD_ATTEMPT_CAP,
+ "a {:?} timer deadline was already in the past at
registration; \
+ the runtime clock is broken or frozen",
+ this.attempt
+ );
+ this.attempt = (this.attempt * 2).min(YIELD_ATTEMPT_CAP);
Review Comment:
fixed in c8107b2a6 with your rewrite as-is: 32-attempt bound, `poll!`-based
registration check, `error!` on exhaustion degrading to the pre-PR no-yield
behaviour, hand-written future and the `debug_assert` gone.
docs corrected to your measured numbers: the module doc now says the bare 1
us sleep wins the race and justifies the retry by the silent losses at the
walk's refill cadence, and the throttle sentence states the roughly-linear
regime above ~10 us. `YIELD_FIRST_ATTEMPT`'s own doc kept as-is per your note.
--
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]
