numinnex commented on code in PR #3970:
URL: https://github.com/apache/iggy/pull/3970#discussion_r3882778616
##########
core/server/src/segment_recovery.rs:
##########
@@ -1001,154 +1035,162 @@ 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)?;
+ // A mis-strided or foreign index decodes to garbage entries that
+ // binary searches would trust, and the walk's anchor bounds only
+ // the tail. Ascending positions keep every surviving entry inside
+ // the truncated log once the tail is floored, so an index that
+ // contradicts itself is dropped whole and rebuilt from the log,
+ // exactly like one the log backs nowhere: entries are derived
+ // from the log, so the index can never say more than the log.
+ let consistent =
+ index_is_consistent(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);
+ if !consistent {
+ return recover_by_walking_log(
+ identity,
+ &mut scanner,
+ messages_path,
+ start_offset,
+ messages_size,
+ )
+ .await;
+ }
// 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.
- 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;
- // TODO(hubcio): for batches that continue the chain exactly this
- // indexed walk trusts the header decode alone (batches that
- // contradict the walk checksum-verify below), 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;
- }
- // A header contradicting the walk -- a foreign partition_id,
- // or a base offset that does not continue the chain in
- // EITHER direction -- is either damage wearing a decodable
- // header or a real record that does not belong at this
- // position. The batch checksum tells them apart, because it
- // covers both fields and the server mints every legal record
- // with it: a batch that fails it is damage, so break and let
- // the probe below classify the residue (a bit flip in a tail
- // batch then truncates like any torn tail, regardless of the
- // flip's direction), while a batch that VERIFIES is durable
- // evidence -- of a misdirected or copied write when the
- // partition stamp is foreign, of duplicated or lost offsets
- // when the chain breaks -- and both absorbing it and
- // truncating it would hide that, so refuse and keep the
- // bytes. Refusing the verified forward gap over absorbing it
- // is deliberate: state transfer's own walk refuses any gap,
- // so an absorbed one would mint a segment no peer can ever
- // install, and the offsets it fabricates seed current_offset
- // under an advance-only superblock persist.
- let foreign_partition = header.partition_id !=
identity.partition_id as u64;
- if foreign_partition || header.base_offset != expected_offset {
- 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 foreign_partition {
- return
Err(identity.refusal(PartitionRecoveryRefusal::ForeignBatch {
- start_offset,
- batch_partition_id: header.partition_id,
- position,
- }));
- }
- return Err(
-
identity.refusal(PartitionRecoveryRefusal::OffsetDiscontinuity {
- start_offset,
- expected_offset,
- found_offset: header.base_offset,
- position,
- }),
+ let mut walk = walk_chain_from_anchor(
+ identity,
+ &mut scanner,
+ messages_path,
+ start_offset,
+ messages_size,
+ last,
+ )
+ .await?;
+ let mut retained_entries = entry_count;
+ if !walk.proved_any {
+ // An index entry is not independent evidence about the log:
+ // the writer derives every entry from a batch the log
+ // already held, so an entry the log cannot back is a torn
+ // tail of the INDEX, never proof that the log lost data.
+ // That shape is ordinary: the two files persist concurrently
+ // under every config, so they have no write ordering between
+ // them, and pass C truncates the log before the index, so a
+ // crash between those two mutations leaves exactly this.
+ // Refusing it handed a benign torn tail a permanent tombstone
+ // on a single-replica node. Step down
+ // to the highest entry the log still proves and floor the
+ // index there, so pass C drops the entries nothing backs.
+ let anchor = find_provable_index_anchor(
Review Comment:
The step-back can anchor **above** interior damage, and nothing ever walks
`[0, anchor.position)`. `refuse_if_survivor_past_damage` only scans forward
from `walk.position` and returns `Ok` immediately when `damage_position >=
messages_size` (:1757), so damage below the anchor is invisible to it.
Reachable without hand-crafting an index. One entry per flushed chunk is
what the writer emits; at-rest bit rot inside chunk 3, plus the same crash
tearing the tail so the last entry cannot prove. The step-back lands on chunk
4, which verifies, and the walk runs from there to EOF and returns `Ok`. The
rot at chunk 3 is never read. Result: a checksum-failing batch inside the
advertised offset range, `stats.increment_messages_count` at :339 counting
offsets that cannot be served, and `current_offset` written into the
advance-only superblock. master refused these bytes.
Suggest the floored path fall through to `recover_by_walking_log` (walk from
byte 0) instead of a second anchored walk. That reaches the probe with
`damage_position < messages_size`, finds the survivor, and refuses with
`InteriorDamage` with the bytes preserved.
Related: the justification at :1080-1085 ("an entry the log cannot back is a
torn tail of the INDEX, never proof that the log lost data") is false in
exactly this case, since the unprovable tail entry *is* evidence the log lost
an interior page.
##########
core/server/src/segment_recovery.rs:
##########
@@ -1001,154 +1035,162 @@ 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)?;
+ // A mis-strided or foreign index decodes to garbage entries that
+ // binary searches would trust, and the walk's anchor bounds only
+ // the tail. Ascending positions keep every surviving entry inside
+ // the truncated log once the tail is floored, so an index that
+ // contradicts itself is dropped whole and rebuilt from the log,
+ // exactly like one the log backs nowhere: entries are derived
+ // from the log, so the index can never say more than the log.
+ let consistent =
+ index_is_consistent(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);
+ if !consistent {
+ return recover_by_walking_log(
+ identity,
+ &mut scanner,
+ messages_path,
+ start_offset,
+ messages_size,
+ )
+ .await;
+ }
// 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.
- 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;
- // TODO(hubcio): for batches that continue the chain exactly this
- // indexed walk trusts the header decode alone (batches that
- // contradict the walk checksum-verify below), 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;
- }
- // A header contradicting the walk -- a foreign partition_id,
- // or a base offset that does not continue the chain in
- // EITHER direction -- is either damage wearing a decodable
- // header or a real record that does not belong at this
- // position. The batch checksum tells them apart, because it
- // covers both fields and the server mints every legal record
- // with it: a batch that fails it is damage, so break and let
- // the probe below classify the residue (a bit flip in a tail
- // batch then truncates like any torn tail, regardless of the
- // flip's direction), while a batch that VERIFIES is durable
- // evidence -- of a misdirected or copied write when the
- // partition stamp is foreign, of duplicated or lost offsets
- // when the chain breaks -- and both absorbing it and
- // truncating it would hide that, so refuse and keep the
- // bytes. Refusing the verified forward gap over absorbing it
- // is deliberate: state transfer's own walk refuses any gap,
- // so an absorbed one would mint a segment no peer can ever
- // install, and the offsets it fabricates seed current_offset
- // under an advance-only superblock persist.
- let foreign_partition = header.partition_id !=
identity.partition_id as u64;
- if foreign_partition || header.base_offset != expected_offset {
- 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 foreign_partition {
- return
Err(identity.refusal(PartitionRecoveryRefusal::ForeignBatch {
- start_offset,
- batch_partition_id: header.partition_id,
- position,
- }));
- }
- return Err(
-
identity.refusal(PartitionRecoveryRefusal::OffsetDiscontinuity {
- start_offset,
- expected_offset,
- found_offset: header.base_offset,
- position,
- }),
+ let mut walk = walk_chain_from_anchor(
Review Comment:
The walk still starts at `last.position`, so `[0, last.position)` is
accepted on index evidence alone, which is the evidence this PR itself declares
insufficient (`server_error.rs:288-294`, and :1370-1377 here).
On master that was safe for a specific reason: `save_frozen_batches` fsyncs
the whole file (`messages_writer.rs:129-131`), not a range, so under the old
serial order "index entry N is durable" implied "log fdatasync N completed"
implied "log bytes [0, N) are all durable". The `futures::future::join` in
`iggy_partition.rs` deletes that implication. So under `enforce_fsync = true`
this is newly unsound rather than pre-existing, and the premise now applies to
every entry rather than only the last.
`stats.increment_messages_count(end_offset - start_offset + 1)` at :339 then
reports a count over a span nothing read.
Suggest checksum-walking from byte 0 whenever the index was floored or the
segment was not cleanly sealed. At minimum, stop deriving `messages_count` from
an unwalked span.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -3090,9 +3094,9 @@ where
// Persist BEFORE eviction so a write failure leaves the rest of
the
// committed prefix resident for retry. The persist is idempotent
on
- // failure: a batch write that lands but whose index save then
fails
- // rewinds the segment write cursor, so the retry overwrites those
- // bytes instead of appending a duplicate. Chunks already durable
+ // failure: it advances its write cursors only once both the batch
+ // and the index are durable, so the retry overwrites the same
+ // positions instead of appending a duplicate. Chunks already
durable
Review Comment:
There is no retry. The only production consumer of a persist `Err` is
`commit_messages_inner` -> `commit_partition_entry` (sets `failed_commit`,
returns false) -> `handle_committed_entries`, which panics at :3246. Nothing
re-invokes the persist against the un-advanced cursors. The shutdown path
(`flush_committed_messages`) warns and moves to the next namespace, and
`state_transfer` re-arms the transfer rather than the persist.
That panic also does not do what its own comment promises. The shard message
pump runs inside `compio::runtime::spawn` (`bootstrap.rs:1274`), which this
repo documents as silently swallowing spawned-task panics
(`core/message_bus/tests/installer_panic_cleanup.rs:20-27`, and the two
production scopeguards built around it at `installer/replica.rs:424-430` and
`installer/tcp.rs:112-117`). There is no `panic = "abort"` and no custom hook.
So an ENOSPC or EIO unwinds one shard's pump and the process stays up: the
listener keeps accepting, other shards keep running, and every partition on
that shard silently stops committing, ticking and replying. `ShardPumpDied` is
only raised by `drain_shard_pump`, which runs at shutdown.
The cursor discipline itself is worth keeping: it is what makes the
shutdown-flush warn-and-continue path non-corrupting, and boot recovery is the
real retry since `SegmentStorage::new` reopens at the recovered lengths.
Suggest dropping the retry framing here and at :3743-3749 and documenting boot
recovery as the sole recovery, then fixing the wedge separately.
--
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]
