This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 7b491c09f fix(cluster): prevent repaired prepares from rewinding WAL
parent (#3978)
7b491c09f is described below
commit 7b491c09fa73caab9aeda60b84215e686fe4967b
Author: Piotr Gankiewicz <[email protected]>
AuthorDate: Thu Aug 27 19:19:41 2026 +0200
fix(cluster): prevent repaired prepares from rewinding WAL parent (#3978)
Metadata repair could backfill an operation below the sequencer
frontier and still replace the checksum used to parent the next
prepare. A promoted replica then wrote a higher operation whose
parent skipped committed WAL entries. Recovery correctly rejected
the interior chain break and refused to truncate a suffix that
can hold committed operations, leaving every node that durably
held the bad entry unable to restart.
Resolve the repaired frontier in one place: walk the contiguous
journal above the current head and update the sequencer and
last_prepare_checksum together or not at all. The checksum is
always read from the journal entry sitting at the resulting
head, so the frame that happened to close a gap cannot stamp its
own checksum as the parent, and a backfill below the head leaves
the head where it is. The head is re-anchored, not only advanced:
handle_start_view moves the sequencer to the announced head
without touching the checksum, so the repair frame that finally
delivers that head is the one chance to make the pair agree
again. The helper lives in consensus and both repair ingest
paths use it.
The partition path carried the same pattern. It has not produced
the same failure, but not because its prepares are unsealed:
project() seals the header checksum on both planes, and only
checksum_body is metadata-only. The partition journal is in
memory, so nothing refuses to boot on it. The chain is still
verified: canonical_headers walks the DVC suffix and refuses a
merge whose headers do not hash-chain, and the partition plane
feeds it real headers. A rewound partition parent left in the
uncommitted suffix therefore surfaces at the next partition view
change as a merge the new primary refuses to install, and that
group cannot complete the view change while the suffix is what
its replicas offer. Fixing it on both planes.
Both ingest sites are covered, and each test fails without the
fix. The partition call site is pinned by unit tests over
apply_repaired_prepare: one for a backfill below the head, one
for the frame that closes a gap. The metadata call site is
covered in the simulator, which drives a real IggyShard: a backup
crashes, misses committed operations, then rejoins and repairs.
It asserts that no entry landing below the sequencer becomes
last_prepare_checksum while the stream runs, and that once the
stream has run the pair describes one and the same journal
entry. That catches the rewind at the backfill rather than
waiting for a projected prepare and for the journal rewrite that
finally makes the broken chain adjacent on disk.
---------
Co-authored-by: Hubert Gruszecki <[email protected]>
---
core/consensus/src/plane_helpers.rs | 62 ++++++++++++++
core/journal/src/prepare_journal.rs | 40 +++++++++
core/partitions/src/iggy_partition.rs | 121 +++++++++++++++++++++++---
core/shard/src/lib.rs | 29 +++----
core/simulator/src/lib.rs | 155 ++++++++++++++++++++++++++++++++++
5 files changed, 378 insertions(+), 29 deletions(-)
diff --git a/core/consensus/src/plane_helpers.rs
b/core/consensus/src/plane_helpers.rs
index 4e4bcd7ef..ff7df19cc 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -771,6 +771,29 @@ pub fn panic_if_hash_chain_would_break_in_same_view(
}
}
+/// Resolve a repair's contiguous head and the checksum of the entry sitting
at it.
+///
+/// A view change can put the sequencer above holes that repair later
backfills.
+/// Such a lower entry changes durable coverage, but not the head the next
prepare
+/// must parent: taking its checksum would rewind the chain while leaving the
+/// sequencer ahead of it. `None` until the journal holds the entry at the
head.
+pub fn repaired_frontier_update(
+ previous_frontier: u64,
+ mut header_at: impl FnMut(u64) -> Option<PrepareHeader>,
+) -> Option<(u64, u128)> {
+ let mut frontier = previous_frontier;
+ // Re-anchored on the head the journal can prove, not only on an advance:
+ // `handle_start_view` moves the sequencer to the announced head without
+ // touching the checksum, so the frame that finally delivers that head is
+ // the only chance to make the pair agree again.
+ let mut checksum = header_at(frontier).map(|header| header.checksum);
+ while let Some(header) = header_at(frontier + 1) {
+ frontier += 1;
+ checksum = Some(header.checksum);
+ }
+ checksum.map(|checksum| (frontier, checksum))
+}
+
/// Ack a prepare back to its primary once the owning plane vouches for it.
///
/// `is_persisted` is the caller's journal-containment verdict for `header`:
@@ -856,6 +879,7 @@ mod tests {
use iggy_common::calculate_checksum;
use message_bus::SendError;
use server_common::{MESSAGE_ALIGN, iobuf::Frozen};
+ use std::collections::BTreeMap;
/// `PrepareHeader`'s alignment, which every suffix body has to satisfy.
const BODY_ALIGN: usize = align_of::<PrepareHeader>();
@@ -2022,4 +2046,42 @@ mod tests {
);
assert!(header.validate().is_ok());
}
+
+ fn header_with_checksum(op: u64, checksum: u128) -> PrepareHeader {
+ PrepareHeader {
+ command: Command::Prepare,
+ op,
+ checksum,
+ ..Default::default()
+ }
+ }
+
+ #[test]
+ fn
given_lower_backfill_when_resolving_repaired_frontier_should_not_rewind() {
+ let headers = BTreeMap::from([
+ (169, header_with_checksum(169, 1690)),
+ (170, header_with_checksum(170, 1700)),
+ (171, header_with_checksum(171, 1710)),
+ ]);
+
+ let update = repaired_frontier_update(171, |op|
headers.get(&op).copied());
+
+ // Not a rewind: the frontier does not move, and the checksum
re-anchors
+ // on the entry the sequencer already points at rather than on the
+ // backfill that just arrived below it.
+ assert_eq!(update, Some((171, 1710)));
+ }
+
+ #[test]
+ fn
given_gap_closure_when_resolving_repaired_frontier_should_use_new_head_checksum()
{
+ let headers = BTreeMap::from([
+ (170, header_with_checksum(170, 1700)),
+ (171, header_with_checksum(171, 1710)),
+ (172, header_with_checksum(172, 1720)),
+ ]);
+
+ let update = repaired_frontier_update(169, |op|
headers.get(&op).copied());
+
+ assert_eq!(update, Some((172, 1720)));
+ }
}
diff --git a/core/journal/src/prepare_journal.rs
b/core/journal/src/prepare_journal.rs
index 4c0f3dc54..278fb8e0e 100644
--- a/core/journal/src/prepare_journal.rs
+++ b/core/journal/src/prepare_journal.rs
@@ -1444,6 +1444,46 @@ mod tests {
);
}
+ #[compio::test]
+ async fn scan_refuses_boot_on_interior_parent_chain_break() {
+ // The shape a repair frontier rewind leaves behind: every entry is
+ // complete and individually well sealed, op 3 parents to op 1 instead
+ // of op 2, and a valid chain continues from op 3. Complete entries
+ // after the break can be quorum committed, so boot must refuse rather
+ // than truncate them away as a torn tail.
+ const BODY: usize = 64;
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ {
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ let op_1 = make_identity_sealed_prepare(1, BODY, 0);
+ let checksum_1 = op_1.header().checksum;
+ journal.append(op_1.deep_copy()).await.unwrap();
+ let op_2 = make_identity_sealed_prepare(2, BODY, checksum_1);
+ journal.append(op_2.deep_copy()).await.unwrap();
+ let op_3 = make_identity_sealed_prepare(3, BODY, checksum_1);
+ let checksum_3 = op_3.header().checksum;
+ journal.append(op_3.deep_copy()).await.unwrap();
+ let op_4 = make_identity_sealed_prepare(4, BODY, checksum_3);
+ journal.append(op_4.deep_copy()).await.unwrap();
+ }
+ let bytes_before = std::fs::metadata(&path).unwrap().len();
+
+ let error = PrepareJournal::open(&path, 0).await.unwrap_err();
+
+ let message = error.to_string();
+ assert!(
+ message.contains("interior WAL corruption") &&
message.contains("does not chain"),
+ "a rewound parent with a complete suffix following must refuse \
+ boot rather than truncate, got: {message}"
+ );
+ assert_eq!(
+ std::fs::metadata(&path).unwrap().len(),
+ bytes_before,
+ "the refusal must leave the WAL bytes untouched"
+ );
+ }
+
#[compio::test]
async fn scan_skips_verification_for_unsealed_entries() {
// A WAL from a pre-sealing build must still open: `checksum` reads as
the
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 17bf8aefa..dc43e0be3 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -40,7 +40,7 @@ use consensus::{
ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind, VsrConsensus,
ack_preflight,
ack_quorum_reached, build_deny_reply_from_request,
build_reply_from_request,
build_reply_message, drain_committable_prefix,
emit_namespace_progress_event,
- emit_partition_diag, emit_sim_event, fence_old_prepare_by_commit,
+ emit_partition_diag, emit_sim_event, fence_old_prepare_by_commit,
repaired_frontier_update,
replicate_frozen_to_next_in_chain, replicate_preflight,
restamp_prepare_view,
send_prepare_ok as send_prepare_ok_common, verify_prepare_integrity,
};
@@ -4447,21 +4447,18 @@ where
// cannot walk. A dropped frame stalls the frontier here; the stall
// retry refills the hole and the next apply resumes the advance
// (walking over ops that were journaled out of order meanwhile).
- let mut frontier = self.consensus().sequencer().current_sequence();
- while self
- .log
- .journal()
- .inner
- .header_by_op(frontier + 1)
- .is_some()
- {
- frontier += 1;
- }
- let consensus = self.consensus();
- if frontier > consensus.sequencer().current_sequence() {
+ // The checksum moves only with the frontier and is read from the
+ // journal header at the new head: a lower backfill must not rewind
+ // the parent the next prepare chains onto.
+ let previous_frontier =
self.consensus().sequencer().current_sequence();
+ let update = repaired_frontier_update(previous_frontier, |op| {
+ self.log.journal().inner.header_by_op(op)
+ });
+ if let Some((frontier, frontier_checksum)) = update {
+ let consensus = self.consensus();
consensus.sequencer().set_sequence(frontier);
+ consensus.set_last_prepare_checksum(frontier_checksum);
}
- consensus.set_last_prepare_checksum(header.checksum);
}
/// Conclude a repair stream: settle the commit floor at the serving
@@ -6567,6 +6564,102 @@ mod tests {
.expect("journal append");
}
+ /// A repaired `SendMessages` prepare with an explicit chain identity, as a
+ /// serving peer ships it.
+ fn repaired_send_prepare(op: u64, parent: u128, checksum: u128) ->
Message<PrepareHeader> {
+ let namespace = IggyNamespace::new(1, 1, 0);
+ let record = build_segment_record(namespace, op);
+ let header_size = std::mem::size_of::<PrepareHeader>();
+ let total = header_size + record.len();
+ let mut message = Message::<PrepareHeader>::new(total);
+ message.as_mut_slice()[header_size..].copy_from_slice(&record);
+ message.transmute_header(|_, header: &mut PrepareHeader| {
+ header.command = Command::Prepare;
+ header.operation = Operation::SendMessages;
+ header.op = op;
+ header.parent = parent;
+ header.checksum = checksum;
+ header.group = namespace.inner();
+ header.size = u32::try_from(total).expect("prepare size fits u32");
+ })
+ }
+
+ #[compio::test]
+ async fn
given_lower_backfill_when_applying_repaired_prepare_should_not_rewind_parent() {
+ // The call-site regression for the WAL frontier rewind: a repaired op
+ // below the DVC-adopted head must leave BOTH halves of the frontier
+ // alone. The old code left the sequencer at the head and rewound
+ // `last_prepare_checksum` to the backfilled entry, so the next prepare
+ // parented past a committed op and recovery refused the WAL.
+ const CHECKSUM_1: u128 = 0x11;
+ const CHECKSUM_2: u128 = 0x22;
+ const CHECKSUM_3: u128 = 0x33;
+ let mut partition = test_partition();
+ partition.repair = Some(armed_session(3, 0, None));
+
+ partition
+ .apply_repaired_prepare(repaired_send_prepare(1, 0, CHECKSUM_1))
+ .await;
+ partition
+ .apply_repaired_prepare(repaired_send_prepare(3, CHECKSUM_2,
CHECKSUM_3))
+ .await;
+ // The hole at op 2 stalls the frontier advance.
+ assert_eq!(partition.consensus().sequencer().current_sequence(), 1);
+
+ // The state a DoViewChange merge leaves: head 3, parented on its
+ // checksum, with the hole at op 2 still unfilled locally.
+ partition.consensus().sequencer().set_sequence(3);
+ partition.consensus().set_last_prepare_checksum(CHECKSUM_3);
+
+ partition
+ .apply_repaired_prepare(repaired_send_prepare(2, CHECKSUM_1,
CHECKSUM_2))
+ .await;
+
+ assert!(
+ partition.log.journal().inner.header_by_op(2).is_some(),
+ "the backfill must be journaled"
+ );
+ assert_eq!(partition.consensus().sequencer().current_sequence(), 3);
+ assert_eq!(
+ partition.consensus().last_prepare_checksum(),
+ CHECKSUM_3,
+ "a lower backfill must not rewind the parent of the next prepare"
+ );
+ }
+
+ #[compio::test]
+ async fn
given_gap_closure_when_applying_repaired_prepare_should_adopt_new_head_checksum()
{
+ // The frame that closes a gap is not the new head: the frontier walks
+ // to the highest contiguous op and the checksum must come from THAT
+ // journal entry, not from the repair frame that happened to arrive
+ // last.
+ const CHECKSUM_1: u128 = 0x11;
+ const CHECKSUM_2: u128 = 0x22;
+ const CHECKSUM_3: u128 = 0x33;
+ let mut partition = test_partition();
+ partition.repair = Some(armed_session(3, 0, None));
+
+ partition
+ .apply_repaired_prepare(repaired_send_prepare(1, 0, CHECKSUM_1))
+ .await;
+ partition
+ .apply_repaired_prepare(repaired_send_prepare(3, CHECKSUM_2,
CHECKSUM_3))
+ .await;
+ assert_eq!(partition.consensus().sequencer().current_sequence(), 1);
+ assert_eq!(partition.consensus().last_prepare_checksum(), CHECKSUM_1);
+
+ partition
+ .apply_repaired_prepare(repaired_send_prepare(2, CHECKSUM_1,
CHECKSUM_2))
+ .await;
+
+ assert_eq!(partition.consensus().sequencer().current_sequence(), 3);
+ assert_eq!(
+ partition.consensus().last_prepare_checksum(),
+ CHECKSUM_3,
+ "a gap closure must adopt the checksum of the new contiguous head"
+ );
+ }
+
#[compio::test]
async fn
given_session_remint_when_attempts_burned_should_survive_on_partition() {
let mut partition = test_partition();
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 1efaa598a..cf220f0d3 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -32,7 +32,8 @@ use consensus::{
DvcSuffix, FatalReason, MergedLog, MetadataHandle, MuxPlane,
PartitionsHandle, Pipeline, Plane,
PlaneKind, STATE_TRANSFER_MAX_DECODE_RETRIES,
STATE_TRANSFER_MAX_STALL_RETRIES, Sequencer,
Status, VsrAction, VsrConsensus, build_deny_reply_from_request_header,
dvc_blank,
- dvc_header_kind, encode_prepare_headers, fatal, restamp_prepare_view,
verify_prepare_integrity,
+ dvc_header_kind, encode_prepare_headers, fatal, repaired_frontier_update,
restamp_prepare_view,
+ verify_prepare_integrity,
};
#[cfg(any(test, feature = "simulator"))]
use crossfire::AsyncRxTrait;
@@ -4473,26 +4474,23 @@ where
// `apply_repaired_prepare`: DVC advertises the sequencer, so a
// hole below a repaired op must stall the advance rather than
// mint an election candidate with an unwalkable log.
- let mut frontier = consensus.sequencer().current_sequence();
+ let previous_frontier = consensus.sequencer().current_sequence();
#[allow(clippy::cast_possible_truncation)]
- while journal.header((frontier + 1) as usize).is_some() {
- frontier += 1;
- }
- if frontier > consensus.sequencer().current_sequence() {
+ let update = repaired_frontier_update(previous_frontier, |op| {
+ journal.header(op as usize).map(|header| *header)
+ });
+ if let Some((frontier, frontier_checksum)) = update {
consensus.sequencer().set_sequence(frontier);
+ consensus.set_last_prepare_checksum(frontier_checksum);
}
- consensus.set_last_prepare_checksum(header.checksum);
return;
}
// A metadata-plane op that did not match above (no metadata consensus
on
// this shard, or a namespace neither plane claims) is DROPPED, never
- // offered to the partition arm. Falling through let a metadata prepare
- // reach `apply_repaired_prepare`: it journals nothing, but it resets
the
- // partition repair session's idle ticks (masking a genuine stall) and
- // carries the metadata prepare's checksum into the partition consensus
- // via `set_last_prepare_checksum` -- inert only while prepare
checksums
- // are structurally zero, and a cross-plane parent stamp the moment the
- // checksum chain is activated (see the note in `consensus::impls`).
+ // offered to the partition arm. Falling through would let a metadata
+ // prepare reach `apply_repaired_prepare`: it journals nothing and
never
+ // reaches the frontier update, but it resets the partition repair
+ // session's idle ticks, masking a genuine stall.
if metadata_plane_op {
tracing::debug!(
shard = self.id,
@@ -9417,9 +9415,10 @@ mod persist_gate_tests {
mod repair_scope_tests {
//! Who parked the log decides what it means.
- use super::{MergedLog, repair_op_in_scope, repair_serve_ceiling};
use iggy_binary_protocol::{Command, PrepareHeader};
+ use super::{MergedLog, repair_op_in_scope, repair_serve_ceiling};
+
fn header(op: u64) -> PrepareHeader {
PrepareHeader {
command: Command::Prepare,
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 8e971c708..4553e1ad6 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -3530,3 +3530,158 @@ mod view_change_data_loss_tests {
);
}
}
+
+#[cfg(test)]
+mod repair_frontier_tests {
+ //! Journal repair moves durable coverage, not the head of the hash chain.
+ //!
+ //! A replica that rejoins behind the group adopts the primary's head and
+ //! then backfills the ops it missed. Those land BELOW that head, so the
+ //! pair `(sequencer, last_prepare_checksum)` has to keep describing one
+ //! and the same entry: the pair is exactly what the next projected prepare
+ //! stamps as `(op, parent)`. Carrying the repaired frame's own checksum
+ //! instead rewinds the parent, and the next prepare then chains past the
+ //! entry that actually precedes it -- every entry individually well
+ //! sealed, the chain broken, and a WAL that refuses to boot as soon as a
+ //! rewrite (checkpoint drain or uncommitted-suffix truncation) puts the
+ //! two entries side by side.
+
+ use super::*;
+ use consensus::Sequencer;
+ use journal::Journal;
+ use std::collections::BTreeMap;
+
+ /// Replica 0 is primary for view 0, so this one stays a backup for the
+ /// whole run: only the repair ingest is under test, not an election.
+ const LAGGING: u8 = 1;
+
+ /// Committed ops the lagging replica misses and has to repair back.
+ const OPS_MISSED: usize = 20;
+
+ /// Steps allowed for the rejoin, the adoption, and the repair stream.
+ const REPAIR_STEPS: usize = 4000;
+
+ /// Highest op the journal probe walks. The workload stays far below it.
+ const OP_PROBE_CEILING: u64 = 512;
+
+ /// `(sequencer, last_prepare_checksum)` of a replica's metadata consensus:
+ /// the `(op, parent)` its next projected prepare would stamp.
+ fn metadata_chain_head(sim: &Simulator, replica: u8) -> (u64, u128) {
+ let consensus = sim.replicas[replica as usize].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .expect("shard 0 owns metadata consensus");
+ (
+ consensus.sequencer().current_sequence(),
+ consensus.last_prepare_checksum(),
+ )
+ }
+
+ /// Every op a replica's metadata journal holds, with its checksum.
+ fn metadata_journal_checksums(sim: &Simulator, replica: u8) ->
BTreeMap<u64, u128> {
+ let journal = sim.replicas[replica as usize].shards[0]
+ .plane
+ .metadata()
+ .journal
+ .as_ref()
+ .expect("shard 0 owns the metadata journal");
+ (1..=OP_PROBE_CEILING)
+ .filter_map(|op| {
+ let slot = usize::try_from(op).expect("op fits usize");
+ Journal::header(journal.as_ref(), slot).map(|header| (op,
header.checksum))
+ })
+ .collect()
+ }
+
+ #[test]
+ fn
given_repair_below_the_head_when_backfilling_should_not_rewind_the_parent() {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let client_id: u128 = 1;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ replica_count as usize,
+ std::iter::once(client_id),
+ network_opts,
+ );
+ let client = SimClient::new(client_id);
+
+ sim.register_client_with_primary(&client);
+ for _ in 0..50 {
+ sim.step();
+ }
+
+ sim.replica_crash(LAGGING);
+ for index in 0..OPS_MISSED {
+ let msg = client.create_stream(&format!("gap-{index}"));
+ sim.submit_request(client_id, 0, msg.into_generic());
+ for _ in 0..30 {
+ sim.step();
+ }
+ }
+
+ sim.replica_restart(LAGGING);
+
+ // Everything the restart recovered locally is the baseline; anything
+ // that appears from here arrived over the wire.
+ let mut journaled = metadata_journal_checksums(&sim, LAGGING);
+ let mut backfills_below_head = 0usize;
+ for _ in 0..REPAIR_STEPS {
+ sim.step();
+ let (head, parent) = metadata_chain_head(&sim, LAGGING);
+ let current = metadata_journal_checksums(&sim, LAGGING);
+ for (&op, &checksum) in ¤t {
+ // A live replicated op IS the head, and the pair moves with
it.
+ // Only entries that landed below the head are repair backfill.
+ if journaled.contains_key(&op) || op >= head {
+ continue;
+ }
+ backfills_below_head += 1;
+ assert_ne!(
+ parent,
+ checksum,
+ "repairing op {op} rewound the parent of the next prepare:
the \
+ sequencer sits at op {head} but last_prepare_checksum now
\
+ describes op {op}, so the next projected prepare would be
op \
+ {} parented past op {head}",
+ head + 1
+ );
+ }
+ journaled = current;
+ }
+
+ // The contract itself, not just the absence of a rewind: after the
+ // repair stream the pair has to describe one and the same entry, or
+ // the next projected prepare parents on something that is not its
+ // predecessor.
+ let (head, parent) = metadata_chain_head(&sim, LAGGING);
+ let journaled = metadata_journal_checksums(&sim, LAGGING);
+ assert_eq!(
+ journaled.get(&head).copied(),
+ Some(parent),
+ "the sequencer sits at op {head} but last_prepare_checksum
describes \
+ op {:?}, so the next projected prepare would parent past op
{head}",
+ journaled
+ .iter()
+ .find(|&(_, &checksum)| checksum == parent)
+ .map(|(&op, _)| op)
+ );
+
+ assert!(
+ backfills_below_head > 0,
+ "the rejoined replica never repaired an op below its own head, so \
+ nothing about the repair frontier was exercised"
+ );
+ }
+}
