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:
**Blocker: this absorption should be reverted rather than patched. One line
closes four separate defects, and the shape it exists to accept cannot occur at
`replica_count = 1`.**
The comment above justifies absorbing with "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." I traced whether a single-replica
group can reach that state, and it cannot:
- `primary_index(view) = view % replica_count`, so at `replica_count = 1` a
solo replica is primary in every view.
- `init()` (`consensus/src/impls.rs:1387-1400`) arms `CommitMessage` for a
primary and `NormalHeartbeat` only for a backup, so a solo replica never arms
the timer whose expiry reaches `start_election` (`:2344`).
- The other two `start_election` call sites need `JoinMode::ProbeAsBackup`
(`:2253`) or a peer's `RequestStartView` (`:3013`), and all three
`ProbeAsBackup` selection sites gate on `replica_count > 1` —
`bootstrap.rs:2410`, `bootstrap.rs:2607`, and `partition_helpers.rs:547`, where
the guard is folded into `restarted` itself (`let restarted = replica_count > 1
&& metadata(...).is_ok()`).
- So `(view, log_view)` stays `(0, 0)` for a solo group's whole life,
`needs_superblock_persist()` is never true, and the only advancing frontier
writers (`persist_offset_frontier{,_at}`) are reached solely from
`state_transfer.rs:2047`/`:2100`. Purge writes only the safe direction —
`record_purge_frontier_reset` writes 0 before the unlinks.
So the frontier can never outrun the log at `replica_count = 1`. That makes
the trade lopsided in the wrong direction: absorption buys availability only at
`replica_count > 1`, where the pre-existing refusal was already lossless
(quarantine, rebuild, refill from a peer), and pays with a durable
unrecoverable corruption everywhere.
**The corruption, reproduced.** `base_offset` is not authenticated at this
point — the indexed arm's own TODO a few lines up says it trusts the header
decode alone. One flipped high bit gives `end_offset = 9223372036854775810`.
That value seeds `current_offset` (`bootstrap.rs:2688`), and `write_superblock`
(`iggy_partition.rs:679`) persists the frontier **advance-only**, so
`restore_offset_frontier` re-seeds it even after an operator deletes the
offending segment. There is no recovery short of wiping the superblock, and
`increment_messages_count` reports ~4.6e18 up the partition/topic/stream chain
meanwhile.
**And keeping absorption requires one rule to hold in five places**, any of
which drifting is invisible until a partition wedges:
1. this arm (`:989`)
2. the index-less arm (`:1092`), which still refuses any `!=`
3. `walk_segment_payload` (`state_transfer.rs:762-767`), which returns
`SegmentWalkError::NonContiguous` for **any** gap — so a segment carrying an
absorbed gap can never be installed by a peer, and `spill_transfer_segment`
(`:1744`) and `adopt_staged_segment` (`:1797`) fail on it on every attempt,
permanently
4. the span-based message count here (`:325`)
5. the same span formula on the install side (`state_transfer.rs:2563-2568`)
Relaxing `walk_segment_payload` to match is the wrong direction: it
validates peer-controlled bytes, while recovery validates local ones. And
rolling a fresh segment at the frontier instead is not the cheap alternative it
looks like — it trades the in-segment gap for a cross-segment `Hole` refusal on
the next boot.
Suggested fix: revert to `header.base_offset != expected_offset` and drop
the `>` warn branch. That closes this finding, the fabricated message count,
the state-transfer wedge, and the two-arm asymmetry, in one line, and leaves
the PR doing what its title says.
If the frontier-bump gap is a shape you have actually observed in the field
rather than a theorised one, say so — it changes the plan, and nobody reviewing
this can settle it from the source. The experiment that would: `SIGKILL` a node
under sustained produce, restart, produce, restart again, and count partitions
whose tail segment holds `base_offset > expected_offset`. One run, no new
tooling.
If absorption is kept regardless, then verifying the batch before believing
its header is mandatory, not optional — `write_batch_header_fields`
(`batch.rs:452-458`) does hash `base_offset` and `decode_batch_slice` verifies
it, so a checksum gate is what would make this arm's own comment ("every offset
adopted is one the primary durably promised") true rather than aspirational.
Cost is ~0.02 ms at the default flush cadence. But it still leaves items 3, 4
and 5 above.
##########
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:
**Blocker: this budget provably cannot fire, so the probe has no cost bound
at all — and the term that replaced the old one is uncharged.**
Candidates advance strictly monotonically from `damage_position + 1` with
`candidate += 1` as the only mutation, so a probe examines exactly `residue −
256` offsets against a limit of `2 × residue`. Modeled across every shape and
every segment count from 1 to 1024 × 1 GiB: **`spent/limit == 0.5000`,
always.** `UnverifiedResidue` is unreachable by any on-disk input, which is why
its own test has to pre-spend `u64::MAX / 2` to reach the path.
Meanwhile `verify_slice` (`:1517`) is charged nothing, and its spill read
runs up to `MAX_RECOVERABLE_BATCH_BYTES` = 256 MiB **per qualifying
candidate**. `0929f697b` deleted `spent_bytes += read_bytes + total_size`,
which was the term that bounded exactly this. The doc at `:1478-1481` says
verify slices are "already bounded by the strictly-forward window advance and
the plausibility cap on claimed sizes" — that sentence is true of *refills* and
false of *verifies*: `verify_slice` preads at an arbitrary position into
`spill` and never touches `window_start`, and the cap bounds each read's size,
not their number.
Measured on this branch:
| shape | read |
|---|---|
| 8 MiB residue, header every 248 B claiming 6 MiB | 49.5 GiB, **1.842 s**
(vs 0.012 s zero-filled) |
| 1 GiB residue, headers every 4 KiB claiming 8 MiB | 2,033 GiB |
| 1 GiB residue, 256 B pitch, 256 MiB claims | 786,433× — **768 TiB** |
| at the 1.25 GiB segment ceiling | ~810 TiB, **~8.6 h** |
The 248-byte pitch is exact and minimal: at period 248 the previous header's
zero-reserved region `[k+52, k+256)` ends precisely at the next header's
`base_offset` field. Producer payload bytes reach the residue verbatim with
encryption off by default, so this is admissible input rather than a crafted
file.
This compounds with two other properties into something worse than a slow
boot. At `replica_count = 1` a refusal is now a permanent tombstone that
re-derives its verdict on **every** boot, so an adversarial residue means the
node never boots again — re-inflicted on every restart attempt, from one write
to the data dir.
**Suggested fix — two counters, refills charged by neither:**
- **Enumeration:** keep `charge_candidate`, 1 unit per candidate, limit `2 ×
residue`. Structurally `residue − 256` against `2 × residue` = 50% margin at
every width, once refills come off it. (Charging refills is what collapsed the
honest margin: measured `spent = 2R − 4.19 MiB` at every width, where the 4.19
MiB is one scan window of walk-leftover luck — 3.12% margin at 64 MiB, ~0.195%
at the ceiling.)
- **Verification:** new counter, charge `total_size` per handed slice —
in-window ones included, since an in-window verify still hashes every message
up to the first bad checksum. Grown by the same `grow_for_residue` call,
residue-derived so knob-immunity is untouched. Check **before** the read, so
exhaustion never pays for the slice that broke the budget.
- **Limit:** either `4 × residue` plus `ProbeOutcome::BudgetExhausted if
chain_end_offset.is_none() => Ok(())`, or `64 × residue` with no degradation.
The degradation is sound because `chain_end_offset` is `None` exactly when the
walk proved no batch, and on that segment exhaustion and `NoSurvivor` reach an
identical outcome — `bounds == None` → `recovered_empty` → the pair moves to
`.fenced.N` and empties are seeded. Bytes preserved either way; refusing there
only adds a permanent tombstone. The looser bound avoids the degradation but
its constant is calibrated to one synthetic fixture. Both close the channel;
worst honest cost at `64R` on a 1 GiB residue is 64 GiB of hashing ≈ 1.3 s
against 4 TiB / ~88 s uncharged today.
**Do not** gate `fits` on `SCAN_WINDOW_CAPACITY` as a cheaper alternative.
That makes a legal >4 MiB survivor invisible, so `NoSurvivor` is returned and
`truncate_to` deletes it — silent destruction of committed data, reachable on
the shipped 64 MiB `max_message_size`.
One note on the test at `:2254`:
`given_zero_padded_records_when_probing_should_scan_whole_residue_and_recover_empty`
was named `..._should_refuse_on_scan_budget` before `0929f697b` renamed it and
inverted its body. It was the tripwire for this bound, so the rename recorded
the bound's removal as expected behaviour. Whatever limit lands, please add a
guard that asserts the handed-byte total against a multiple of `residue_bytes`
on a shape *with* a walked prefix — that is the test whose absence let this
through, and without it the next refactor removes the bound again.
##########
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.
##########
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:
**Blocker, and the second iteration this string has been flagged: it is now
actively false, and it renders.**
`cfd2265e1` made the single-replica damage path leave the refused files
exactly where they are — the branch at `bootstrap.rs:1947-1988` returns before
`quarantine_segment_files` and its own log line says "leaving the refused
segment files in place". This sentence still tells the operator boot "moves the
partition's segment files into a sibling `<partition dir>.fenced.N` directory
and keeps them".
It is not dead text. `recover_partition_segments` wraps the call in
`.map_err(|source| { error!(..., error = %source, ...); source })` at
`bootstrap.rs:2537-2545`, and `%source` is `Display` on a `thiserror` type, so
the full `#[error]` paragraph is formatted for every refusal raised inside
`load_persisted_segments` — which is nine of the ten construction sites,
including pass C's storage-open guard. Only `hydrate_reopen_error`'s site
escapes it (bare `?`, no wrapper). So a single boot logs `.fenced.N` and
"leaving the refused segment files in place" two lines apart, and the one
operator-facing string sends them to a directory that does not exist.
Suggested fix: drop the disposition claim from the `#[error]` entirely and
let the `bootstrap.rs` arms own it — they already log per-outcome and they are
the only place that knows which branch ran. If the sentence stays, it has to
branch on `replica_count` and on the quarantine's own success, which is more
conditional logic than an error string should carry.
While in here: the tombstone `error!` at `bootstrap.rs:1975-1983` carries no
`%reason`, and it is the line an operator greps to enumerate dark partitions.
The rest of the earlier logging ask is satisfied — the headline text is
accurate on both branches now and nothing is double-logged.
##########
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:
**Blocker: this gate has no exit, and the rc=1 allowlist routes around it.
Both need fixing, and the order matters.**
**The fence is unliftable through the API.** `untombstone` has exactly one
caller — `ConfirmRemove` (`shard/src/lib.rs:2231`) — only
`tear_down_owned_partition` enqueues it, and `reconcile_removals` (`:798`)
iterates `partitions.namespaces()`, which never contains a namespace that was
tombstoned before it was ever materialised. So nothing lifts a boot fence, and
nothing deletes the refused files either.
Reproduced: `DeleteTopic` + `CreateTopic` on the same ids leaves
`contains=false tombstoned=true routed=None` permanently. Slab keys recycle, so
the recreated stream/topic/partition gets an identical `ns` — meaning a
**freshly created, empty topic inherits a fence for bytes it never had**, and a
restart does not clear it because boot re-derives the refusal off the same
files. Compare the in-map branch at `:560`, which at least has
`has_pending_delete_failure` as an escape. The operator's only exit today is
moving files by hand.
**And the rc=1 rebuild allowlist is an escape hatch from this same fix.**
`bootstrap.rs:1935-1941` exempts `Hole` and `EmptyNonTailSegment` from
tombstoning at `replica_count = 1`, on the rationale that "their segment bytes
sit intact in quarantine and no damage verdict needs surfacing". Neither
verdict establishes that. `Hole` fires on `next.start_offset !=
previous.end_offset + 1` with both segments holding data; `EmptyNonTailSegment`
fires on `previous.size == 0` with data-bearing segments after it. Both then
quarantine everything and rebuild empty, so polls succeed with zero messages
over bytes that are all present — verbatim the failure
`bootstrap.rs:1925-1929`'s own comment gives as the reason the tombstone exists.
`git log -S"rebuild_for_rejoin"` and `-S"EmptyNonTailSegment { .. }"` on
`master..HEAD` both return only `2a01e2df3`, and `master`'s `bootstrap.rs` has
zero occurrences — so the allowlist is this PR's own code, not pre-existing
behaviour.
There is also a route from the guard added at `shard/src/lib.rs:2179` into
that allowlist: the discard drops a build that already truncated segment 0 (its
own comment says so), the next boot walks a 0-byte non-tail segment, that
raises `EmptyNonTailSegment`, and the allowlist rebuilds it empty.
**Suggested fix, and please land them in this order:**
1. **First**, give the fence an exit: have `reconcile_removals` also collect
namespaces that are tombstoned and absent from both `partitions` and the
committed target, and route them through `tear_down_owned_partition`, which
already does fence-writes → `delete_partitions_from_disk` → `ConfirmRemove` →
`untombstone` and already refuses a cross-shard delete. The delete only fires
when metadata says the partition is gone, so it destroys nothing an operator
did not delete. A bare `untombstone` is not enough on its own — it leaves the
cause in place, so the next boot re-derives the refusal, and in the window
before that the reconciler builds fresh over the refused files, which is the
exact truncation this gate was added to stop.
2. **Then** narrow the allowlist: at `replica_count = 1`, tombstone both
verdicts unless every planned segment is size 0. The evidence-based form — gate
on the planned chain's recoverable bytes rather than on the refusal's variant —
needs the byte total carried on the refusal, which is the same field
`partition_helpers.rs:657-663` already says is needed, so it is a step the code
plans anyway. Ship the simple form if that field is out of scope.
Not the other order. Narrowing the allowlist first sends more namespaces
into a fence that has no exit, so the tombstone population grows before the
door exists.
Separately, and as a follow-up rather than part of this: a tombstoned
namespace answers `IggyError::TransientNotAccepted` (`shard/src/lib.rs:2571`,
via `ParkOutcome::Tombstoned`). Clients do not hang — the deny path exists
precisely to avoid the plane's silent drop — but a terminal, operator-only
state is reported as **retriable**, so every SDK burns its full retry budget
per call, forever. `zero_out_all` also makes the partition read as a healthy
empty one through `GetTopic` and `/stats`. That wants a terminal discriminant
and a counter, which means six SDK mirrors, so it does not belong in this PR.
##########
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:
**Blocker: this constant is documented as an invariant, and it is not
enforced where batches are actually admitted.**
The doc says the ceiling caps "the widest batch record any admission path
can persist", and `MAX_RECOVERABLE_BATCH_BYTES` in `segment_recovery.rs` is
derived from it and used to reject headers. But the new validator only checks
`message_bus.max_message_size`, and the HTTP produce path never goes through
the frame decoder:
- `partition_write_replicated` builds the request in-process via
`build_request_message` (`http/wire.rs:199`, called from `http/submit.rs:383`),
so `framing::read_message`'s `total_size <= max_message_size` check
(`message_bus/src/framing.rs:107-122`) never runs. `framing::write_message` has
no size check either — only the read side caps — so at `replica_count > 1` the
primary journals the oversize batch locally and the *peer's* read rejects the
frame, with the bytes already on disk.
- There is no batch-total cap anywhere on produce: `SendMessages::validate`
→ `IggyMessagesBatch::validate` (`messages_batch.rs:219-226`) checks
`MAX_PAYLOAD_SIZE` **per message**, never the sum, and
`SendMessagesOwned::from_messages` (`send_messages.rs:157`) computes
`batch_length` with no ceiling.
- `http.max_request_size` has no validator in either validators file and is
`#[config_env(leaf)]`, so it is env-settable.
The code already says this outright at `shard/src/lib.rs:963-968`: *"Derived
from the BUS frame cap, not `MAX_PAYLOAD_SIZE`: the server never enforces the
latter (its only enforcement sites are the legacy server and the SDK batch
types), so the largest appendable batch is whatever the message bus will
frame."* The HTTP path is not framed by the bus, so on that path nothing frames
it.
Reachability is precise and operator-gated, not accidental: JSON bodies
carry base64, so raw payload is ~3/4 of the body, and you need
`http.max_request_size` above roughly 342 MiB with five or more messages of ≤64
MB each. Shipped default is 2 MB, so there are three orders of magnitude of
headroom. But once past it, `peek_header` returns `None` on a legally admitted,
checksum-valid batch: the walk breaks and the tail is silently truncated, or
`InteriorDamage` refuses and at `replica_count = 1` tombstones permanently.
**Suggested fix:** validate `http.max_request_size <=
MAX_MESSAGE_SIZE_UPPER_BYTES` at boot. Sound because a produce request carries
at most one batch and base64 leaves ~25% slack, so bounding the body bounds the
record; `http/forward.rs:290-297` re-reads the same key so forwards inherit it.
Optionally add a batch-total check beside the per-message one in
`IggyMessagesBatch::validate` — if so, please **reuse
`IggyError::TooBigMessagePayload`** rather than minting a discriminant, since
codes are mirrored in the Go and Node tables and a new one for a config-only
failure is not worth the regeneration. Note that validator lives in published
`iggy_common` and the Rust SDK calls it client-side, so it is a behaviour
change in a published crate, and that `MAX_PAYLOAD_SIZE = 64 * 1000 * 1000` is
*decimal* while everything here is binary MiB.
Either enforce it or drop the "widest batch record any admission path can
persist" claim from this doc and from `segment_recovery.rs:78-86`. Shipping the
claim without the enforcement is the part that will mislead the next reader.
**Two more things this ceiling needs documented.** A deployment currently
running above 256 MiB has no non-destructive upgrade path: it cannot boot until
the knob is lowered, and once booted recovery treats the wide batches already
on disk as implausible and truncates or refuses the segment holding them.
Nothing tells the operator to drain and re-produce below the ceiling first. And
the validator ordering hides the ceiling error — `validators.rs:164-177` runs
before `message_bus.validate()` at `:205`, so `max_message_size = "512 MiB"`
hits the artifact-floor error first and sends the operator to raise
`transfer_artifact_bytes_max`; only after doing that do they meet the real
ceiling, i.e. two boot cycles to learn the edit is impossible.
##########
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:
**Blocker: at the cap this is a fixed point, so release builds spin forever
with no signal — and the `debug_assert!` covers the one build where the failure
cannot happen.**
`(YIELD_ATTEMPT_CAP * 2).min(YIELD_ATTEMPT_CAP) == YIELD_ATTEMPT_CAP`, so
once `attempt` reaches 1 s the loop keeps allocating a `Box::pin` per
iteration, polling, and retrying — no log, no timeout, on the boot path. Debug
panics on the assert instead. Neither is a usable terminal shape: an
unyieldable reactor should cost throughput, not the boot.
The assert also names the one condition that cannot trigger it. A frozen
clock makes `now + attempt > now`, so `TimerRuntime::insert` *succeeds* and the
yield hangs — no assert. Reaching the cap requires the clock to **advance** by
≥ `attempt` between `sleep()`'s `Instant::now()` and `insert()`'s re-read, 32
consecutive times with doubling stalls.
**Suggested fix — bound the retry and drop the hand-written `Future`, in one
edit.** `futures` and `tracing` are already `server_common` dependencies:
```rust
const YIELD_MAX_ATTEMPTS: u32 = 32;
pub async fn yield_to_reactor() {
let mut attempt = YIELD_FIRST_ATTEMPT;
for _ in 0..YIELD_MAX_ATTEMPTS {
let mut timer = std::pin::pin!(compio::time::sleep(attempt));
if futures::poll!(timer.as_mut()).is_pending() {
timer.await;
return;
}
attempt = (attempt * 2).min(YIELD_ATTEMPT_CAP);
}
tracing::error!(
"no timer registration won in {YIELD_MAX_ATTEMPTS} attempts; \
continuing without yielding to the reactor"
);
}
```
Identical semantics, one allocation fewer per attempt, ~60 lines shorter,
and the terminal case degrades to the pre-PR no-yield behaviour and says so.
With the bound there is nothing left to assert about the clock.
**The design is right and the mechanism holds — but the module doc's
justification is wrong by ~500× and in the wrong direction, and so is the
commit message.** `:26-27` says "a debug-build cold path loses a 1 us head
start essentially always". Measured on this branch: a bare
`sleep(from_micros(1))` registers 19999/20000 warm release, 19998/20000 warm
debug, 3000/3000 cold release, and **2994/3000 cold debug** — so it *wins* that
race 99.8% of the time. `0929f697b`'s body cites the same figure. The retry is
still worth keeping: at ~512 refills per GiB of walk, a 2e-3 failure rate
silently skips a yield per GiB on a debug build, and the skip is invisible
because the pass still completes. But please correct the number rather than
shipping a rationale that inverts its own measurement. `YIELD_FIRST_ATTEMPT`'s
own doc ("on a warm path one microsecond outlives the registration window and
the first attempt wins") is correct on the new figures and should stay.
Also worth correcting at `:49-51`: "the attempted duration does not throttle
the caller". Measured µs/yield against the first attempt: 200 ns → 1.7, 1 µs →
11.5-12.6, 10 µs → 11-33, 50 µs → 49-65, 100 µs → 98-104, 500 µs → 485-498. It
throttles roughly linearly above 10 µs. Harmless at 1 µs, but someone will
raise the constant on the strength of that sentence.
For the record on the code itself, since it is easy to doubt: the first poll
returns `Pending` across 400,000+ calls in debug and release, warm and cold,
with zero exceptions, where `sleep(Duration::ZERO)` was `Ready` 20000/20000.
Failed attempts leak nothing — `insert` returns `None` before any wheel entry
exists — and `poll_timer` stores the caller's real waker, so the stored timer's
wake reaches the task. The fix works; only its stated reason does not.
##########
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:
**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.
And `:468` "the index is rebuilt when it was damaged" contradicts `:469` in
the same paragraph and the code: a rebuild happens only in the arm reached when
the index holds no whole entry (`segment_recovery.rs:1040`); a
damaged-but-nonempty index is floored to whole entries or refuses the
partition. Repair versus tombstone is the widest gap between doc and code here,
and none of the three index refusals is on the rc=1 rebuild allowlist.
**Suggested replacement**, written assuming the absorption is reverted and
both variants come off the rc=1 allowlist. If either lands differently, the two
swap paragraphs below apply.
```
# 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 (no whole entry left) is rebuilt from
# the batches the walk proves; 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 batches; with an intact index the walk trusts batch
# headers. Either walk refuses a batch whose base offset does not continue
the
# chain. 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 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 A**, if the forward-gap absorption is kept — replace "Either walk
refuses a batch whose base offset does not continue the chain." with:
```
# Either walk refuses a batch whose base offset REGRESSES. A forward gap in a
# byte-clean chain is absorbed instead, and the segment's end offset then
# covers offsets no batch in it holds: reported message counts for that topic
# are inflated by the gap, and a peer can never install the segment (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.
```
**Swap B**, if the rc=1 allowlist stays — append to the second paragraph:
```
# Two directory-shape refusals are the exception and still rebuild empty at
# replica_count = 1: a hole in the segment sequence, and an empty non-tail
# segment. Neither proves the fenced files held nothing, so a single-replica
# partition CAN return empty through them. The boot log is the only authority
# on which path a partition took.
```
Separately, the 256 MiB ceiling should be stated inline at
`max_message_size` (`:962`) the way every other bounded knob in this file
states its bound (`:910`, `:929`, `:946`, `:958`), along with the fact that the
HTTP produce path is capped by `http.max_request_size` rather than by that
knob, and that four of six SDKs cap response frames at 64 MiB — the Go SDK at
`foreign/go/internal/vsr/header.go:30` is a hard const and cannot be
configured, so anything above 64 MiB is unreachable from Go. The advice at
`:942` ("raise this one by the same amount in the same edit") is also now wrong
above 256 MiB, where no `transfer_artifact_bytes_max` value boots.
--
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]
