hubcio commented on code in PR #3946:
URL: https://github.com/apache/iggy/pull/3946#discussion_r3831781897
##########
core/server/src/segment_recovery.rs:
##########
@@ -320,44 +445,196 @@ fn
sweep_scratch_files_and_collect_offsets(partition_path: &str) -> Result<Vec<u
Ok(start_offsets)
}
-fn file_len(path: &str) -> u64 {
- fs::metadata(path).map_or(0, |metadata| metadata.len())
+/// Byte length of a segment file, a missing file reading as empty.
+///
+/// Any other stat failure is fail-stop, mirroring the `NotFound`-only leniency
+/// of the directory listing above: recovery physically truncates files to the
+/// bounds derived from these lengths, so folding a transient `EACCES` or
+/// `EIO` into 0 would route a healthy segment into recover-as-empty and
+/// truncate it to nothing (worst route: an index stat error floors a healthy
+/// sealed index to a 0-byte target while its entries still load).
+fn file_len(path: &str) -> Result<u64, ServerError> {
+ match fs::metadata(path) {
+ Ok(metadata) => Ok(metadata.len()),
+ Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(0),
+ Err(source) => {
+ error!(
+ path,
+ error = %source,
+ "failed to stat a segment file during recovery"
+ );
+ Err(IggyError::CannotReadFileMetadata.into())
+ }
+ }
+}
+
+/// Physically truncates a segment file to its recovered byte length, so disk
+/// and the seeded size counters agree before storage reopens: reopen verifies
+/// the on-disk length against the recovered size and refuses a divergence,
+/// and before that check existed a leftover tail silently resurrected through
+/// the writers' re-stat of the raw length. Truncation also protects state
+/// transfer: the sender sizes each artifact from `segment.size` and hashes
+/// exactly `[0, segment.size)`, so resurrected garbage INSIDE that range
+/// would poison every artifact a torn replica offers once it serves as
+/// primary.
+///
+/// The tail being discarded was proven dead by the bounds walk: nothing past
+/// the recovered size decodes (the interior-damage probe refuses recovery
+/// outright when something does), so polls could never serve those bytes.
+///
+/// Stats the file fresh instead of trusting a length carried from pass A: the
+/// whole chain was walked in between, and the mutation must key on what is on
+/// disk now. Synchronous `std::fs` on purpose (see [`FileScanner`]). The
+/// fsync bounds the crash window: a power cut right after `set_len` may
+/// re-present the torn tail on the next boot, which only walks and truncates
+/// again (idempotent), but the sync keeps the common case deterministic.
+fn truncate_to(path: &str, target_size: u64) -> Result<(), ServerError> {
+ let current_size = file_len(path)?;
+ if current_size == target_size {
+ return Ok(());
+ }
+ // Unreachable by construction (walked bounds never exceed the file they
+ // were walked from); extending would fabricate a zero-filled tail, and
+ // zero bytes decode as valid-looking index entries -- three bare
+ // little-endian u64s with no magic to reject them -- so fail stop.
+ if target_size > current_size {
+ error!(
+ path,
+ current_size,
+ target_size,
+ "recovered bounds exceed the file they were walked from; \
+ refusing to extend a segment file"
+ );
+ return Err(IggyError::CannotWriteToFile.into());
+ }
+ warn!(
+ path,
+ current_size,
+ target_size,
+ "truncating a segment file to its recovered bounds; discarding \
+ torn tail bytes"
+ );
+ let file = fs::OpenOptions::new()
+ .write(true)
+ .open(path)
+ .map_err(|source| {
+ error!(
+ path,
+ error = %source,
+ "failed to open a segment file for truncation during recovery"
+ );
+ ServerError::from(IggyError::CannotWriteToFile)
+ })?;
+ file.set_len(target_size).map_err(|source| {
+ error!(
+ path,
+ target_size,
+ error = %source,
+ "failed to truncate a segment file to its recovered bounds"
+ );
+ ServerError::from(IggyError::CannotWriteToFile)
+ })?;
+ file.sync_all().map_err(|source| {
+ error!(
+ path,
+ error = %source,
+ "failed to fsync a segment file after truncation"
+ );
+ ServerError::from(IggyError::CannotSyncFile)
+ })?;
+ Ok(())
+}
+
+/// Persists the index rebuilt by the index-less walk, replacing whatever
+/// partial or stale bytes the crash left. Without this a SEALED segment --
+/// which never flushes again -- would keep an empty index forever and pay a
+/// full log scan on every poll.
+///
+/// Written straight to the final path: a crash mid-write leaves a shorter
+/// index whose whole entries are a valid prefix of this same rebuild, and the
+/// next boot walks and rewrites it again -- recovery is itself the repair
+/// path for a torn index, so no rename dance is needed.
+fn write_rebuilt_index(path: &str, entries: &[u8]) -> Result<(), ServerError> {
Review Comment:
fixed in 2a01e2df3 - the rebuild is staged during pass A ({index}.staging,
fsynced, entries dropped from memory) and renamed + dir-fsynced in pass C. the
boot sweep already unlinks orphaned staging files. the dead create(true) went
away with the rewrite.
##########
core/server/src/segment_recovery.rs:
##########
@@ -147,46 +235,91 @@ pub async fn load_persisted_segments(
stream_id,
topic_id,
partition_id,
- path = %messages_path,
+ path = %plan.messages_path,
error = %source,
"failed to open persisted segment storage during recovery"
);
- source
+ // The seed-vs-stat guard refusing the open means disk diverged
+ // from the size this pass just truncated to: structural, and the
+ // heal path for data directories an earlier size-counter bug left
+ // with resurrected tails. Everything else here is transient I/O
+ // and stays node-fatal.
+ match source {
+ IggyError::SegmentSizeMismatchAtOpen(on_disk_bytes,
expected_bytes) => identity
+ .refusal(PartitionRecoveryRefusal::StorageSizeMismatch {
+ start_offset: plan.segment.start_offset,
+ on_disk_bytes,
+ expected_bytes,
+ }),
+ transient => transient.into(),
+ }
})?;
- let mut segment = Segment::new(start_offset, max_size);
- segment.sealed = true;
- segment.start_timestamp = start_timestamp;
- segment.end_timestamp = end_timestamp;
- segment.max_timestamp = end_timestamp;
- segment.end_offset = end_offset;
- segment.size = IggyByteSize::from(effective_messages_size);
- segment.current_position = effective_messages_size;
-
stats.increment_segments_count(1);
- stats.increment_size_bytes(effective_messages_size);
- if effective_messages_size > 0 {
+ stats.increment_size_bytes(messages_size);
+ if messages_size > 0 {
// Offsets in a segment are contiguous, so the message count is the
// inclusive span between the first (segment start) and last
offset.
- stats.increment_messages_count(end_offset - start_offset + 1);
+ stats.increment_messages_count(plan.segment.end_offset -
plan.segment.start_offset + 1);
Review Comment:
fixed in 2a01e2df3 - the indexed walk now refuses OffsetDiscontinuity on any
header whose base_offset doesn't continue from the anchor, mirroring the
index-less arm, so the subtraction can't underflow and a lowered end_offset
can't regress the partition offset anymore. your repro (segment 100, batch base
5) is now a unit test asserting refusal instead of panic.
##########
core/server/src/segment_recovery.rs:
##########
@@ -438,106 +739,1026 @@ 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;
}
+ refuse_if_survivor_past_damage(
+ identity,
+ &mut scanner,
+ messages_path,
+ position,
+ messages_size,
+ start_timestamp.map(|_| end_offset),
+ start_offset,
+ )?;
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(())
+}
+
+/// 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, truncating the whole log to zero.
+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)
}
-fn read_batch_header(
- messages: &fs::File,
- position: u64,
+/// 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.
+/// Unlike the WAL there is NO width cap on the damage: a segment flush chunk
+/// is unbounded, so any amount of trailing garbage can still be one torn
+/// write.
+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 survivor = scanner
+ .probe_for_survivor(damage_position, chain_end_offset, start_offset)
+ .map_err(|source| scan_read_failure(identity, messages_path,
&source))?;
+ if let Some(survivor_position) = survivor {
+ return Err(identity.refusal(PartitionRecoveryRefusal::InteriorDamage {
+ start_offset,
+ damage_position,
+ survivor_position,
+ }));
+ }
+ Ok(())
+}
+
+/// 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>,
+}
+
+impl<'scan> FileScanner<'scan> {
+ fn new(file: &'scan fs::File, file_len: u64, scratch: &'scan mut
ScanScratch) -> Self {
+ let ScanScratch { window, spill } = scratch;
+ window.clear();
+ Self {
+ file,
+ file_len,
+ window,
+ window_start: 0,
+ spill,
+ }
+ }
+
+ /// Bytes `[position, position + len)`, or `None` when they run past the
+ /// end of the file.
+ fn slice_at(&mut self, position: u64, len: usize) ->
io::Result<Option<&[u8]>> {
+ let Some(end) = position.checked_add(len as u64) else {
+ return Ok(None);
+ };
+ if end > self.file_len {
+ return Ok(None);
+ }
+ if len > SCAN_WINDOW_CAPACITY {
+ // A batch larger than the window: one direct read, no windowing.
+ self.spill.resize(len, 0);
+ self.file.read_exact_at(&mut self.spill[..], position)?;
+ 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;
+ }
+ // In-window by the branch above, and the window is capacity-bounded,
+ // so the try_from cannot fail.
+ let start = usize::try_from(position - self.window_start).unwrap_or(0);
+ Ok(Some(&self.window[start..start + len]))
+ }
+
+ /// The batch command header at `position`, or `None` when it does not fit
+ /// the file or does not decode (torn header, garbage bytes).
+ fn peek_header(&mut self, position: u64) ->
io::Result<Option<BatchHeader>> {
+ let Some(bytes) = self.slice_at(position, COMMAND_HEADER_SIZE)? else {
+ return Ok(None);
+ };
+ Ok(BatchHeader::decode(bytes).ok())
+ }
+
+ /// Position of the first complete, checksum-verifying batch starting
+ /// after `damage_position`, or `None` when the residue holds none.
+ ///
+ /// 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 offset is a candidate. The header decode pre-filters candidates
+ /// cheaply -- 204 reserved bytes must be zero -- and offset sanity plus
+ /// length bounds run before a checksum is paid, so the full verify only
+ /// runs on byte positions that already look like a plausible chain
+ /// continuation.
+ fn probe_for_survivor(
Review Comment:
fixed in 2a01e2df3 - residue wider than max_message_size refuses before any
scan, the probe carries a byte budget at 2x the cap counting bytes read plus
bytes handed to verification, and exhausting either refuses via a new
UnverifiedResidue reason - it never falls through to the truncating no-survivor
verdict. the window now scans candidates in place and refills strictly forward
instead of re-anchoring per candidate, and refills yield via compio sleep(0).
your zero-padded record shape is a unit test asserting refusal with files
byte-identical.
##########
core/server/src/segment_recovery.rs:
##########
@@ -438,106 +739,1026 @@ 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;
}
+ refuse_if_survivor_past_damage(
+ identity,
+ &mut scanner,
+ messages_path,
+ position,
+ messages_size,
+ start_timestamp.map(|_| end_offset),
+ start_offset,
+ )?;
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(
Review Comment:
fixed in 1cc830ab9 - the index writer now fsyncs before advancing the
cursor, matching the messages writer, so the retry overwrites the same slot
instead of appending a duplicate. on the boot-time scan: it streams through the
shared 4 MiB window rather than loading the index resident, so the cost is read
bandwidth only - left ungated.
##########
core/server/src/segment_recovery.rs:
##########
@@ -92,53 +134,99 @@ pub async fn load_persisted_segments(
.system
.get_index_path(stream_id, topic_id, partition_id,
start_offset);
- let messages_size = file_len(&messages_path);
- let index_size = file_len(&index_path);
+ let raw_messages_size = file_len(&messages_path)?;
let bounds = recover_segment_bounds(
+ identity,
&index_path,
&messages_path,
start_offset,
- messages_size,
- stream_id,
- topic_id,
- partition_id,
+ raw_messages_size,
+ &mut scratch,
)
.await?;
- // `bounds == None` now means the log holds no whole BATCH either (the
- // index-less path above already tried walking the log), so there is
- // nothing to recover: zeroed sizes make the next append overwrite the
- // torn bytes, where counting them with `end_offset == start_offset`
- // would fabricate one phantom message for the bootstrap non-empty
- // filters and strand undecodable garbage inside the readable range.
- // Note this is NOT tail-only -- a torn index is reachable mid-chain on
- // the shipped `enforce_fsync = false`, which is why the walk above
- // exists rather than refusing the partition.
- let (start_timestamp, end_timestamp, end_offset,
effective_messages_size) =
- if let Some((start_timestamp, end_timestamp, end_offset,
walked_size)) = bounds {
- (start_timestamp, end_timestamp, end_offset, walked_size)
- } else {
- if messages_size > 0 {
- warn!(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- messages_size,
- "segment log holds bytes but its index holds no whole \
- entry (torn write); recovering the segment as empty"
- );
- }
- (0, 0, start_offset, 0)
- };
- let effective_index_size = if bounds.is_some() { index_size } else { 0
};
+ // `bounds == None` means the log holds no whole batch ANYWHERE: the
+ // index-less walk tried from byte 0 and the damage probe found no
+ // surviving batch deeper in the file. There is nothing to recover:
+ // zeroed sizes make the next append overwrite the torn bytes, where
+ // counting them with `end_offset == start_offset` would fabricate one
+ // phantom message for the bootstrap non-empty filters and strand
+ // undecodable garbage inside the readable range. Note this is NOT
+ // tail-only -- a torn index is reachable mid-chain on the shipped
+ // `enforce_fsync = false`, which is why the walk exists rather than
+ // refusing the partition.
+ let bounds = bounds.unwrap_or_else(|| {
Review Comment:
fixed in 2a01e2df3 - the no-recoverable-bytes verdict now renames both files
into a fresh {partition_dir}.fenced.N (keeping their stems) and seeds empty
files in their place instead of set_len(0). the warn names both fenced paths
and byte counts. test updated to assert the original bytes survive byte-exact.
##########
core/common/src/error/iggy_error.rs:
##########
@@ -425,6 +430,11 @@ pub enum IggyError {
InvalidOptionValue(String) = 4042,
#[error("Options block exceeds its limits: {0}")]
OptionsBlockTooLarge(String) = 4043,
+ /// The on-disk segment file length disagrees with the recovered bounds the
+ /// writer was seeded with; appending would corrupt the segment, so the
+ /// open fails instead. Field order: `(on_disk, expected)`.
+ #[error("Segment file size on disk: {0} does not match expected size:
{1}")]
+ SegmentSizeMismatchAtOpen(u64, u64) = 4044,
Review Comment:
taken in 18c107bcf - moved to 4102, go and node tables regenerated, guard
kept as a post-condition assertion.
##########
core/server/src/server_error.rs:
##########
@@ -281,14 +273,16 @@ pub enum ServerError {
ShardJoinFailures { failures: Vec<ShardJoinFailure> },
}
-/// Why a recovered segment chain cannot be served.
+/// Why a partition's recovered segments cannot be served.
///
-/// Both shapes mean the same thing operationally -- the local files do not
form
-/// a chain this replica can serve -- but they are distinguished because they
-/// point at different causes: an empty non-tail segment is a failed rebuild's
-/// orphan pairing, a hole is a stray or half-unlinked file.
+/// Every shape here is structural -- the local files contradict themselves or
Review Comment:
reworded in 2a01e2df3 - the variant doc now says offsets-not-contiguous and
names the byte-clean frontier crash window as a possible cause; the refusal
stays. filed the upstream seal-and-rotate as a follow-up. one nuance: the
frontier is the commit frontier and the superblock isn't rewritten per commit,
so the shape needs a view change or install landing inside the commit-to-flush
window - reachable, but not the routine crash path.
##########
core/server/src/server_error.rs:
##########
@@ -164,19 +164,23 @@ pub enum ServerError {
},
// 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.
+ // 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}. The boot path quarantines this partition's
segment \
+ files beside its directory and rebuilds it empty for the rejoin path;
\
+ restore from a healthy replica, or repair the quarantined files
offline."
Review Comment:
done in 2a01e2df3 - the repair-offline clause is gone, the message names the
.fenced.N scheme, and the outcome wording is conditional now (quarantine
failure tombstones without rebuild; replica_count 1 has no rejoin).
##########
core/server/src/segment_recovery.rs:
##########
@@ -52,14 +74,23 @@ pub struct RecoveredSegment {
/// Loads every persisted segment for a partition, sorted by start offset.
///
/// Segment offsets and timestamps are recovered from the 24-byte sparse index
-/// (see module docs); segment byte size comes from the `.log` file. The last
-/// segment is left unsealed so it can accept further writes.
+/// (see module docs); segment byte size comes from walking the `.log` batch
+/// chain. Recovery runs in three passes: every segment is bounded READ-ONLY
+/// first, then the chain guard runs over those bounds, and only an accepted
+/// chain is made physical -- torn tails truncated, index-less indexes rebuilt
+/// -- before storage opens over it. A refusal at any point therefore leaves
+/// every file byte-identical to what boot found. The last segment is left
Review Comment:
scoped in 2a01e2df3 - the byte-identity claim now names passes A and B only
and states pass C is not atomic across segments.
##########
core/partitions/src/messages_writer.rs:
##########
@@ -256,4 +271,40 @@ mod tests {
assert_eq!(writer.file.metadata().await.unwrap().len(), 0);
}
+
+ #[compio::test]
+ async fn
given_seeded_size_matching_disk_when_opening_existing_file_should_keep_counter()
{
Review Comment:
deleted in 1cc830ab9 - kept the divergence twins.
##########
core/server/src/segment_recovery.rs:
##########
@@ -438,106 +739,1026 @@ 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;
}
+ refuse_if_survivor_past_damage(
+ identity,
+ &mut scanner,
+ messages_path,
+ position,
+ messages_size,
+ start_timestamp.map(|_| end_offset),
+ start_offset,
+ )?;
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(())
+}
+
+/// 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, truncating the whole log to zero.
+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)
}
-fn read_batch_header(
- messages: &fs::File,
- position: u64,
+/// 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.
+/// Unlike the WAL there is NO width cap on the damage: a segment flush chunk
Review Comment:
you're right - the record is the thing to bound, not the flush chunk.
replaced the comment with that justification in 2a01e2df3; since exhaustion
refuses, an under-tight cap errs toward preserving bytes.
##########
core/common/src/error/iggy_error.rs:
##########
@@ -22,11 +22,16 @@ use std::sync::Arc;
use strum::{EnumDiscriminants, FromRepr, IntoStaticStr};
use thiserror::Error;
-// A gap in the discriminants is a RETIRED code, not free space. Shipped SDKs
-// keep their own code tables (foreign/go/errors/errors.yaml,
-// foreign/node/src/wire/error.code.ts) that still map the old meaning, and
-// Go's is a typed error matched by errors.Is, so refilling a gap reroutes
-// caller control flow. Allocate above the highest code in its range.
+// Codes are allocated per semantic family: a new code goes one above its
Review Comment:
restored the previous wording verbatim in 18c107bcf - agreed the family rule
didn't survive contact with the file it governs.
##########
core/server/src/bootstrap.rs:
##########
@@ -1911,13 +1914,16 @@ async fn build_shard_for_thread(
{
Ok(partition) => partition,
// ONE damaged local chain must not take the node down. The shapes
- // this refuses are exactly what a failed state-transfer quarantine
- // leaves behind, so fence that group the same way the runtime path
- // does -- move its segment files aside, keeping the superblock so
it
- // cannot re-enter view 0 -- and materialise it fresh. The ordinary
- // rejoin path (repair, then state transfer on a refused floor)
- // recovers its data from a peer.
- Err(ServerError::PartitionChainRefused { dir, reason, .. }) => {
+ // this refuses are structural -- what a failed state-transfer
+ // quarantine leaves behind, or damage the recovery walk proved
+ // inside a segment -- so fence that group the same way the runtime
+ // path does -- move its segment files aside, keeping the
superblock
+ // so it cannot re-enter view 0 -- and materialise it fresh. The
+ // ordinary rejoin path (repair, then state transfer on a refused
+ // floor) recovers its data from a peer; a single-replica group has
+ // no peer, so it comes back EMPTY while every refused byte stays
+ // in the quarantine directory for the operator.
+ Err(ServerError::PartitionRecoveryRefused { dir, reason, .. }) => {
Review Comment:
fixed in 2a01e2df3 - at replica_count 1 the arm now quarantines then
tombstones instead of rebuilding empty. went slightly wider than
IndexLogDivergence: the four reasons this PR introduced have the same
silent-empty problem, so they tombstone too; Hole and EmptyNonTailSegment keep
the pre-existing fence+rebuild. one correction while in there: the superblock
arm tombstones unconditionally, not only at replica_count 1, so this now
matches its shape.
##########
core/server/src/bootstrap.rs:
##########
@@ -2763,6 +2788,40 @@ async fn hydrate_partition_log(
Ok(())
}
+/// Routes a hydrate-reopen writer failure. The seed-vs-stat divergence guard
+/// (`SegmentSizeMismatchAtOpen`) is the same structural contradiction the
+/// recovery walk refuses on -- and the heal path for data directories an
Review Comment:
kept as defense-in-depth and said so plainly in 2a01e2df3 - both rationale
comments now call it a post-condition assertion on pass C's own truncation,
firing only if the filesystem lies or a future change breaks the
truncate-then-open contract. agreed truncate_to is the actual heal.
##########
core/server/config.toml:
##########
@@ -463,6 +463,14 @@ 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
Review Comment:
corrected in 2a01e2df3 - 'last decodable batch', a note that bytes below the
last index entry are the poll path's checksum job, the replica_count 1 outcome
(tombstoned, files kept in .fenced.N), and the WAL sentence now says torn tails
truncate there too, only interior damage refuses.
--
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]
