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 73a0bed8b fix(consensus): count an unservable header as a nack (#4109)
73a0bed8b is described below
commit 73a0bed8be2cf4ed247f5f735106dbc3f0ac85ef
Author: Krishna Vishal <[email protected]>
AuthorDate: Fri Sep 11 18:26:58 2026 +0530
fix(consensus): count an unservable header as a nack (#4109)
A view change can reach a state where an op is neither recoverable from
the `DoViewChange` messages in hand nor provably uncommitted.
`merge_dvc_quorum` answers `AwaitingRepair` and waits for every replica
to report, so a permanently crashed replica leaves two healthy survivors
unable to elect a primary. The view number keeps climbing while
`log_view` stays put, and client requests retry forever.
The op was decidable from the quorum already present. A sender that
holds the header but cannot serve the body counted as neither a copy nor
a nack. A prepare is journaled before it is acked and nothing compacts
an op above the commit point, so a missing body proves that sender never
acked it, which is what a nack asserts.
`tally_op` now derives a nack from that sender. The quorum is unchanged
at `replica_count - quorum_replication + 1`, so replicas provably
outside the ack set still leave fewer than a replication quorum inside
it, and a sender that can serve the body is untouched. The derived nack
stops at the sender's own commit point, where a missing body means
compaction rather than absence.
Two unit tests now assert truncation where they asserted a stall, one
pins the boundary at a single servable copy, and the undecidable case
moves to abstention. A simulator test replays the seed that found this:
without the change the cluster reaches view 103 against `log_view` 3
with both survivors caught up, and one request retries 266 times
unanswered.
---
core/consensus/src/dvc_merge.rs | 86 +++++++++++++------
core/consensus/src/impls.rs | 82 +++++++++++++------
core/consensus/src/plane_helpers.rs | 150 +++++++++++++++++++++++++++++++++-
core/metadata/src/impls/metadata.rs | 19 ++++-
core/partitions/src/iggy_partition.rs | 49 ++++++-----
core/shard/src/lib.rs | 31 ++++---
core/simulator/src/lib.rs | 78 ++++++++++++++++++
7 files changed, 401 insertions(+), 94 deletions(-)
diff --git a/core/consensus/src/dvc_merge.rs b/core/consensus/src/dvc_merge.rs
index 516ef8ff8..0f8dfca4d 100644
--- a/core/consensus/src/dvc_merge.rs
+++ b/core/consensus/src/dvc_merge.rs
@@ -210,8 +210,9 @@ fn tally_op<'a>(
};
let held = dvc.suffix.valid_header_at(index);
+ let offers_body = dvc.suffix.offers_body(index);
if let (Some(held), Some(canonical)) = (held, canonical)
- && dvc.suffix.offers_body(index)
+ && offers_body
&& held.checksum == canonical.checksum
{
copies += 1;
@@ -221,21 +222,31 @@ fn tally_op<'a>(
// Explicit: the sender proves it never prepared this op.
nacks += 1;
} else if let Some(held) = held {
- // Only a sender BEHIND the canonical log_view can implicitly nack.
- // Without this, corrupting one canonical header in transit turns
every
- // honest sender's correct header into an implicit nack against the
- // garbage: a nack quorum on three replicas. A same-log_view
- // disagreement is evidence, not a vote, and goes through
`conflict`.
- let may_nack_implicitly = dvc.log_view < canonical_log_view;
- match canonical {
- // Implicit: the sender holds a DIFFERENT prepare, so not this
one.
- Some(canonical) if may_nack_implicitly && held.checksum !=
canonical.checksum => {
- nacks += 1;
+ if !offers_body && op > dvc.commit {
+ // A prepare is journaled before it is acked, so a header this
+ // sender cannot serve, above its own commit, proves it is
outside
+ // the ack set. Below that commit a missing body is compaction.
+ nacks += 1;
+ } else {
+ // Only a sender BEHIND the canonical log_view can implicitly
+ // nack. Without this, corrupting one canonical header in
transit
+ // turns every honest sender's correct header into an implicit
+ // nack against the garbage: a nack quorum on three replicas. A
+ // same-log_view disagreement is evidence, not a vote, and goes
+ // through `conflict`.
+ let may_nack_implicitly = dvc.log_view < canonical_log_view;
+ match canonical {
+ // Implicit: the sender holds a DIFFERENT prepare, so not
this one.
+ Some(canonical)
+ if may_nack_implicitly && held.checksum !=
canonical.checksum =>
+ {
+ nacks += 1;
+ }
+ // Implicit: no canonical sender holds anything here, so a
newer
+ // view already truncated this op and the sender holds a
corpse.
+ None if may_nack_implicitly => nacks += 1,
+ _ => {}
}
- // Implicit: no canonical sender holds anything here, so a
newer
- // view already truncated this op and the sender holds a
corpse.
- None if may_nack_implicitly => nacks += 1,
- _ => {}
}
}
}
@@ -707,23 +718,24 @@ mod tests {
}
#[test]
- fn
given_header_without_a_servable_body_when_replicas_outstanding_should_await_repair()
{
- // Both senders have op 4's header, neither can serve its body, and
replica 2
- // has not reported. A head whose body nobody holds would wedge the
view.
+ fn
given_header_without_a_servable_body_when_replicas_outstanding_should_truncate()
{
+ // Neither sender journaled op 4, so the ack set was at most replica 2:
+ // short of a replication quorum, so it never committed.
let mut quorum = dvc_quorum_array_empty();
let headers = suffix_headers(2, 4, 1);
let header_only = DvcSuffix::new(headers, 0, 0b110);
dvc_record(&mut quorum, dvc(0, 1, 4, 2, header_only.clone()));
dvc_record(&mut quorum, dvc(1, 1, 4, 2, header_only));
- assert_eq!(
- merge_dvc_quorum(&quorum, quorums_r3()),
- MergeOutcome::AwaitingRepair { undecided_op: 4 }
- );
+ let MergeOutcome::Ready(log) = merge_dvc_quorum(&quorum, quorums_r3())
else {
+ panic!("two senders provably outside the ack set decide op 4");
+ };
+ assert_eq!(log.op_head, 3, "op 4 is truncated, not awaited");
+ assert_eq!(log.commit_max, 2);
}
#[test]
- fn given_all_replicas_reported_and_op_undecidable_should_deadlock() {
+ fn
given_all_replicas_reported_and_no_body_should_truncate_rather_than_deadlock() {
let mut quorum = dvc_quorum_array_empty();
let headers = suffix_headers(2, 4, 1);
let header_only = DvcSuffix::new(headers, 0, 0b110);
@@ -731,10 +743,32 @@ mod tests {
dvc_record(&mut quorum, dvc(1, 1, 4, 2, header_only.clone()));
dvc_record(&mut quorum, dvc(2, 1, 4, 2, header_only));
+ let MergeOutcome::Ready(log) = merge_dvc_quorum(&quorum, quorums_r3())
else {
+ panic!("no replica can serve op 4 and all three say so");
+ };
+ assert_eq!(log.op_head, 3);
+ }
+
+ #[test]
+ fn
given_one_servable_copy_against_a_header_only_sender_should_keep_the_op() {
+ // Boundary: one servable copy outranks one derived nack.
+ let mut quorum = dvc_quorum_array_empty();
+ let headers = suffix_headers(2, 4, 1);
+ dvc_record(
+ &mut quorum,
+ dvc(0, 1, 4, 2, DvcSuffix::new(headers.clone(), 0, 0b111)),
+ );
+ dvc_record(
+ &mut quorum,
+ dvc(1, 1, 4, 2, DvcSuffix::new(headers, 0, 0b110)),
+ );
+
+ let MergeOutcome::Ready(log) = merge_dvc_quorum(&quorum, quorums_r3())
else {
+ panic!("op 4 is recoverable from replica 0, so the view must
start");
+ };
assert_eq!(
- merge_dvc_quorum(&quorum, quorums_r3()),
- MergeOutcome::Deadlocked { undecided_op: 4 },
- "with every replica in, an unrecoverable op stalls the view
forever"
+ log.op_head, 4,
+ "a servable copy keeps the op against a single derived nack"
);
}
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 4a2ae9785..649843209 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -1106,8 +1106,8 @@ where
/// built-in default.
probe_attempts_max: Cell<u32>,
- /// This replica's own uncommitted suffix, with the `(op, commit)` the
journal
- /// was at when it was read.
+ /// This replica's own uncommitted suffix, with the head, commit point and
+ /// mutation count the journal was at when it was read.
///
/// Installed by the shard via [`Self::set_local_dvc_suffix`] before any
handler
/// that could enter a view change. Snapshotted rather than recomputed per
send
@@ -1115,11 +1115,24 @@ where
/// replica's log, and silently retracting one lets the new primary
assemble a
/// quorum that never simultaneously existed.
///
- /// Tagged by `(op, commit)`, not by view, because that is what the suffix
- /// describes: a view advance leaves the log alone so the snapshot
survives,
- /// while anything moving the head or commit point makes the tag mismatch,
- /// which reads as no snapshot at all.
- local_dvc_suffix: RefCell<Option<(u64, u64, DvcSuffix)>>,
+ /// Tagged by `(op, commit, journal_mutations)`, not by view, because that
is
+ /// what the suffix describes: a view advance leaves the log alone so the
+ /// snapshot survives, while anything moving the head, the commit point or
the
+ /// journal's contents makes the tag mismatch, which reads as no snapshot
at
+ /// all.
+ local_dvc_suffix: RefCell<Option<(u64, u64, u64, DvcSuffix)>>,
+
+ /// Counts mutations of the journal this replica's suffix is read from.
+ ///
+ /// `(op, commit)` alone says how far the log reaches, not what it holds. A
+ /// backup that adopted a `StartView` sits at the announced head with the
+ /// bodies still missing, so repair filling one moves neither number, and
the
+ /// header-only snapshot taken before it would keep going out. The merge
reads
+ /// a header no sender can serve as proof that sender never journaled the
op,
+ /// which is now a nack: a stale snapshot would nack an op this replica
holds
+ /// and acked. Bumped by every append, truncation, drain and eviction on
both
+ /// planes.
+ journal_mutations: Cell<u64>,
/// The log a DVC quorum settled on, parked until this replica's journal
can
/// serve all of it.
@@ -1425,6 +1438,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
probe_attempts: Cell::new(0),
probe_attempts_max: Cell::new(PROBE_ATTEMPTS_MAX),
local_dvc_suffix: RefCell::new(None),
+ journal_mutations: Cell::new(0),
pending_view_log: RefCell::new(None),
do_view_change_from_all_replicas:
RefCell::new(dvc_quorum_array_empty()),
do_view_change_quorum: Cell::new(false),
@@ -2191,28 +2205,44 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
/// Installing twice for one view overwrites: the shard refreshes before
each
/// handler, and a later suffix is at least as complete (repair only adds).
pub fn set_local_dvc_suffix(&self, suffix: DvcSuffix) {
- let (op, commit) = self.local_dvc_suffix_tag();
- *self.local_dvc_suffix.borrow_mut() = Some((op, commit, suffix));
+ let (op, commit, mutations) = self.local_dvc_suffix_tag();
+ *self.local_dvc_suffix.borrow_mut() = Some((op, commit, mutations,
suffix));
}
/// Drop the cached suffix snapshot.
///
- /// The `(op, commit)` tag tracks how far the log reaches, not what it
still
- /// contains, so a mutation that removes entries without moving either
- /// (truncating a diverging uncommitted range) leaves a snapshot reading as
- /// current while offering bodies this replica can no longer serve. A peer
that
- /// picks it as a body source then waits out the whole view change.
- ///
- /// Call from the mutation site. The next refresh re-reads the journal.
+ /// For a change the journal counter cannot see: `start_pending_view`
takes the
+ /// parked log the snapshot was stitched over, which alters the suffix
while
+ /// every journal entry stays put.
pub fn invalidate_local_dvc_suffix(&self) {
self.local_dvc_suffix.borrow_mut().take();
}
- /// The `(op, commit)` a snapshot must match to still describe this log.
- /// `commit` is clamped to `op` exactly as the outgoing DVC clamps it.
- fn local_dvc_suffix_tag(&self) -> (u64, u64) {
+ /// Record that the journal backing this replica's suffix changed: an
append, a
+ /// truncation, a drain, an eviction.
+ ///
+ /// Call from the mutation site. The next refresh re-reads the journal, and
+ /// until it does the snapshot reads as absent rather than as a
description of
+ /// a log that has moved on. Both directions matter: a removal leaves a
+ /// snapshot offering bodies this replica can no longer serve, stranding a
peer
+ /// that picks it as a repair source, and an append leaves one nacking an
op
+ /// this replica has since journaled and acked, which is a licence to
truncate
+ /// committed data.
+ pub fn note_journal_mutation(&self) {
+ self.journal_mutations
+ .set(self.journal_mutations.get().wrapping_add(1));
+ }
+
+ /// The `(op, commit, journal_mutations)` a snapshot must match to still
+ /// describe this log. `commit` is clamped to `op` exactly as the outgoing
DVC
+ /// clamps it.
+ fn local_dvc_suffix_tag(&self) -> (u64, u64, u64) {
let op = self.sequencer.current_sequence();
- (op, self.commit_max.get().min(op))
+ (
+ op,
+ self.commit_max.get().min(op),
+ self.journal_mutations.get(),
+ )
}
/// This replica's suffix snapshot, or an empty one when none matches the
log's
@@ -2223,7 +2253,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
pub fn local_dvc_suffix(&self) -> DvcSuffix {
let tag = self.local_dvc_suffix_tag();
match &*self.local_dvc_suffix.borrow() {
- Some((op, commit, suffix)) if (*op, *commit) == tag =>
suffix.clone(),
+ Some((op, commit, mutations, suffix)) if (*op, *commit,
*mutations) == tag => {
+ suffix.clone()
+ }
_ => DvcSuffix::empty(),
}
}
@@ -2235,7 +2267,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
let tag = self.local_dvc_suffix_tag();
!matches!(
&*self.local_dvc_suffix.borrow(),
- Some((op, commit, _)) if (*op, *commit) == tag
+ Some((op, commit, mutations, _)) if (*op, *commit, *mutations) ==
tag
)
}
@@ -2617,9 +2649,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
.reset(TimeoutKind::DoViewChangeMessage);
// NOT the snapshot the first send used: `build_do_view_change`
re-reads
- // `local_dvc_suffix()`, whose `(op, commit)` tag can have moved
since, in
- // which case it answers EMPTY, retracting every nack and body offer
already
- // sent. Survivable only because `dvc_record` drops a duplicate
sender, so the
+ // `local_dvc_suffix()`, whose tag can have moved since, in which case
it
+ // answers EMPTY, retracting every nack and body offer already sent.
+ // Survivable only because `dvc_record` drops a duplicate sender, so
the
// candidate keeps the first vote. Allow a retransmit to replace a
seated vote
// and this must pin the snapshot instead.
let action =
self.build_do_view_change(self.primary_index(self.view.get()));
diff --git a/core/consensus/src/plane_helpers.rs
b/core/consensus/src/plane_helpers.rs
index cb9605c35..437d2376d 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -1584,6 +1584,41 @@ mod tests {
(header, body)
}
+ /// A DVC carrying `op` and `commit` only. Abstention: counts toward the
+ /// view-change quorum, says nothing about any op.
+ fn dvc_numbers_only(
+ replica: u8,
+ view: u32,
+ log_view: u32,
+ op: u64,
+ commit: u64,
+ ) -> (iggy_binary_protocol::DoViewChangeHeader, Body) {
+ use iggy_binary_protocol::DoViewChangeHeader;
+
+ let headers: Vec<PrepareHeader> = Vec::new();
+ let body = encode_body(&headers);
+ let header = DoViewChangeHeader {
+ checksum: 0,
+ checksum_body: 0,
+ cluster: 0,
+ size: u32::try_from(std::mem::size_of::<DoViewChangeHeader>() +
body.len())
+ .expect("synthetic DVC frame fits u32"),
+ view,
+ release: 0,
+ command: Command::DoViewChange,
+ replica,
+ reserved_frame: [0; 66],
+ op,
+ commit,
+ group: 0,
+ log_view,
+ reserved: [0; 68],
+ nack_bitset: 0,
+ present_bitset: 0,
+ };
+ (header, body)
+ }
+
/// Headers for `low..=high`, high-to-low as a suffix requires, sealed and
/// chained the way a real producer writes them.
///
@@ -1647,6 +1682,116 @@ mod tests {
consensus.set_local_dvc_suffix(crate::dvc_merge::suffix_all_present(headers));
}
+ /// The replica that prepared the head is gone for good, one survivor
holds its
+ /// header without the body and the other never had it. Both survivors are
+ /// outside the ack set, so the op is truncated and the view starts.
Counting the
+ /// header-only sender as neither copy nor nack instead waits for the
crashed
+ /// replica forever.
+ #[test]
+ fn
given_a_crashed_body_holder_when_merging_should_truncate_and_start_the_view() {
+ // View 3 of 3 replicas elects this one.
+ let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus,
LocalPipeline::new());
+ consensus.init();
+ consensus.restore_commit_state(2, 2);
+ consensus.sequencer().set_sequence(4);
+ // Bit 0 is op 4: its header, never journaled.
+ let local = suffix_headers(2, 4, 0);
+
consensus.set_local_dvc_suffix(crate::view_change_quorum::DvcSuffix::new(local,
0, 0b110));
+
+ let _ = consensus.handle_start_view_change(PlaneKind::Metadata,
&svc_header(1, 3));
+
+ // Replica 1 stops at op 3. Replica 2 held the only body and never
reports.
+ let (dvc, body) = dvc_with_suffix(1, 3, 0, 3, 2, None);
+ let _ = consensus.handle_do_view_change(PlaneKind::Metadata, &dvc,
&body);
+
+ let pending = consensus
+ .pending_view_log()
+ .expect("two senders outside op 4's ack set decide it without
replica 2");
+ assert_eq!(
+ pending.op_head, 3,
+ "op 4 never committed, so it is truncated"
+ );
+ assert_eq!(pending.commit_max, 2);
+ }
+
+ /// The shard's refresh, gate included: it re-reads the journal only when
the
+ /// snapshot no longer describes it (`refresh_metadata_dvc_suffix`).
+ fn refresh_local_suffix_if_stale(
+ consensus: &VsrConsensus<NoopBus, LocalPipeline>,
+ journal: crate::view_change_quorum::DvcSuffix,
+ ) {
+ if consensus.local_dvc_suffix_stale() {
+ consensus.set_local_dvc_suffix(journal);
+ }
+ }
+
+ /// A backup that adopted a `StartView` sits at the announced head with the
+ /// bodies still missing, so repair filling one moves neither the head nor
the
+ /// commit point. Tag the snapshot by those two alone and it keeps reading
as
+ /// current, which the merge takes as proof this replica never journaled
the op.
+ #[test]
+ fn
given_a_repaired_body_under_an_unmoved_head_when_tagging_should_read_the_snapshot_stale()
{
+ let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus,
LocalPipeline::new());
+ consensus.init();
+ consensus.restore_commit_state(2, 2);
+ consensus.sequencer().set_sequence(4);
+ let headers = suffix_headers(2, 4, 0);
+ consensus
+
.set_local_dvc_suffix(crate::view_change_quorum::DvcSuffix::new(headers, 0,
0b110));
+ assert!(!consensus.local_dvc_suffix_stale());
+
+ consensus.note_journal_mutation();
+
+ assert!(
+ consensus.local_dvc_suffix_stale(),
+ "the head and the commit point did not move, so only the journal
counter \
+ can report the body landing"
+ );
+ assert!(
+ consensus.local_dvc_suffix().is_empty(),
+ "a snapshot that no longer describes the log must nack nothing"
+ );
+ }
+
+ /// The same op, carried through the merge: once the refresh is allowed to
run,
+ /// this replica offers op 4's body and the peer that never prepared it is
one
+ /// nack short of the quorum, so the view starts keeping the op it acked.
+ #[test]
+ fn
given_a_repaired_body_under_an_unmoved_head_when_merging_should_keep_the_acked_op()
{
+ // View 3 of 3 replicas elects this one.
+ let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus,
LocalPipeline::new());
+ consensus.init();
+ consensus.restore_commit_state(2, 2);
+ consensus.sequencer().set_sequence(4);
+ // Adopted the view's header for op 4, body still being repaired.
+ let headers = suffix_headers(2, 4, 0);
+
consensus.set_local_dvc_suffix(crate::view_change_quorum::DvcSuffix::new(
+ headers.clone(),
+ 0,
+ 0b110,
+ ));
+ // Repair journals the body and this replica acks it: a replication
quorum
+ // with the primary, so op 4 is committed even though nobody's commit
point
+ // has caught up yet.
+ consensus.note_journal_mutation();
+ refresh_local_suffix_if_stale(&consensus,
crate::dvc_merge::suffix_all_present(headers));
+
+ let _ = consensus.handle_start_view_change(PlaneKind::Metadata,
&svc_header(1, 3));
+
+ // Replica 1's log stops at op 3, an explicit nack for op 4. Replica 2
is the
+ // crashed primary and never reports.
+ let (dvc, body) = dvc_with_suffix(1, 3, 0, 3, 2, None);
+ let _ = consensus.handle_do_view_change(PlaneKind::Metadata, &dvc,
&body);
+
+ let pending = consensus
+ .pending_view_log()
+ .expect("op 4 is servable from this replica, so the view can
start");
+ assert_eq!(
+ pending.op_head, 4,
+ "one nack cannot discard an op this replica journaled and acked"
+ );
+ }
+
#[test]
fn
given_an_undecidable_quorum_when_a_later_dvc_decides_it_should_start_the_view()
{
// Reaching a view-change quorum is not the same as deciding a log.
Latching
@@ -1666,10 +1811,9 @@ mod tests {
let _ = consensus.handle_start_view_change(PlaneKind::Metadata,
&svc_header(1, 5));
- // Two peers report, reaching the quorum of 3. All three hold op 4's
header,
- // none can serve its body, and two replicas have yet to report.
+ // Two peers abstain, reaching the quorum of 3 and leaving op 4 one
nack short.
for replica in [1u8, 2] {
- let (dvc, body) = dvc_with_suffix(replica, 5, 0, 4, 2, Some(4));
+ let (dvc, body) = dvc_numbers_only(replica, 5, 0, 4, 2);
let actions = consensus.handle_do_view_change(PlaneKind::Metadata,
&dvc, &body);
assert!(actions.is_empty());
}
diff --git a/core/metadata/src/impls/metadata.rs
b/core/metadata/src/impls/metadata.rs
index 347209493..6a74224fd 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -1365,6 +1365,12 @@ where
return;
}
+ // Paired with the append, not with the sequencer advance below: a
backup
+ // repairing under a `StartView` it already adopted is at the
announced head
+ // already, so the entry arriving moves neither number the suffix
snapshot
+ // is otherwise tagged by.
+ consensus.note_journal_mutation();
+
// Journal mutation done; wire traffic below must not hold the gate.
drop(journal_gate);
@@ -1853,10 +1859,10 @@ where
.await
.map_err(SnapshotError::Io)?;
if removed > 0 {
- // The DVC snapshot's `(op, commit)` tag does not move when
- // entries are removed under it; left stale it would advertise
+ // The DVC snapshot's head and commit point do not move when
+ // entries are removed under them; left stale it would
advertise
// headers this replica can no longer serve.
- consensus.invalidate_local_dvc_suffix();
+ consensus.note_journal_mutation();
tracing::warn!(
snapshot_seq,
removed,
@@ -3354,7 +3360,12 @@ where
}
}
- if let Err(e) = coordinator.drain(journal, snap_op).await {
+ let drained = coordinator.drain(journal, snap_op).await;
+ // On the error path too: a drain that fails part-way still removed
+ // whatever it reached, and a snapshot left offering those bodies
strands
+ // the peer that picks this replica as a repair source.
+ consensus.note_journal_mutation();
+ if let Err(e) = drained {
error!(
target: "iggy.metadata.diag",
plane = "metadata",
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 52da8e519..630189f9e 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -5128,12 +5128,7 @@ where
// batches for segment-commit thresholds, which do not
// apply to offset ops.
let frozen = message.into_frozen();
- self.log
- .journal()
- .inner
- .append(frozen.clone())
- .await
- .map_err(|_| IggyError::CannotAppendMessage)?;
+ self.journal_append(frozen.clone()).await?;
match header.operation {
Operation::StoreConsumerOffset => {
@@ -5180,6 +5175,21 @@ where
}
}
+ /// Journal one prepare and record the mutation.
+ ///
+ /// Every partition-plane append goes through here. The DVC suffix
snapshot is
+ /// tagged by the head and the commit point, and a repair filling a body
under
+ /// a `StartView` this replica already adopted moves neither: the head is
the
+ /// announced one and the op is not committed yet. An append that skipped
the
+ /// counter would leave the header-only snapshot reading as current, and
the
+ /// merge takes a header no sender can serve as proof that sender never
+ /// journaled the op.
+ async fn journal_append(&self, entry: Frozen<4096>) -> Result<(),
IggyError> {
+ let appended = self.log.journal().inner.append(entry).await;
+ self.consensus.note_journal_mutation();
+ appended.map_err(|_| IggyError::CannotAppendMessage)
+ }
+
async fn append_send_messages_to_journal(
&mut self,
message: Message<PrepareHeader>,
@@ -5278,12 +5288,7 @@ where
journal_info.max_timestamp =
journal_info.max_timestamp.max(batch.base_timestamp);
let frozen = message.into_frozen();
- self.log
- .journal()
- .inner
- .append(frozen.clone())
- .await
- .map_err(|_| IggyError::CannotAppendMessage)?;
+ self.journal_append(frozen.clone()).await?;
self.note_append_live();
self.dirty_offset
@@ -5394,7 +5399,7 @@ where
self.offset_space.committed_seeded = false;
}
}
- self.consensus.invalidate_local_dvc_suffix();
+ self.consensus.note_journal_mutation();
let commit_max = self.consensus.commit_max();
self.pending_consumer_offset_commits
.retain(|op, _| *op < from_op || *op <= commit_max);
@@ -5785,6 +5790,10 @@ where
return;
}
let retained = self.log.journal().inner.evict_prefix(count).await;
+ // Eviction drains the ring and re-appends the retained tail, so the
slots
+ // a suffix snapshot describes move under it without the head or the
+ // commit point changing.
+ self.consensus.note_journal_mutation();
let mut retained_info = JournalInfo::default();
for (entry, meta) in &retained {
// Purge floor: a retained pre-purge batch must not fold its
@@ -7984,12 +7993,7 @@ where
// `first_batch_offset`: that anchors the floor-connect check, and
// purged bytes cannot stand in for durable state.
if op <= self.purge_floor_op {
- self.log
- .journal()
- .inner
- .append(message.into_frozen())
- .await
- .map_err(|_| IggyError::CannotAppendMessage)?;
+ self.journal_append(message.into_frozen()).await?;
return Ok(None);
}
@@ -8024,12 +8028,7 @@ where
journal_info.max_timestamp =
journal_info.max_timestamp.max(base_timestamp);
let frozen = message.into_frozen();
- self.log
- .journal()
- .inner
- .append(frozen)
- .await
- .map_err(|_| IggyError::CannotAppendMessage)?;
+ self.journal_append(frozen).await?;
self.note_append_live();
self.dirty_offset
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index e492a8fb3..ede938135 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -5157,6 +5157,14 @@ where
);
return;
}
+ // The body landing is invisible to the head and the commit point:
a
+ // backup repairing under a `StartView` it already adopted sits at
the
+ // announced head with its commit point unmoved. Leave the suffix
+ // snapshot tagged as current and the next `DoViewChange` reports
this
+ // op header-only, which the merge reads as proof this replica
never
+ // journaled it, one nack away from truncating an op it is about to
+ // acknowledge.
+ consensus.note_journal_mutation();
// Contiguous-frontier advance, mirroring
// `apply_repaired_prepare`: DVC advertises the sequencer, so a
// hole below a repaired op must stall the advance rather than
@@ -6091,10 +6099,11 @@ where
// mutations through gate-taking metadata methods would make it
structural.
match journal.handle().truncate_from(from_op).await {
Ok(removed) => {
- // The snapshot's `(op, commit)` tag does not move when
entries are
- // removed under it, so the next `DoViewChange` would
advertise the
- // dropped headers and offer bodies this replica cannot serve.
- consensus.invalidate_local_dvc_suffix();
+ // The snapshot's head and commit point do not move when
entries
+ // are removed under them, so without this the next
`DoViewChange`
+ // would advertise the dropped headers and offer bodies this
replica
+ // cannot serve.
+ consensus.note_journal_mutation();
tracing::warn!(
shard = self.id,
from_op,
@@ -9899,10 +9908,10 @@ where
{
match journal.handle().truncate_from(stuck_op).await {
Ok(removed) => {
- // The snapshot's `(op, commit)` tag does not move when entries
- // are removed under it, so the next `DoViewChange` would
- // otherwise advertise headers this replica can no longer
serve.
- consensus.invalidate_local_dvc_suffix();
+ // The snapshot's head and commit point do not move when
entries
+ // are removed under them, so without this the next
`DoViewChange`
+ // would advertise headers this replica can no longer serve.
+ consensus.note_journal_mutation();
tracing::warn!(
shard = self.id,
stuck_op,
@@ -10309,9 +10318,9 @@ fn rebuild_pipeline_entries<B, P>(
///
/// Called before every handler that could start or join a view change:
consensus
/// records its own `DoViewChange` there and has no journal to read. A stale
-/// snapshot is never reused; consensus tags it with its `(op, commit)` and
falls
-/// back to an empty suffix, stalling the view change rather than nacking an op
-/// since acquired.
+/// snapshot is never reused; consensus tags it with the journal's head, commit
+/// point and mutation count, and falls back to an empty suffix, stalling the
view
+/// change rather than nacking an op since acquired.
fn refresh_metadata_dvc_suffix<B, P, MJ>(consensus: &VsrConsensus<B, P>,
journal: Option<&MJ>)
where
B: MessageBus,
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 0b25090dc..747e5b70b 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -4536,6 +4536,84 @@ mod tests {
/// finds two replicas at the same op passes in silence, so `ops_compared`
counts
/// only ops witnessed on more than one replica, the subset that exercised
the
/// property.
+ /// A replica prepares an op and stays down. One survivor holds its header
+ /// without the body, the other never had it, and the merge must read that
as
+ /// proof both are outside the ack set. Otherwise it waits for the crashed
+ /// replica: on this seed the cluster reached view 103 against `log_view`
3 with
+ /// both survivors caught up and one request retried 266 times unanswered.
+ ///
+ /// Asserts the drain, so it fails as the fuzzer does with the
outstanding-request
+ /// report attached.
+ #[test]
+ fn view_change_completes_without_the_replica_that_prepared_the_head() {
+ use crate::workload::{
+ self, FaultInjector, Workload,
+ options::{ActionWeights, WorkloadOptions},
+ oracle,
+ };
+
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;
+ // `workload-fuzz --seed 211 --replicas 3 --ticks 4000 --plane metadata
+ // --journal-slots 80 --crash-prob 0.02 --restart-prob 0.08
--crash-primary`.
+ let seed = 211u64;
+ let root = tempfile::tempdir().expect("temp dir for the simulator's
snapshots");
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 1,
+ seed,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ // A bounded journal is what drives the WAL drain behind the
header-only sender.
+ let mut sim = Simulator::with_checkpoints(
+ usize::from(replica_count),
+ std::iter::once(client_id),
+ network_opts,
+ false,
+ root.path(),
+ );
+ sim.set_metadata_journal_slots(80);
+
+ let ns = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(ns);
+ let client = SimClient::new(client_id);
+ sim.register_client_with_primary(&client);
+
+ let mut options = WorkloadOptions::new(seed, replica_count, vec![ns]);
+ options.client_count = 1;
+ options.crash_per_tick_ratio = 0.02;
+ options.restart_per_tick_ratio = 0.08;
+ options.spare_primary = false;
+ options.weights = ActionWeights::metadata_only();
+ let mut workload = Workload::new(options);
+
+ let clients = [client];
+ let mut injector = FaultInjector::new(seed, replica_count);
+ let _ = workload::run_with_faults(
+ &mut sim,
+ &mut workload,
+ &clients,
+ 4_000,
+ u64::MAX,
+ &mut injector,
+ );
+
+ assert!(
+ injector.crashes() > 0,
+ "no replica crashed, so no view change ran and this proves nothing"
+ );
+ assert!(
+ oracle::drive_to_quiesce(&mut sim, &mut workload, 50_000),
+ "{}",
+ oracle::quiesce_failure_report(&sim, &workload),
+ );
+ }
+
#[test]
fn committed_metadata_agrees_across_replicas() {
use crate::workload::{
