This is an automated email from the ASF dual-hosted git repository.
hubcio 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 a2a4a2cb9 fix(consensus): keep a replica off a hole in its committed
prefix (#4073)
a2a4a2cb9 is described below
commit a2a4a2cb987b7c92d232c073c2138ce1b2237251
Author: Krishna Vishal <[email protected]>
AuthorDate: Wed Sep 9 03:16:18 2026 +0530
fix(consensus): keep a replica off a hole in its committed prefix (#4073)
A replica could be promoted to primary while missing operations the
cluster had already committed. The promotion coverage scan opened at
the merged commit point and never looked below it, so a hole under
that point went unseen. The replica then served reads from incomplete
state and walked commit_min over it, failing a swallowed assert.
The scan, its repair requests and its retries now open at the lower
of the merged commit point and commit_min + 1, and the scan reads the
evicted repair ring, so eviction no longer looks like a gap. Repair
sessions clear when satisfied, reopen where they were armed, and
rotate their sources as a ring. Commit advances only over a contiguous
run from commit_min + 1, and both journal walks stop below the
pipeline head, the only entry that can answer the caller. A view
change re-decides the recovery barrier it can truncate, so no primary
waits out operations nobody will re-prepare.
The simulator compares partition journal contents, not only commit
positions, and fails a run on a wedged barrier, a commit hold that
never clears, or a fenced partition task.
---
core/consensus/src/impls.rs | 440 +++++++++---
core/consensus/src/plane_helpers.rs | 169 ++++-
core/metadata/src/impls/metadata.rs | 18 +
core/partitions/src/iggy_partition.rs | 150 +++-
core/server/src/http/reads.rs | 10 +-
core/shard/src/lib.rs | 983 +++++++++++++++++++++++----
core/shard/src/router.rs | 9 +
core/simulator/src/bin/workload-fuzz.rs | 44 +-
core/simulator/src/lib.rs | 138 +++-
core/simulator/src/workload/invariants.rs | 164 ++++-
core/simulator/src/workload/oracle.rs | 14 +
core/simulator/src/workload/state_checker.rs | 170 ++++-
12 files changed, 2034 insertions(+), 275 deletions(-)
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 159e0ca5e..0d1ebce94 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -1729,6 +1729,41 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
self.recovery_barrier.set(required_commit);
}
+ /// Re-decide the barrier against a log head the cluster just settled.
+ ///
+ /// Boot arms it at the recovered journal head: those ops were acked
before the
+ /// restart, so admitting writes before they re-commit rolls back committed
+ /// history. It otherwise clears only by `commit_max` passing it, which
never
+ /// happens when a view change discards the suffix instead of
re-committing it.
+ /// The boot re-pipeline already ran, so nothing re-prepares those ops,
+ /// `is_caught_up_primary` stays shut, and the primary drops the very
requests
+ /// that would raise `commit_max`.
+ ///
+ /// Call this wherever the head is authoritatively re-decided: a merged
log at
+ /// view start, an adopted `StartView`. `head` lowers the barrier when the
view
+ /// truncated the suffix, keeps it when the suffix survived.
+ ///
+ /// Lowered, never cleared. `is_caught_up_primary` reads it against
`commit_max`
+ /// and a met barrier costs it nothing, but `await_recovery_barrier` reads
it
+ /// against `commit_min`, and adoption raises `commit_max` before walking
the
+ /// suffix into the state machine. Zeroing a met barrier would open that
read
+ /// gate over the unapplied window it exists to hold.
+ ///
+ /// `head == 0` returns instead of lowering: zero is the DISARMED value,
not a
+ /// met barrier. `barrier_state` (`server/src/http/reads.rs`) reads zero as
+ /// "nothing gated" and skips the `commit_min` comparison, so writing it
would
+ /// open the gate rather than lower it. Reachable on the wire:
+ /// `adopt_start_view_suffix` returns `StartViewHeader.op` verbatim when
the
+ /// suffix fails to decode, `validate` does not require `op >= 1`, and the
+ /// low-op guard binds only while `msg_view == log_view`.
+ fn redecide_recovery_barrier(&self, head: u64) {
+ let barrier = self.recovery_barrier.get();
+ if barrier == 0 || head == 0 {
+ return;
+ }
+ self.recovery_barrier.set(barrier.min(head));
+ }
+
/// Deadline paired with [`Self::recovery_barrier`]; only meaningful while
the
/// barrier is armed (non-zero).
#[must_use]
@@ -3363,6 +3398,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
// frame, and either value leaves this replica chasing an unservable
head.
let announced = self.adopt_start_view_suffix(header, suffix_body);
self.sequencer.set_sequence(announced);
+ // Settle a gated suffix's fate as a backup too, so a later election
inherits
+ // a decided barrier rather than a latched one.
+ self.redecide_recovery_barrier(announced);
// Update timeouts for normal backup operation
{
@@ -3654,20 +3692,46 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
/// and the view change is blocked on the round-trip.
#[must_use]
pub fn pending_view_body_sources(&self, op: u64) -> Vec<u8> {
+ self.pending_view_sources(|dvc| {
+ dvc.suffix
+ .index_of(dvc.op, op)
+ .is_some_and(|index| dvc.suffix.offers_body(index))
+ })
+ }
+
+ /// Recorded `DoViewChange` senders that `serves` accepts,
most-recent-`log_view`
+ /// first, never this replica.
+ ///
+ /// Ordering and self-exclusion are the same for every source list; only
the
+ /// "can this sender serve the op" predicate differs.
+ fn pending_view_sources(&self, serves: impl Fn(&StoredDvc) -> bool) ->
Vec<u8> {
let quorum = self.do_view_change_from_all_replicas.borrow();
let mut sources: Vec<(u32, u8)> = dvc_iter(&quorum)
.filter(|dvc| dvc.replica != self.replica)
- .filter_map(|dvc| {
- let index = dvc.suffix.index_of(dvc.op, op)?;
- dvc.suffix
- .offers_body(index)
- .then_some((dvc.log_view, dvc.replica))
- })
+ .filter(|dvc| serves(dvc))
+ .map(|dvc| (dvc.log_view, dvc.replica))
.collect();
sources.sort_unstable_by_key(|(log_view, _)|
std::cmp::Reverse(*log_view));
sources.into_iter().map(|(_, replica)| replica).collect()
}
+ /// Replicas that committed `op`, most-recent-`log_view` first.
+ ///
+ /// The fallback for an op below the DVC suffixes. A suffix spans
`commit..=op`,
+ /// so [`Self::pending_view_body_sources`] answers nothing about the
committed
+ /// prefix and a merged log whose coverage gap sits there would have no
source
+ /// at all. A sender that committed the op either still journals it or has
+ /// compacted it under a checkpoint, and both answers move the requester
+ /// forward: the prepare, or the `RangeEvicted` that says repair cannot
close
+ /// this gap.
+ ///
+ /// Presence is not proven the way an offered body is, so prefer
+ /// [`Self::pending_view_body_sources`] wherever it returns anything.
+ #[must_use]
+ pub fn pending_view_commit_sources(&self, op: u64) -> Vec<u8> {
+ self.pending_view_sources(|dvc| dvc.commit >= op)
+ }
+
/// Finish the parked view change: this replica's journal now covers the
merged
/// log, so it can serve any op it is about to announce.
///
@@ -3703,6 +3767,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
self.status.set(Status::Normal);
self.ceded_primaryship.set(false);
self.sequencer.set_sequence(new_op);
+ // Only place a promotion learns what the merge did to a gated suffix.
+ self.redecide_recovery_barrier(new_op);
if let Some(head) = merged.headers.first() {
// Keep the hash chain continuous: the next prepare must chain
onto the
// head this view adopted, not onto whatever was appended last.
@@ -4418,31 +4484,48 @@ mod pipeline_entry_tests {
}
}
+/// Fixtures every consensus test module needs.
#[cfg(test)]
-mod timestamp_clamp_tests {
- //! Pin the monotonic-floor contract: a new primary must never stamp a
- //! prepare below timestamps already in the replicated log, even when its
- //! wall clock lags the predecessor's.
-
- use super::*;
- use crate::LocalPipeline;
- use message_bus::BusMessage;
+pub mod test_bus {
+ use super::{Command, METADATA_GROUP, Message, StartViewHeader};
+ use message_bus::{BusMessage, MessageBus};
use server_common::MESSAGE_ALIGN;
use server_common::iobuf::Frozen;
- /// Clock frozen at a fixed instant, standing in for a lagging wall
- /// clock on a freshly elected primary.
- struct FixedClock(u64);
-
- impl clock::Clock for FixedClock {
- type Realtime = IggyTimestamp;
-
- fn realtime(&self) -> Self::Realtime {
- IggyTimestamp::from(self.0)
- }
+ /// A `StartView` at `view` announcing head `op`. `commit == op` is the
steady
+ /// case (no suffix); a lower `commit` keeps an uncommitted suffix.
+ ///
+ /// # Panics
+ /// Never: a zeroed buffer of the right size is a valid `StartViewHeader`.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn make_start_view(
+ view: u32,
+ op: u64,
+ commit: u64,
+ replica: u8,
+ incarnation: u128,
+ ) -> Message<StartViewHeader> {
+ let size = std::mem::size_of::<StartViewHeader>();
+ let mut msg = Message::<StartViewHeader>::new(size);
+ let header = bytemuck::checked::try_from_bytes_mut::<StartViewHeader>(
+ &mut msg.as_mut_slice()[..size],
+ )
+ .expect("zeroed bytes are a valid StartViewHeader");
+ header.command = Command::StartView;
+ header.cluster = 1;
+ header.view = view;
+ header.op = op;
+ header.commit = commit;
+ header.replica = replica;
+ header.incarnation = incarnation;
+ header.group = METADATA_GROUP;
+ header.size = size as u32;
+ msg
}
- struct NoopBus;
+ /// A [`MessageBus`] that accepts everything and remembers nothing.
+ pub struct NoopBus;
impl MessageBus for NoopBus {
async fn send_to_client(
@@ -4466,6 +4549,30 @@ mod timestamp_clamp_tests {
fn set_client_forward_fn(&self, _f: message_bus::ClientForwardFn) {}
fn track_background(&self, _handle: message_bus::JoinHandle<()>) {}
}
+}
+
+#[cfg(test)]
+mod timestamp_clamp_tests {
+ //! Pin the monotonic-floor contract: a new primary must never stamp a
+ //! prepare below timestamps already in the replicated log, even when its
+ //! wall clock lags the predecessor's.
+
+ use super::*;
+ use crate::LocalPipeline;
+
+ /// Clock frozen at a fixed instant, standing in for a lagging wall
+ /// clock on a freshly elected primary.
+ struct FixedClock(u64);
+
+ impl clock::Clock for FixedClock {
+ type Realtime = IggyTimestamp;
+
+ fn realtime(&self) -> Self::Realtime {
+ IggyTimestamp::from(self.0)
+ }
+ }
+
+ use crate::test_bus::{NoopBus, make_start_view};
#[test]
fn observed_log_timestamp_floors_new_primary_stamps() {
@@ -4519,31 +4626,6 @@ mod timestamp_clamp_tests {
);
}
- #[allow(clippy::cast_possible_truncation)]
- fn make_start_view(
- view: u32,
- op: u64,
- replica: u8,
- incarnation: u128,
- ) -> Message<StartViewHeader> {
- let size = std::mem::size_of::<StartViewHeader>();
- let mut msg = Message::<StartViewHeader>::new(size);
- let header = bytemuck::checked::try_from_bytes_mut::<StartViewHeader>(
- &mut msg.as_mut_slice()[..size],
- )
- .expect("zeroed bytes are a valid StartViewHeader");
- header.command = Command::StartView;
- header.cluster = 1;
- header.view = view;
- header.op = op;
- header.commit = op;
- header.replica = replica;
- header.incarnation = incarnation;
- header.group = METADATA_GROUP;
- header.size = size as u32;
- msg
- }
-
#[test]
fn
given_recovering_replica_when_start_view_incarnation_foreign_should_ignore() {
// A StartView addressed to a PREVIOUS incarnation, still in flight
when the
@@ -4571,7 +4653,7 @@ mod timestamp_clamp_tests {
assert_eq!(consensus.status(), Status::Recovering);
// Same view, head behind ours, foreign incarnation: ignored.
- let stale = make_start_view(1, 4, 1, STALE);
+ let stale = make_start_view(1, 4, 4, 1, STALE);
assert!(
consensus
.handle_start_view(PlaneKind::Metadata, stale.header(), &[])
@@ -4591,7 +4673,7 @@ mod timestamp_clamp_tests {
// Same view and head but echoing our current incarnation: adopted,
since
// the match proves the reply post-dates our restart.
- let fresh = make_start_view(1, 4, 1, CURRENT);
+ let fresh = make_start_view(1, 4, 4, 1, CURRENT);
assert!(
!consensus
.handle_start_view(PlaneKind::Metadata, fresh.header(), &[])
@@ -4688,7 +4770,7 @@ mod timestamp_clamp_tests {
consensus
.handle_start_view(
PlaneKind::Metadata,
- make_start_view(7, 104, 1, 0).header(),
+ make_start_view(7, 104, 104, 1, 0).header(),
&[]
)
.is_empty(),
@@ -4706,7 +4788,7 @@ mod timestamp_clamp_tests {
!consensus
.handle_start_view(
PlaneKind::Metadata,
- make_start_view(7, 105, 1, 0).header(),
+ make_start_view(7, 105, 105, 1, 0).header(),
&[]
)
.is_empty(),
@@ -5075,6 +5157,8 @@ mod vsr_consensus_tests {
// now measuring op 2 rather than carrying op 1's elapsed ticks.
consensus.advance_commit_max(1);
assert_eq!(drain_committable_prefix(&consensus).len(), 1);
+ // As real callers do, per entry: the next drain starts at the op now
owed.
+ consensus.advance_commit_min(1);
assert!(
prepare_ticking(&consensus),
"a remaining prepare keeps the timer armed"
@@ -5082,6 +5166,7 @@ mod vsr_consensus_tests {
consensus.advance_commit_max(2);
assert_eq!(drain_committable_prefix(&consensus).len(), 1);
+ consensus.advance_commit_min(2);
assert!(
!prepare_ticking(&consensus),
"draining the last prepare disarms the timer without waiting for
it to fire"
@@ -5335,34 +5420,8 @@ mod quorum_tests {
use super::*;
use crate::LocalPipeline;
- use message_bus::BusMessage;
- use server_common::MESSAGE_ALIGN;
- use server_common::iobuf::Frozen;
- struct NoopBus;
-
- impl MessageBus for NoopBus {
- async fn send_to_client(
- &self,
- _client_id: u128,
- _data: impl Into<BusMessage>,
- ) -> Result<(), message_bus::SendError> {
- Ok(())
- }
-
- async fn send_to_replica(
- &self,
- _replica: u8,
- _data: Frozen<MESSAGE_ALIGN>,
- ) -> Result<(), message_bus::SendError> {
- Ok(())
- }
-
- fn set_connection_lost_fn(&self, _f: message_bus::ConnectionLostFn) {}
- fn set_replica_forward_fn(&self, _f: message_bus::ReplicaForwardFn) {}
- fn set_client_forward_fn(&self, _f: message_bus::ClientForwardFn) {}
- fn track_background(&self, _handle: message_bus::JoinHandle<()>) {}
- }
+ use crate::test_bus::NoopBus;
fn consensus_with_replica_count(replica_count: u8) ->
VsrConsensus<NoopBus, LocalPipeline> {
VsrConsensus::new(
@@ -5404,3 +5463,222 @@ mod quorum_tests {
}
};
}
+
+#[cfg(test)]
+mod view_source_tests {
+ //! Who a primary-elect may ask for an op its merged log names. A
`DoViewChange`
+ //! suffix spans `commit..=op`, so the two selectors cover disjoint halves
of
+ //! the merged log and the split is what keeps a coverage gap under the
merged
+ //! commit point askable at all.
+
+ use super::*;
+ use crate::LocalPipeline;
+ use crate::view_change_quorum::{DvcSuffix, StoredDvc, dvc_record};
+
+ use crate::test_bus::NoopBus;
+
+ fn consensus() -> VsrConsensus<NoopBus, LocalPipeline> {
+ VsrConsensus::new(1, 0, 3, METADATA_GROUP, NoopBus,
LocalPipeline::new())
+ }
+
+ /// A sender whose suffix runs `commit..=op` with every body offered,
which is
+ /// the widest window a real `DoViewChange` can carry.
+ fn sender(replica: u8, log_view: u32, op: u64, commit: u64) -> StoredDvc {
+ let headers: Vec<PrepareHeader> = (commit..=op)
+ .rev()
+ .map(|op| PrepareHeader {
+ command: Command::Prepare,
+ op,
+ view: log_view,
+ ..Default::default()
+ })
+ .collect();
+ let present = (1u128 << headers.len()) - 1;
+ StoredDvc {
+ replica,
+ log_view,
+ op,
+ commit,
+ suffix: DvcSuffix::new(headers, 0, present),
+ }
+ }
+
+ fn record(consensus: &VsrConsensus<NoopBus, LocalPipeline>, senders:
[StoredDvc; 2]) {
+ let mut quorum =
consensus.do_view_change_from_all_replicas.borrow_mut();
+ for dvc in senders {
+ assert!(dvc_record(&mut quorum, dvc));
+ }
+ }
+
+ #[test]
+ fn
given_an_op_inside_the_suffixes_when_selecting_should_return_the_body_offers() {
+ let consensus = consensus();
+ record(&consensus, [sender(1, 5, 12, 10), sender(2, 4, 12, 10)]);
+
+ assert_eq!(
+ consensus.pending_view_body_sources(11),
+ vec![1, 2],
+ "both senders offer op 11, freshest log_view first"
+ );
+ }
+
+ #[test]
+ fn
given_an_op_below_every_commit_point_when_selecting_should_need_the_committers()
{
+ let consensus = consensus();
+ record(&consensus, [sender(1, 5, 12, 10), sender(2, 4, 12, 10)]);
+
+ assert!(
+ consensus.pending_view_body_sources(7).is_empty(),
+ "a suffix spans commit..=op, so it says nothing about op 7"
+ );
+ assert_eq!(
+ consensus.pending_view_commit_sources(7),
+ vec![1, 2],
+ "a sender that committed op 7 holds it or compacted it, and either
\
+ answer moves the requester forward"
+ );
+ }
+
+ #[test]
+ fn given_a_sender_behind_the_op_when_selecting_committers_should_skip_it()
{
+ let consensus = consensus();
+ record(&consensus, [sender(1, 5, 12, 10), sender(2, 6, 6, 5)]);
+
+ assert_eq!(
+ consensus.pending_view_commit_sources(7),
+ vec![1],
+ "replica 2 never committed op 7, so asking it wastes a retry
interval \
+ on a RangeEvicted it has no standing to send"
+ );
+ }
+
+ #[test]
+ fn
given_this_replica_in_the_quorum_when_selecting_should_never_return_self() {
+ let consensus = consensus();
+ record(&consensus, [sender(0, 5, 12, 10), sender(2, 4, 12, 10)]);
+
+ assert_eq!(
+ consensus.pending_view_commit_sources(7),
+ vec![2],
+ "a replica cannot repair from itself"
+ );
+ }
+}
+
+#[cfg(test)]
+mod recovery_barrier_tests {
+ //! The gate holding a restarted replica's reads and writes until the
recovered
+ //! WAL suffix re-commits. Boot arms it at the recovered head; only a view
that
+ //! settles that suffix's fate may move it, and only downward.
+
+ use super::*;
+ use crate::LocalPipeline;
+ use crate::test_bus::{NoopBus, make_start_view};
+
+ /// Recovered at head 120, proven committed only through 100, in view 7.
+ fn recovered_with_gated_suffix() -> VsrConsensus<NoopBus, LocalPipeline> {
+ let mut consensus =
+ VsrConsensus::new(1, 0, 3, METADATA_GROUP, NoopBus,
LocalPipeline::new());
+ consensus.set_view(7);
+ consensus.set_log_view(7);
+ consensus.sequencer().set_sequence(120);
+ consensus.restore_commit_state(100, 100);
+ consensus.set_recovery_barrier(120);
+ consensus
+ }
+
+ /// The `commit_max >= recovery_barrier` clause of `is_caught_up_primary`,
read
+ /// directly so these tests need not satisfy the primary/status clauses.
+ fn write_gate_open<B: MessageBus, P>(consensus: &VsrConsensus<B, P>) ->
bool
+ where
+ P: Pipeline<Entry = PipelineEntry>,
+ {
+ consensus.commit_max() >= consensus.recovery_barrier()
+ }
+
+ /// The wedge `redecide_recovery_barrier` exists for: a view change
discards the
+ /// recovered suffix, the replica later wins an election, and a barrier
pinned
+ /// to a head that no longer exists shuts admission.
+ #[test]
+ fn
given_a_discarded_suffix_when_adopting_a_view_should_lower_the_barrier() {
+ let consensus = recovered_with_gated_suffix();
+
+ // The view's head is 105: 106..=120 committed nowhere, so the view
drops
+ // them and nothing re-prepares them.
+ assert!(
+ !consensus
+ .handle_start_view(
+ PlaneKind::Metadata,
+ make_start_view(7, 105, 105, 1, 0).header(),
+ &[]
+ )
+ .is_empty(),
+ "the StartView at the commit floor must be adopted"
+ );
+ assert_eq!(
+ consensus.recovery_barrier(),
+ 105,
+ "the adopted head settled the suffix's fate, so the barrier must
fall to \
+ it rather than latch at a head the view discarded"
+ );
+ assert!(
+ write_gate_open(&consensus),
+ "a barrier the adopted commit point covers must stop gating
admission"
+ );
+ assert!(
+ consensus.commit_min() < consensus.recovery_barrier(),
+ "adoption raises commit_max before applying, so the local read
gate \
+ (which reads commit_min) must still hold"
+ );
+ }
+
+ /// The other half: lowering the barrier to the adopted head must not read
as
+ /// clearing it while the suffix survives.
+ #[test]
+ fn given_a_surviving_suffix_when_adopting_a_view_should_keep_the_barrier()
{
+ let consensus = recovered_with_gated_suffix();
+
+ // Head 120, commit still 100: 101..=120 re-replicate under the new
view.
+ assert!(
+ !consensus
+ .handle_start_view(
+ PlaneKind::Metadata,
+ make_start_view(7, 120, 100, 1, 0).header(),
+ &[]
+ )
+ .is_empty(),
+ "the StartView carrying the surviving suffix must be adopted"
+ );
+ assert_eq!(
+ consensus.recovery_barrier(),
+ 120,
+ "a suffix the view kept is still unproven, so its gate must stand"
+ );
+ assert!(
+ !write_gate_open(&consensus),
+ "an unproven suffix must keep admission shut"
+ );
+ }
+
+ #[test]
+ fn given_no_recovered_suffix_when_redeciding_should_stay_disarmed() {
+ // Nothing gated, so no view change may invent a gate.
+ let consensus = VsrConsensus::new(1, 0, 3, METADATA_GROUP, NoopBus,
LocalPipeline::new());
+ consensus.redecide_recovery_barrier(9);
+ assert_eq!(consensus.recovery_barrier(), 0);
+ }
+
+ /// Zero is disarmed, not met: `barrier_state` skips the `commit_min`
comparison
+ /// on it, so writing it opens the HTTP read gate over the window the
barrier
+ /// holds.
+ #[test]
+ fn given_a_zero_head_when_redeciding_should_leave_the_barrier_armed() {
+ let consensus = recovered_with_gated_suffix();
+ consensus.redecide_recovery_barrier(0);
+ assert_eq!(
+ consensus.recovery_barrier(),
+ 120,
+ "a zero head must not disarm the gate"
+ );
+ }
+}
diff --git a/core/consensus/src/plane_helpers.rs
b/core/consensus/src/plane_helpers.rs
index 195120af8..a72995cf0 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -433,10 +433,48 @@ where
false
}
+/// Whether a repair session may still run.
+///
+/// `Normal` is the ordinary case. The exception is a primary-elect that
parked a
+/// merged log: its repair runs in `ViewChange` by design, because the log it
is
+/// repairing toward is one nobody has started yet. Reading `is_normal` alone
at an
+/// ingest site drops the frame AND the session, so the coverage scan re-arms
every
+/// tick and nothing it fetches can ever land.
+///
+/// Every site that fences a repair session on status must use this, so the
arming
+/// side and the ingest side cannot drift apart.
+pub fn repair_session_live<B, P>(consensus: &VsrConsensus<B, P>) -> bool
+where
+ B: MessageBus,
+ P: Pipeline<Entry = PipelineEntry>,
+{
+ consensus.is_normal()
+ || (consensus.view_log_is_pending() &&
consensus.is_primary_for_view(consensus.view()))
+}
+
/// Drain and return committable prepares from the pipeline head.
///
-/// Entries are drained only from the head and only while their op is covered
-/// by the current commit frontier.
+/// Entries are drained from the head, while covered by the commit frontier,
and
+/// only as a contiguous run starting at the next op owed to the state machine.
+///
+/// Callers `advance_commit_min` per entry, so a run starting above
+/// `commit_min + 1` or breaking partway hits that counter's sequential-advance
+/// assert. Pipeline-side twin of `commit_journal`'s gap-stop.
+///
+/// Reported, not asserted: a promoted partition primary reaches it
legitimately
+/// while its journal walk clears the apply backlog. Repair arms on the level,
and
+/// a hold that never clears fails the simulator's contiguity invariant.
+///
+/// Do NOT read that as "the journal walk always finishes". It can stall on the
+/// partition plane: `committed_headers_from` reads resident headers only,
while
+/// `commit_messages` evicts up to `commit_max` (the cluster frontier), so ops
past
+/// the per-call cap can be flushed out from under the walk. Pre-existing; see
+/// `IggyPartition::collect_committable_from_journal`.
+///
+/// A head at or below `commit_min` is the other shape: applied already, no
repair
+/// owed, only a pop that can no longer happen. The journal walks stop below
the
+/// head so they cannot create it, but a `set_commit_floor` jump past a live
+/// pipeline still can, so it is logged apart rather than treated as
impossible.
///
/// # Panics
/// If `head()` returns `Some` but `pop()` returns `None` (unreachable).
@@ -446,18 +484,31 @@ where
P: Pipeline<Entry = PipelineEntry>,
{
let commit = consensus.commit_max();
+ let commit_min = consensus.commit_min();
let mut drained = Vec::new();
consensus.with_pipeline_mut(|pipeline| {
+ let mut next = commit_min + 1;
while let Some(head_op) = pipeline.head().map(|entry| entry.header.op)
{
if head_op > commit {
break;
}
+ if head_op != next {
+ report_uncommittable_head(
+ consensus.replica(),
+ head_op,
+ commit_min,
+ commit,
+ drained.len(),
+ );
+ break;
+ }
let entry = pipeline
.pop()
.expect("drain_committable_prefix: head exists");
drained.push(entry);
+ next += 1;
}
});
@@ -473,7 +524,8 @@ where
drained
}
-/// Header of the pipeline head, iff its op is covered by the commit frontier.
+/// Header of the pipeline head, iff its op is the next one this replica owes
its
+/// state machine and is covered by the commit frontier.
///
/// Peek-only counterpart of [`drain_committable_prefix`] for commit paths that
/// must survive their driving future being canceled between "committable" and
@@ -482,15 +534,68 @@ where
/// revalidates that the head is still this exact entry before popping and
/// applying it. A driver dropped at an await strands nothing; a sibling driver
/// that committed the op first fails the caller's revalidation and re-peeks.
+///
+/// Bounded below for the reason [`drain_committable_prefix`] is, and reported
+/// rather than asserted for the same one. Holding is safe: a shard pump's
panic is
+/// swallowed by `compio::runtime::spawn`, while `tick_metadata` re-arms
repair on
+/// the level.
pub fn peek_committable_head<B, P>(consensus: &VsrConsensus<B, P>) ->
Option<PrepareHeader>
where
B: MessageBus,
P: Pipeline<Entry = PipelineEntry>,
{
let commit = consensus.commit_max();
- consensus
+ let commit_min = consensus.commit_min();
+ let head = consensus
.pipeline_head_header()
- .filter(|header| header.op <= commit)
+ .filter(|header| header.op <= commit)?;
+ if head.op != commit_min + 1 {
+ report_uncommittable_head(consensus.replica(), head.op, commit_min,
commit, 0);
+ return None;
+ }
+ Some(head)
+}
+
+/// Log a held pipeline head, with the remedy for the arm it is in.
+///
+/// `debug`, not `warn`, matching `tick_partitions` / `tick_metadata`: a hold
is the
+/// steady state for a whole rejoin, so `warn` is one line per group per tick.
A
+/// hold that never clears is caught by a simulator invariant, not by this
line.
+fn report_uncommittable_head(
+ replica: u8,
+ head_op: u64,
+ commit_min: u64,
+ commit_max: u64,
+ drained: usize,
+) {
+ if head_op > commit_min {
+ tracing::debug!(
+ replica,
+ head_op,
+ expected_op = commit_min + 1,
+ commit_min,
+ commit_max,
+ drained,
+ "committable head sits above a hole in the committed prefix;
holding the commit \
+ walk until the ops below it are journaled"
+ );
+ } else {
+ // Reported, not called a defect: a commit-floor jump reaches this
state
+ // legitimately. `IggyMetadata`'s state-transfer install and
+ // `IggyPartition::complete_repair` both raise `commit_min` past a live
+ // pipeline via `set_commit_floor` without the `clear_pipeline` the
+ // partition transfer path pairs with it. Nothing pops a head below the
+ // floor, so it holds until a view change clears the pipeline.
+ tracing::warn!(
+ replica,
+ head_op,
+ commit_min,
+ commit_max,
+ drained,
+ "committable head sits at or below the applied commit point;
nothing can pop \
+ or answer this entry until a view change clears the pipeline"
+ );
+ }
}
/// Build reply for a committed prepare.
@@ -1991,10 +2096,64 @@ mod tests {
assert_eq!(sent[0].0, 1);
}
+ /// `advance_commit_min` is strictly sequential, so draining op 7 with 6
never
+ /// applied panics the shard pump. Both gates must hold instead.
+ #[test]
+ fn given_a_hole_below_the_head_when_committing_should_hold_both_gates() {
+ let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus,
LocalPipeline::new());
+ consensus.init();
+ consensus.restore_commit_state(5, 5);
+
+ // Op 6 never arrived; 7 and 8 did, and the cluster committed through
8.
+ consensus.pipeline_message(PlaneKind::Metadata, &prepare_message(7, 0,
70));
+ consensus.pipeline_message(PlaneKind::Metadata, &prepare_message(8,
70, 80));
+ consensus.advance_commit_max(8);
+
+ assert!(
+ peek_committable_head(&consensus).is_none(),
+ "the head is covered by the frontier but op 6 is not applied"
+ );
+ assert!(
+ drain_committable_prefix(&consensus).is_empty(),
+ "the drain must hold on the same hole the peek does"
+ );
+ assert_eq!(
+ consensus.pipeline_head_header().map(|header| header.op),
+ Some(7),
+ "holding must not consume the entry"
+ );
+ }
+
+ /// The shape the journal-walk caps prevent. Both gates must still refuse
it:
+ /// re-applying an applied op panics `advance_commit_min`.
+ #[test]
+ fn given_an_applied_head_when_committing_should_refuse_it() {
+ let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus,
LocalPipeline::new());
+ consensus.init();
+ consensus.restore_commit_state(4, 4);
+
+ consensus.pipeline_message(PlaneKind::Metadata, &prepare_message(5, 0,
50));
+ consensus.advance_commit_max(6);
+ // The journal walk got there first and applied 5 and 6 out of the WAL.
+ consensus.advance_commit_min(5);
+ consensus.advance_commit_min(6);
+
+ assert!(
+ peek_committable_head(&consensus).is_none(),
+ "op 5 is already applied; re-applying it panics advance_commit_min"
+ );
+ assert!(
+ drain_committable_prefix(&consensus).is_empty(),
+ "the drain must refuse an applied head too"
+ );
+ }
+
#[test]
fn drains_only_up_to_commit_frontier_even_without_quorum_flags() {
let consensus = VsrConsensus::new(1, 0, 3, 0, NoopBus,
LocalPipeline::new());
consensus.init();
+ // Pipeline opens at op 5, so the state machine must already be
through 4.
+ consensus.restore_commit_state(4, 4);
consensus.pipeline_message(PlaneKind::Metadata, &prepare_message(5, 0,
50));
consensus.pipeline_message(PlaneKind::Metadata, &prepare_message(6,
50, 60));
diff --git a/core/metadata/src/impls/metadata.rs
b/core/metadata/src/impls/metadata.rs
index 3c4d19f47..1522cfbb5 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -3653,6 +3653,24 @@ where
applied += 1;
let op = consensus.commit_min() + 1;
+ // Never apply the op the pipeline holds: `on_ack` pops that entry
+ // and advances past it, so applying it here strands it below
+ // `peek_committable_head`'s floor. Nothing pops it after that (the
+ // only `pop_committed_prepare` sits inside that loop), so its wire
+ // reply is never built and its `reply_sender` neither fires nor
drops.
+ //
+ // Compared per op, not hoisted: the body read below awaits and a
+ // sibling driver can move the head. Comparing against `op` also
gets
+ // the two edge cases right for free -- an absent head is a
backup's
+ // empty pipeline and caps nothing, and a head at or below
`commit_min`
+ // is already stranded and must not freeze the walk on top of that.
+ if consensus
+ .pipeline_head_header()
+ .is_some_and(|head| head.op == op)
+ {
+ break;
+ }
+
let Some(header) = journal.handle().header(op as usize) else {
// Gap-stop: the walk halts at the first missing prepare and
// resumes once it is refilled -- by the primary's retransmit
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 5d363ef24..e6a221971 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -43,9 +43,9 @@ use consensus::{
PlaneKind, Project, 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,
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,
+ emit_partition_diag, emit_sim_event, fence_old_prepare_by_commit,
repair_session_live,
+ 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,
};
use iggy_binary_protocol::requests::consumer_offsets::{
DeleteConsumerOffsetRequest, StoreConsumerOffsetRequest,
@@ -4304,13 +4304,34 @@ where
/// Committable entries (ops `commit_min+1 ..= commit_max`) read from the
/// journal, for a backup whose pipeline is empty. Stops at the first
missing
/// op: a replication gap must not be skipped, or `advance_commit_min`'s
- /// sequential contract breaks. Like the metadata plane's `commit_journal`,
- /// the journal keeps its committed entries until they are flushed
- /// (`commit_messages` drains only the committed prefix), so this read
finds
- /// every committed op while the uncommitted tail stays resident.
+ /// sequential contract breaks.
+ ///
+ /// KNOWN GAP: resident headers only. `commit_messages` evicts up to
+ /// `commit_max` (the cluster frontier, not this replica's commit point)
while
+ /// `committed_headers_from` never reads the evicted ring, so a backlog
past
+ /// [`COMMIT_WALK_OPS_MAX`] can have its un-reached ops flushed out from
under
+ /// it and stop. Repair refetches them and the simulator's contiguity
invariant
+ /// catches a walk that never recovers. Reading the ring here would close
it
+ /// directly, but not as a one-line swap: the apply path needs batch bytes
and
+ /// the ring is capacity-bounded, so headers it cannot back with bytes
would
+ /// fence the partition instead of stalling it.
fn collect_committable_from_journal(&self, max_ops: usize) ->
Vec<PipelineEntry> {
let from_op = self.consensus.commit_min() + 1;
+ // Stop below the pipeline head. The drain above holds rather than pops
+ // when the head is not the op owed next; walking that op out of the
+ // journal instead advances `commit_min` past a still-resident entry,
and
+ // `on_ack` then finds the drain empty on every later ack -- no reply
is
+ // ever shipped and each stranded entry leaves its awaiter parked.
+ //
+ // Only a head at or above `from_op` lowers the ceiling: an absent
head is
+ // a backup's empty pipeline, and a lower head is already stranded and
must
+ // not freeze the walk on top of that.
let commit_max = self.consensus.commit_max();
+ let commit_max = self
+ .consensus
+ .pipeline_head_header()
+ .filter(|head| head.op >= from_op)
+ .map_or(commit_max, |head| commit_max.min(head.op - 1));
self.log
.journal()
.inner
@@ -6708,7 +6729,11 @@ where
return;
};
let consensus = self.consensus();
- if !consensus.is_normal() || consensus.view() != session.view {
+ // NOT `is_normal` alone: a primary-elect repairing toward its parked
+ // merged log runs this in `ViewChange`, and dropping the session on
its
+ // first inbound frame leaves the coverage scan re-arming every tick
over
+ // a stream it can never keep.
+ if !repair_session_live(consensus) || consensus.view() != session.view
{
self.repair = None;
return;
}
@@ -8806,28 +8831,40 @@ mod tests {
assert!(!dir.path().join("2").exists());
}
+ /// `AckLevel::NoAck` stores apply on the primary only and never
replicate, so
+ /// which replicas hold an offset is not agreed and a committed delete can
+ /// legitimately find nothing. Erroring on that fails the committed apply,
+ /// fences the partition, then crash-loops on every replay of the op.
+ ///
+ /// Both kinds, because they are separate maps with separate directories.
#[compio::test]
async fn
given_absent_offset_file_when_delete_commits_should_skip_directory_sync() {
- let dir = tempfile::tempdir().unwrap();
- let (mut partition, sent) = recording_partition_at(0, 3);
- partition.consumer_offsets_path =
- Some(dir.path().join("missing").to_string_lossy().into_owned());
- partition.stage_consumer_offset_delete(1, ConsumerKind::Consumer, 7);
- partition.consensus.restore_commit_state(0, 1);
- let header = PrepareHeader {
- op: 1,
- operation: Operation::DeleteConsumerOffset,
- client: 42,
- request: 1,
- ..Default::default()
- };
- partition
- .handle_committed_entries(vec![PipelineEntry::new(header)],
&repair_config(), true)
- .await;
- assert!(partition.fatal.is_none());
- assert_eq!(partition.consensus.commit_min(), 1);
- assert_eq!(partition.offset_dir_sync_count.get(), 0);
- assert_eq!(sent.borrow().len(), 1);
+ for (op, kind) in [
+ (1, ConsumerKind::Consumer),
+ (2, ConsumerKind::ConsumerGroup),
+ ] {
+ let dir = tempfile::tempdir().unwrap();
+ let (mut partition, sent) = recording_partition_at(0, 3);
+ let missing =
Some(dir.path().join("missing").to_string_lossy().into_owned());
+ partition.consumer_offsets_path.clone_from(&missing);
+ partition.consumer_group_offsets_path = missing;
+ partition.stage_consumer_offset_delete(op, kind, 7);
+ partition.consensus.restore_commit_state(op - 1, op);
+ let header = PrepareHeader {
+ op,
+ operation: Operation::DeleteConsumerOffset,
+ client: 42,
+ request: 1,
+ ..Default::default()
+ };
+ partition
+ .handle_committed_entries(vec![PipelineEntry::new(header)],
&repair_config(), true)
+ .await;
+ assert!(partition.fatal.is_none(), "{kind:?} delete must not
fence");
+ assert_eq!(partition.consensus.commit_min(), op);
+ assert_eq!(partition.offset_dir_sync_count.get(), 0);
+ assert_eq!(sent.borrow().len(), 1);
+ }
}
#[compio::test]
@@ -11073,6 +11110,63 @@ mod tests {
.expect("journal append");
}
+ /// Walking through the head advances `commit_min` past a resident entry
only
+ /// `on_ack` can pop and answer, after which every later ack finds the
drain
+ /// empty and no reply is ever shipped.
+ #[compio::test]
+ async fn
given_a_pipeline_head_when_walking_the_journal_should_stop_below_it() {
+ let partition = test_partition();
+ for op in 1..=4 {
+ journal_prepare(&partition, op, Operation::CreateStream).await;
+ }
+ partition.consensus.restore_commit_state(0, 4);
+ partition.consensus.pipeline_message(
+ PlaneKind::Partitions,
+ &pipeline_prepare(3, Operation::CreateStream),
+ );
+
+ let ops: Vec<u64> = partition
+ .collect_committable_from_journal(COMMIT_WALK_OPS_MAX)
+ .into_iter()
+ .map(|entry| entry.header.op)
+ .collect();
+ assert_eq!(
+ ops,
+ vec![1, 2],
+ "the walk stops at the op the pipeline holds, leaving 3 to on_ack"
+ );
+ }
+
+ /// A backup journals replicated prepares and never populates a pipeline,
so an
+ /// absent head must mean NO ceiling. Read as a ceiling of zero it would
stop
+ /// every backup's commit walk.
+ #[compio::test]
+ async fn given_an_empty_pipeline_when_walking_the_journal_should_not_cap()
{
+ let partition = test_partition();
+ for op in 1..=3 {
+ journal_prepare(&partition, op, Operation::CreateStream).await;
+ }
+ partition.consensus.restore_commit_state(0, 3);
+ assert!(partition.consensus.pipeline_head_header().is_none());
+
+ let ops: Vec<u64> = partition
+ .collect_committable_from_journal(COMMIT_WALK_OPS_MAX)
+ .into_iter()
+ .map(|entry| entry.header.op)
+ .collect();
+ assert_eq!(ops, vec![1, 2, 3], "a backup walks its whole committed
run");
+ }
+
+ fn pipeline_prepare(op: u64, operation: Operation) ->
Message<PrepareHeader> {
+ let size = std::mem::size_of::<PrepareHeader>();
+ Message::<PrepareHeader>::new(size).transmute_header(|_, header: &mut
PrepareHeader| {
+ header.command = Command::Prepare;
+ header.op = op;
+ header.operation = operation;
+ header.size = u32::try_from(size).expect("prepare header size fits
in u32");
+ })
+ }
+
/// A repaired `SendMessages` prepare with an explicit chain identity, as a
/// serving peer ships it.
pub(super) fn repaired_send_prepare(
diff --git a/core/server/src/http/reads.rs b/core/server/src/http/reads.rs
index dc3262ff3..bcb387aa1 100644
--- a/core/server/src/http/reads.rs
+++ b/core/server/src/http/reads.rs
@@ -276,17 +276,23 @@ pub(in crate::http) async fn await_recovery_barrier(
let Some(consensus) = shard.plane.metadata().consensus.as_ref() else {
return Ok(());
};
- let barrier = consensus.recovery_barrier();
// Gate on commit_MIN (locally applied), not commit_max (known committed):
// a StartView adoption advances commit_max first and only then walks the
// journal applying ops, and this task interleaves with that walk at its
// await points -- a commit_max gate would serve state from before the
// suffix applied (e.g. a pre-restart password change not yet visible).
- if barrier_state(barrier, consensus.commit_min(), false) ==
BarrierWait::Ready {
+ if barrier_state(consensus.recovery_barrier(), consensus.commit_min(),
false)
+ == BarrierWait::Ready
+ {
return Ok(());
}
let deadline = std::time::Instant::now() + consensus.recovery_deadline();
loop {
+ // Re-read per poll, like `commit_min`. `redecide_recovery_barrier`
lowers
+ // the barrier when a view change settles the recovered suffix, so a
reader
+ // holding the value it captured on entry waits out a barrier that no
longer
+ // exists and then 503s on a replica that is serving.
+ let barrier = consensus.recovery_barrier();
let expired = std::time::Instant::now() >= deadline;
match barrier_state(barrier, consensus.commit_min(), expired) {
BarrierWait::Ready => return Ok(()),
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 6f6f9d546..ae32c2261 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -906,6 +906,14 @@ pub const REPAIR_CHUNK_MAX: u64 = 128;
#[derive(Debug, Clone, Copy)]
struct MetadataRepairSession {
nonce: u128,
+ /// Lowest op this session must fetch, and the floor its stall retry
reopens at.
+ ///
+ /// Not re-derivable from `commit_min + 1`: the merged-log scan opens at
+ /// [`merged_log_scan_floor`], above the snapshot floor, while its
+ /// `committed_elsewhere` fallback reports ops below even that. A
recomputed
+ /// retry asks for a different window than the one reported missing, and
never
+ /// re-asks for the op that was.
+ from_op: u64,
to_op: u64,
/// Consensus view this session was armed in. A later view decides the log
/// again, so the window this names may no longer be the one to fetch;
@@ -1585,11 +1593,28 @@ where
/// [`Self::metadata_transfer_decode_failures`].
metadata_transfer_attempts: Cell<u32>,
- /// Consecutive stalled re-requests on the live metadata repair session,
- /// against [`partitions::REPAIR_MAX_STALL_RETRIES`]. Survives the session,
- /// so rotating the peer cannot reset it; cleared by an accepted repaired
- /// prepare.
- metadata_repair_attempts: Cell<u32>,
+ /// Consecutive stall rounds burned by the metadata repair session, against
+ /// [`partitions::REPAIR_MAX_STALL_RETRIES`]. Bounds how long one quiet
peer
+ /// pins the commit walk.
+ ///
+ /// On the shard, not the session: rotation mints a fresh session, so a
+ /// per-session counter would reset itself. Nothing on the rotation path
may
+ /// clear it either.
+ ///
+ /// Cleared only by [`Self::note_metadata_repair_walked`], which takes
evidence
+ /// attributable to the targeted peer. A repair prepare carries no sender
and
+ /// no nonce, so it restarts the stall clock only. Net effect: this bounds
peers
+ /// that go silent, serve unusable bytes, or terminate without closing the
gap.
+ /// It does not bound merely slow peers, whose chunks keep the clock from
+ /// firing.
+ ///
+ /// Paired with the view the rounds were charged in, and spent per view.
The
+ /// merged-log arm refuses to re-arm once the budget is out, and the
exhausting
+ /// path leaves no session behind, so nothing would ever be superseded or
walk:
+ /// unfenced, one spent budget would refuse merged-log repair for every
later
+ /// view for the life of the process, and a replica that keeps winning
+ /// elections would never repair again.
+ metadata_repair_attempts: Cell<(u32, u32)>,
/// Decode failures charged against one snapshot generation, as
/// `(snapshot_seq, failures)`. `None` until a pulled artifact set first
@@ -1744,7 +1769,7 @@ where
superblock_wedged_fatal_failures: Cell::new(0),
bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE),
metadata_transfer_attempts: Cell::new(0),
- metadata_repair_attempts: Cell::new(0),
+ metadata_repair_attempts: Cell::new((0, 0)),
metadata_transfer_decode_failures: Cell::new(None),
})
}
@@ -2231,7 +2256,7 @@ where
superblock_wedged_fatal_failures: Cell::new(0),
bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE),
metadata_transfer_attempts: Cell::new(0),
- metadata_repair_attempts: Cell::new(0),
+ metadata_repair_attempts: Cell::new((0, 0)),
metadata_transfer_decode_failures: Cell::new(None),
}
}
@@ -5037,6 +5062,12 @@ where
// Applies to both planes, and is why a backup parks a log at all.
The
// view already decided which prepare belongs at this op; a
different
// one forks the log. An op the parked log omits is unconstrained.
+ //
+ // Which covers most of the range under `pending.commit_max`: the
merged
+ // log names headers only from the DVC suffixes, and those span
+ // `commit..=op` per sender. Identity below that rests on
crash-stop --
+ // a committed op is the quorum's op. `verify_prepare_integrity`
below
+ // guards corruption; neither guards Byzantine faults.
let disagrees = consensus
.with_pending_view_log(|pending| {
pending
@@ -5069,15 +5100,19 @@ where
let Some(journal) = planes.0.journal.as_ref() else {
return;
};
- // Above the two returns below, not after them: only SILENCE should
- // age the stream, and an in-scope frame proves the peer is serving
- // whether or not this replica still needs the op it carries. The
- // ops a re-request re-serves are exactly the ones already held, so
- // counting accepted frames alone rotates away from a live peer.
- if let Some(session) = self.metadata_repair.borrow_mut().as_mut() {
- session.idle_ticks = 0;
- }
- self.note_metadata_repair_progress();
+ // Below the divergence and integrity returns, above the two under
it.
+ //
+ // Only silence should age the stream, and only a frame this
replica
+ // would have accepted proves anything is being served. A forked or
+ // corrupted frame is neither: the peer re-serves the same stored
bytes
+ // every re-request, so crediting those holds off the retry
forever.
+ //
+ // Still above the dedup return: a re-request re-serves ops already
+ // held, and a stream re-covering ground is still a stream.
+ //
+ // Clock only -- the frame has no sender and no nonce, so it
cannot be
+ // attributed. The budget is cleared from the terminator and the
walk.
+ self.note_metadata_repair_clock();
let journal = journal.handle();
#[allow(clippy::cast_possible_truncation)]
if journal.header(header.op as usize).is_some() {
@@ -5190,6 +5225,14 @@ where
done,
"metadata journal repair walked"
);
+ // The one attributable credit: fenced on `session.nonce`
+ // above, so it came from the targeted peer, and the walk
moved.
+ // Gating on the walk and not the frame keeps a peer that
+ // terminates every round while serving nothing useful from
+ // clearing its own budget and becoming un-rotatable.
+ if done || commit_min > before {
+ self.note_metadata_repair_walked();
+ }
if done {
*self.metadata_repair.borrow_mut() = None;
} else if repair_chunk_walked(before, commit_min,
header.op) {
@@ -5225,6 +5268,95 @@ where
// `RangeEvicted` again if the primary checkpointed mid
// transfer -- that reraises through the same path, and
each
// round lifts the local floor, so it converges.
+ //
+ // Never as primary-elect. A transfer replaces
snapshot-shaped
+ // state wholesale, and this replica has a merged log
parked
+ // against that state naming ops it was just told it cannot
+ // serve; installing under it starts the view over a log
the new
+ // state no longer matches. Re-target instead, and let the
+ // view-change timeout escalate if nobody can serve it.
+ //
+ // DROP the session before returning. The serving peer
follows
+ // `RangeEvicted` with `RepairDone(from_op - 1)` on the
same
+ // nonce, and a `RepairDone` at or below `commit_min` walks
+ // nothing, so `repair_chunk_walked` is trivially true and
that
+ // arm re-requests at once: no tick gate, no debounce, no
attempt
+ // burned -- an unthrottled request/reply loop across two
pumps.
+ // Dropping first lands the trailing frame on the `is_none`
+ // guard, as it did before this arm existed.
+ if consensus.view_log_is_pending()
+ && consensus.is_primary_for_view(consensus.view())
+ {
+ tracing::warn!(
+ shard = self.id,
+ peer = header.replica,
+ retained_from = header.op,
+ local_commit = consensus.commit_min(),
+ "merged-log repair peer evicted the requested
range; \
+ re-targeting rather than transferring state mid
view change"
+ );
+ // Definitive, not a stall: this sender has said it
cannot
+ // serve the window, so rotate now rather than spend a
retry
+ // interval on a stream that will not come. Still
charge a
+ // round, so a quorum that all answer this way stops
asking
+ // instead of cycling the ring until the timeout.
+ if self.burn_metadata_repair_attempt(consensus.view())
{
+ *self.metadata_repair.borrow_mut() = None;
+ tracing::warn!(
+ shard = self.id,
+ from_op = session.from_op,
+ to_op = session.to_op,
+ "merged-log repair exhausted its senders;
leaving the view \
+ change to its timeout"
+ );
+ return;
+ }
+ self.rotate_stalled_metadata_repair(
+ consensus,
+ header.replica,
+ session.from_op,
+ session.to_op,
+ )
+ .await;
+ return;
+ }
+
+ // The floor must also be ABOVE the op this replica needs.
A
+ // peer behind the requested window walks its serve range
off
+ // the end and answers `RangeEvicted` at the requested
floor
+ // itself, having retained nothing and evicted nothing;
+ // converting on that arms a transfer against a replica
with
+ // less state than this one and fences repair for a full
+ // transfer backoff. Drop the session and let the level
trigger
+ // re-request from the primary instead.
+ if header.op <= consensus.commit_min() + 1 {
+ tracing::warn!(
+ shard = self.id,
+ peer = header.replica,
+ retained_from = header.op,
+ local_commit = consensus.commit_min(),
+ "metadata repair peer retained nothing in the
requested range; \
+ re-requesting rather than converting to state
transfer"
+ );
+ // Charge the round, and ACT on exhaustion.
`gap_repair_peer`
+ // re-picks the primary deterministically, so dropping
the
+ // session on its own re-arms the same peer at the
debounce
+ // interval forever: the stall path never runs, so
rotation
+ // is never reached and the state-transfer escalation
this
+ // guard replaced stays out of reach.
+ if self.burn_metadata_repair_attempt(consensus.view())
{
+ self.rotate_stalled_metadata_repair(
+ consensus,
+ header.replica,
+ session.from_op,
+ session.to_op,
+ )
+ .await;
+ return;
+ }
+ *self.metadata_repair.borrow_mut() = None;
+ return;
+ }
if consensus.state_transfer_stage() ==
consensus::StateTransferStage::Idle {
*self.metadata_repair.borrow_mut() = None;
consensus.begin_state_transfer_await();
@@ -5269,7 +5401,12 @@ where
if header.nonce != session.nonce {
return;
}
- if !partition.consensus().is_normal() || partition.consensus().view()
!= session.view {
+ // Twin of the `apply_repaired_prepare` gate: a primary-elect's
merged-log
+ // session legitimately runs outside `Normal`, and dropping it here
would
+ // discard the terminator that closes the window it is repairing.
+ if !consensus::repair_session_live(partition.consensus())
+ || partition.consensus().view() != session.view
+ {
partition.repair = None;
return;
}
@@ -5559,11 +5696,23 @@ where
// cannot serve the state transfer it itself needs. A committed op
// cannot diverge from the merged log, and its bytes stay serveable
// from the evicted ring or the flushed segments.
+ let floor = ScanFloor {
+ repair_floor: consensus.commit_min(),
+ commit_min: consensus.commit_min(),
+ };
+ // Ring AND resident, one pass. `header_by_op` reads the resident
vec
+ // only, while `commit_messages` evicts up to `commit_max` (the
cluster
+ // frontier), so a primary-elect with an apply backlog reads
`None` for
+ // ops it holds in the repair ring and parks on a hole that is not
one.
+ // One pass also because `header_by_op` is a linear scan and this
window
+ // is the apply backlog, not the `prepare_queue_max` span the merge
+ // bounds -- probing per op is quadratic.
let missing = {
let journal = partition.log.journal();
- first_op_not_covered(&pending, consensus.commit_min(), |op| {
- journal.inner.header_by_op(op)
- })
+ let window = journal
+ .inner
+
.repair_headers_in(floor.opens_at(&pending)..=pending.op_head);
+ first_op_not_covered(&pending, floor, |op|
window.get(&op).copied())
};
if let Some(missing_op) = missing {
tracing::debug!(
@@ -5573,6 +5722,12 @@ where
op_head = pending.op_head,
"partition view change waiting on op {missing_op} before
starting the view"
);
+ // And a way out of the wait: nothing else fetches this op.
+ // `maybe_request_partition_repair` refuses outside `Normal`
and
+ // the sweep's gap detector needs `probe.normal`. Partition
twin of
+ // the metadata plane's view repair.
+ self.request_partition_view_repair(partition, missing_op,
pending.op_head, None)
+ .await;
return;
}
@@ -5606,6 +5761,11 @@ where
///
/// Repair frames are fire-and-forget, so a lost one leaves the session
armed
/// forever with the commit walk pinned below the frontier.
+ ///
+ /// Re-requests from the SAME peer while its stall budget holds. A peer
that
+ /// never answers is a different problem, and
+ /// [`Self::rotate_stalled_metadata_repair`] owns it: the target lives on
the
+ /// session, so it must be replaced there rather than shadowed for one
send.
#[allow(clippy::future_not_send)]
async fn retry_stalled_metadata_repair<P>(&self, consensus:
&VsrConsensus<B, P>)
where
@@ -5640,7 +5800,7 @@ where
"metadata repair session walked or superseded; closing it"
);
*self.metadata_repair.borrow_mut() = None;
- self.note_metadata_repair_progress();
+ self.note_metadata_repair_walked();
return;
}
@@ -5654,55 +5814,32 @@ where
return None;
}
session.idle_ticks = 0;
- Some((session.peer, session.nonce, session.to_op))
+ Some((session.peer, session.nonce, session.from_op,
session.to_op))
})
};
- if let Some((peer, nonce, to_op)) = stalled {
- // A session pins its peer and fences every arming site while it
- // stands, so a peer that cannot answer wedges the plane harder
than
- // having no session at all -- and the gap-stopped-primary rotation
- // can pick a peer that is simply down. Past the budget the session
- // is dropped and re-armed one step around the ring; an ordinary
lost
- // frame is re-requested long before that.
- if self.burn_metadata_repair_attempt() {
- let next_peer = next_transfer_peer(
- consensus.replica(),
- peer,
- consensus.replica_count(),
- consensus.primary_index(consensus.view()),
- );
- tracing::warn!(
+ if let Some((peer, nonce, session_from_op, to_op)) = stalled {
+ let from_op =
+ stalled_repair_from_op(session_from_op,
consensus.commit_min(), repairing_view);
+ if from_op > to_op {
+ // `from_op` past `to_op` without `commit_min` reaching it: the
+ // primary-elect window above starts at the merged log's commit
+ // point, which can sit above what this replica has walked. The
+ // top-of-tick check closes the ordinary case; this closes the
+ // one it cannot see. Leaving it armed wedges the replica: no
+ // `RepairDone` clears a window the walk is already past, and
the
+ // `is_none` gate then blocks the session the ops above it
need.
+ tracing::info!(
shard = self.id,
- peer,
- next_peer,
to_op,
- "metadata repair stalled past its retry budget; re-arming
from another \
- replica"
+ peer,
+ "metadata repair window fully requested; closing the
stalled session"
);
*self.metadata_repair.borrow_mut() = None;
- self.note_metadata_repair_progress();
- if next_peer != peer {
- self.maybe_request_metadata_repair(consensus, next_peer)
- .await;
- }
- // Nobody else to name (a solo group, or a two-replica group
- // whose only peer went quiet): dropping the session is still
- // right, since it unfences the detector, which re-arms after
its
- // debounce and logs the state each interval.
- return;
- }
- // Primary-elect only. Its window starts at the merged log's commit
- // point, which can sit below local `commit_min` (the headers
inherited
- // from senders behind the canonical log_view live there), so
- // `commit_min + 1` would skip them. A backup's parked `StartView`
- // suffix is only a verification reference; resuming from its
commit
- // point would restart at the view's opening head, not at the gap.
- let from_op = consensus
- .is_primary_for_view(consensus.view())
- .then(|| consensus.with_pending_view_log(|pending|
pending.commit_max.max(1)))
- .flatten()
- .unwrap_or_else(|| consensus.commit_min() + 1);
- if from_op <= to_op {
+ self.note_metadata_repair_walked();
+ } else if self.burn_metadata_repair_attempt(consensus.view()) {
+ self.rotate_stalled_metadata_repair(consensus, peer, from_op,
to_op)
+ .await;
+ } else {
tracing::info!(
shard = self.id,
from_op,
@@ -5720,22 +5857,95 @@ where
consensus.group(),
)
.await;
- } else {
- // `from_op` past `to_op` without `commit_min` reaching it: the
- // primary-elect window above starts at the merged log's commit
- // point, which can sit above what this replica has walked. The
- // top-of-tick check closes the ordinary case; this closes the
- // one it cannot see.
- tracing::info!(
+ }
+ }
+ }
+
+ /// Re-arm a repair session that spent its stall budget against another
replica.
+ ///
+ /// A session pins its peer and fences every arming site while it stands,
so a
+ /// peer that cannot answer wedges the walk harder than having no session
at
+ /// all. Past the budget the session is dropped and re-armed one step on;
an
+ /// ordinary lost frame is re-requested long before that. Mirrors the
partition
+ /// rotation in [`Self::tick_partitions`].
+ ///
+ /// Two rings, because two things decide who can serve. A `Normal` backup
is
+ /// repairing its committed tail and any replica ahead of it will do, so it
+ /// walks the cluster preferring the primary. A primary-elect is repairing
+ /// toward a merged log, and only the `DoViewChange` senders that named
the op
+ /// can serve it: walking the whole ring lands on a replica that answers
+ /// `RangeEvicted` for a range it never held.
+ ///
+ /// Does NOT spend the budget. It lives on the shard so rotation cannot
reset it
+ /// (see [`Self::metadata_repair_attempts`]); clearing it here would bound
one
+ /// round and re-target forever.
+ #[allow(clippy::future_not_send)]
+ async fn rotate_stalled_metadata_repair<P>(
+ &self,
+ consensus: &VsrConsensus<B, P>,
+ peer: u8,
+ from_op: u64,
+ to_op: u64,
+ ) where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ *self.metadata_repair.borrow_mut() = None;
+
+ if consensus.view_log_is_pending() &&
consensus.is_primary_for_view(consensus.view()) {
+ let sources = view_repair_sources(consensus, from_op);
+ let Some(next_peer) = next_view_repair_peer(&sources, Some(peer))
else {
+ // Only the quiet peer named this op. The session is dropped
either
+ // way: `advance_pending_metadata_view` re-scans on the next
tick and
+ // re-requests it, and a peer that never comes back leaves the
+ // view-change timeout to escalate.
+ tracing::warn!(
shard = self.id,
- to_op,
peer,
- "metadata repair window fully requested; closing the
stalled session"
+ from_op,
+ "no other replica offers op {from_op} for the merged log; \
+ view change is stalled"
);
- *self.metadata_repair.borrow_mut() = None;
- self.note_metadata_repair_progress();
- }
+ return;
+ };
+ tracing::warn!(
+ shard = self.id,
+ peer,
+ next_peer,
+ from_op,
+ to_op,
+ "merged-log repair stalled past its retry budget; re-arming
from \
+ another do_view_change sender"
+ );
+ self.arm_metadata_repair_session(consensus, next_peer, from_op,
to_op)
+ .await;
+ return;
+ }
+
+ let primary = consensus.primary_index(consensus.view());
+ let next_peer = next_transfer_peer(
+ consensus.replica(),
+ peer,
+ consensus.replica_count(),
+ primary,
+ );
+ if next_peer == peer {
+ // The ring had nobody else to offer (a solo group, or a
two-replica
+ // cluster whose only peer went quiet). Dropping the session is
still
+ // right: it unfences the level trigger below, which re-requests
the
+ // window on the next tick.
+ return;
}
+ tracing::warn!(
+ shard = self.id,
+ peer,
+ next_peer,
+ from_op,
+ to_op,
+ "metadata repair stalled past its retry budget; re-arming from
another replica"
+ );
+ self.maybe_request_metadata_repair(consensus, next_peer)
+ .await;
}
/// Compare this replica's log against the headers the view decided, and
drop or
@@ -5926,7 +6136,11 @@ where
// one back: demanding one parks the view change forever on an op
already
// applied and durable in the snapshot.
let repair_floor = journal.handle().snapshot_op();
- let missing = first_op_not_covered(&pending, repair_floor, |op| {
+ let floor = ScanFloor {
+ repair_floor,
+ commit_min: consensus.commit_min(),
+ };
+ let missing = first_op_not_covered(&pending, floor, |op| {
usize::try_from(op)
.ok()
.and_then(|slot| journal.handle().header(slot))
@@ -5967,7 +6181,23 @@ where
// Stream already running; the stall retry covers it drying up.
return;
}
- let sources = consensus.pending_view_body_sources(missing_op);
+ // Level-triggered, so it would re-arm against the head of the same
source
+ // list next tick -- including the sender that just answered
`RangeEvicted`,
+ // which is how the budget got spent. Once every sender has been asked
and
+ // charged, asking again is not progress; the view-change timeout is.
+ //
+ // Per view. A later view is a new merged log from a new quorum, so the
+ // senders that refused this one say nothing about it.
+ if self.metadata_repair_exhausted(consensus.view()) {
+ tracing::debug!(
+ shard = self.id,
+ missing_op,
+ view = consensus.view(),
+ "merged-log repair is out of retries for this view; not
re-arming"
+ );
+ return;
+ }
+ let sources = view_repair_sources(consensus, missing_op);
let Some(peer) = sources.first().copied() else {
// The merge only returns a startable log when some replica
offered each
// body, so an empty source list means that offer was withdrawn
(peer
@@ -5980,14 +6210,6 @@ where
return;
};
- let nonce = iggy_common::random_id::get_uuid();
- *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
- nonce,
- to_op: pending.op_head,
- view: consensus.view(),
- peer,
- idle_ticks: 0,
- });
tracing::info!(
shard = self.id,
missing_op,
@@ -5995,13 +6217,45 @@ where
to_op = pending.op_head,
"repairing toward the merged log before starting the view"
);
+ self.arm_metadata_repair_session(consensus, peer, missing_op,
pending.op_head)
+ .await;
+ }
+
+ /// Mint a metadata repair session and send its first request.
+ ///
+ /// Three sites arm one (the view-change scan, the tail-repair funnel, the
+ /// stall rotation). Keeping the invariant here -- fresh nonce, arming
view,
+ /// clock from zero, a `from_op` the retry can reopen at -- stops a re-arm
from
+ /// shipping a window its own retry cannot reproduce.
+ ///
+ /// Callers decide whether to arm; this decides what an armed session is.
+ #[allow(clippy::future_not_send)]
+ async fn arm_metadata_repair_session<P>(
+ &self,
+ consensus: &VsrConsensus<B, P>,
+ peer: u8,
+ from_op: u64,
+ to_op: u64,
+ ) where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+ {
+ let nonce = iggy_common::random_id::get_uuid();
+ *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
+ nonce,
+ from_op,
+ to_op,
+ view: consensus.view(),
+ peer,
+ idle_ticks: 0,
+ });
self.send_request_prepares(
consensus.cluster(),
consensus.replica(),
peer,
nonce,
- missing_op,
- pending.op_head,
+ from_op,
+ to_op,
consensus.group(),
)
.await;
@@ -6013,11 +6267,14 @@ where
/// Every TAIL arming site funnels through here -- `StartView` adoption,
the
/// commit-heartbeat backstop, the state-transfer fallbacks, and
/// `tick_metadata`'s gap detector -- so the guards below are what make the
- /// level-triggered one idempotent. The one session this does not mint is
- /// the view-change repair `advance_pending_metadata_view` builds inline:
it
- /// repairs toward a merged log rather than the commit frontier, from a
peer
- /// that offered the body rather than from the primary, so none of the
- /// guards below describe it.
+ /// level-triggered one idempotent.
+ ///
+ /// It does not decide the merged-log sessions:
+ /// [`Self::advance_pending_metadata_view`] arms them and
+ /// [`Self::rotate_stalled_metadata_repair`] re-targets them. Those repair
+ /// toward a parked merged log rather than the commit frontier, from a
sender
+ /// that named the op rather than the primary, so none of the guards below
fit.
+ /// All three share [`Self::arm_metadata_repair_session`].
#[allow(clippy::future_not_send)]
async fn maybe_request_metadata_repair<P>(&self, consensus:
&VsrConsensus<B, P>, peer: u8)
where
@@ -6038,7 +6295,6 @@ where
&& consensus.commit_min() < consensus.commit_max()
&& self.metadata_repair.borrow().is_none()
{
- let nonce = iggy_common::random_id::get_uuid();
let to_op = consensus.commit_max();
let from_op = consensus.commit_min() + 1;
// Spent here rather than at the detector, so the edge-triggered
@@ -6046,13 +6302,6 @@ where
// the next tick would otherwise leave the count saturated and hand
// the next real gap an arm on its first tick.
self.metadata_gap_ticks.set(0);
- *self.metadata_repair.borrow_mut() = Some(MetadataRepairSession {
- nonce,
- to_op,
- view: consensus.view(),
- peer,
- idle_ticks: 0,
- });
tracing::info!(
shard = self.id,
from_op,
@@ -6060,16 +6309,8 @@ where
peer,
"metadata behind the group frontier; requesting repair"
);
- self.send_request_prepares(
- consensus.cluster(),
- consensus.replica(),
- peer,
- nonce,
- from_op,
- to_op,
- consensus.group(),
- )
- .await;
+ self.arm_metadata_repair_session(consensus, peer, from_op, to_op)
+ .await;
}
}
@@ -6600,21 +6841,23 @@ where
/// on the shard here for the same reason `metadata_transfer_attempts`
does:
/// one metadata group per node. It has to outlive the SESSION either way,
/// or the rotation that mints a new one would reset the count and
re-target
- /// forever without ever giving up on a peer.
- fn burn_metadata_repair_attempt(&self) -> bool {
- let attempts = self.metadata_repair_attempts.get() + 1;
- self.metadata_repair_attempts.set(attempts);
+ /// forever without giving up on a peer. Only
+ /// [`Self::note_metadata_repair_walked`] clears it.
+ fn burn_metadata_repair_attempt(&self, view: u32) -> bool {
+ let (charged_view, attempts) = self.metadata_repair_attempts.get();
+ let attempts = if charged_view == view {
+ attempts + 1
+ } else {
+ 1
+ };
+ self.metadata_repair_attempts.set((view, attempts));
attempts > partitions::REPAIR_MAX_STALL_RETRIES
}
- /// The serving peer answered: reset the budget, so it bounds CONSECUTIVE
- /// silence rather than the stalls a long healthy stream accumulates.
- ///
- /// Any in-scope repair frame, not only an accepted one. A re-request
- /// re-serves ops this replica already holds, so charging those as silence
- /// rotates away from a peer that is answering.
- fn note_metadata_repair_progress(&self) {
- self.metadata_repair_attempts.set(0);
+ /// Whether this view has already spent its merged-log repair budget.
+ const fn metadata_repair_exhausted(&self, view: u32) -> bool {
+ let (charged_view, attempts) = self.metadata_repair_attempts.get();
+ charged_view == view && attempts > partitions::REPAIR_MAX_STALL_RETRIES
}
/// Burn one retry round; `true` once the budget is exhausted.
@@ -6633,6 +6876,41 @@ where
self.metadata_transfer_attempts.set(0);
}
+ /// A usable repair frame landed in the window: restart the stall clock.
+ ///
+ /// A window is served in `REPAIR_CHUNK_MAX` slices and nothing else resets
+ /// `idle_ticks`, so a healthy multi-chunk stream would cross the retry
interval
+ /// on its own and rotate off a peer that is answering.
+ ///
+ /// Clock only. A repair prepare carries no sender and no session nonce
(the
+ /// frame IS the stored prepare, and its identity checksum covers every
byte
+ /// that could hold one), so a peer this session already rotated away from
can
+ /// land in-flight frames here and be credited to its successor. On the
clock
+ /// that costs one retry interval and is bounded, since nothing
re-requests from
+ /// that peer. On the budget it would cost rotation itself. See
+ /// [`Self::note_metadata_repair_walked`].
+ fn note_metadata_repair_clock(&self) {
+ if let Some(session) = self.metadata_repair.borrow_mut().as_mut() {
+ session.idle_ticks = 0;
+ }
+ }
+
+ /// The repair this session asked for is landing: restart the clock and the
+ /// budget.
+ ///
+ /// Takes only attributable signals, which a bare frame is not.
Terminators are
+ /// fenced on `session.nonce` before reaching here, so they came from the
+ /// targeted peer; an advanced `commit_min` is the gap actually closing,
+ /// whoever supplied the bytes.
+ ///
+ /// Stricter than the "any frame" rule it replaced: a re-request re-serves
ops
+ /// already held, so a peer answering with nothing new used to clear its
own
+ /// budget and could never be rotated away from.
+ fn note_metadata_repair_walked(&self) {
+ self.metadata_repair_attempts.set((0, 0));
+ self.note_metadata_repair_clock();
+ }
+
/// Charge one decode failure against `snapshot_seq`'s generation; `true`
/// once that generation's budget is spent. A different generation restarts
/// the count: the peer checkpointed since, so the artifacts are new bytes
@@ -7275,13 +7553,17 @@ where
walk_cursor.get_or_insert(namespace);
}
}
- let consensus_normal = partition.consensus().is_normal();
let consensus_view = partition.consensus().view();
let commit_min = partition.consensus().commit_min();
let cluster = partition.consensus().cluster();
let self_id = partition.consensus().replica();
+ // A primary-elect's merged-log session legitimately runs
outside
+ // `Normal` (`request_partition_view_repair`). Same predicate
as the
+ // two ingest sites, so the arming side and the ingest side
cannot
+ // drift apart.
+ let session_live =
consensus::repair_session_live(partition.consensus());
let repair_finished = partition.repair.is_some_and(|session| {
- if !consensus_normal || consensus_view != session.view {
+ if !session_live || consensus_view != session.view {
return true;
}
// Floored at the LIVE commit point, like
`complete_repair`:
@@ -7313,7 +7595,7 @@ where
continue;
}
let due = partition.repair.as_mut().and_then(|session| {
- if !consensus_normal {
+ if !session_live {
return None;
}
session.idle_ticks += 1;
@@ -7362,6 +7644,20 @@ where
);
partition.repair = None;
partition.note_repair_progress();
+ // A parked view change re-arms from the merged log's
senders,
+ // not around the cluster ring:
+ // `maybe_request_partition_repair` refuses outside
`Normal`,
+ // and a replica that never named the op answers
`RangeEvicted`
+ // for a range it never held.
+ if partition.consensus().view_log_is_pending()
+ && partition
+ .consensus()
+ .is_primary_for_view(partition.consensus().view())
+ {
+ self.request_partition_view_repair(partition, from_op,
to_op, Some(peer))
+ .await;
+ continue;
+ }
if next_peer == peer {
// The ring had nobody else to offer (a solo group, or
a
// two-replica group whose only peer is the one that
@@ -8582,6 +8878,82 @@ where
true
}
+ /// Repair a primary-elect's merged log before it starts the view.
+ ///
+ /// Sibling of [`Self::maybe_request_partition_repair`], which refuses
outside
+ /// `Normal` because its window comes from the live commit frontier. This
window
+ /// comes from the parked merged log, so it runs in `ViewChange` for the
replica
+ /// that parked it. Without it the coverage scan in
+ /// [`Self::start_pending_partition_view`] reports an op nothing ever
fetches --
+ /// the sweep's gap detector needs `probe.normal` too -- and only the
+ /// view-change timeout moves the replica.
+ ///
+ /// `avoid` is the peer a stall just gave up on, so rotation lands on a
+ /// different sender instead of the head of the same list.
+ #[allow(clippy::future_not_send)]
+ async fn request_partition_view_repair(
+ &self,
+ partition: &mut IggyPartition<B, SB>,
+ from_op: u64,
+ to_op: u64,
+ avoid: Option<u8>,
+ ) where
+ B: MessageBus,
+ {
+ if partition.repair.is_some() || from_op > to_op {
+ return;
+ }
+ if self.partition_repairs_inflight.get() >=
PARTITION_REPAIRS_INFLIGHT_MAX {
+ return;
+ }
+ let consensus = partition.consensus();
+ let sources = view_repair_sources(consensus, from_op);
+ let Some(peer) = next_view_repair_peer(&sources, avoid) else {
+ // Nobody else named this op. The view-change timeout escalates;
+ // re-arming the same silent sender would pin the scan behind a
+ // session for nothing.
+ tracing::warn!(
+ shard = self.id,
+ namespace_raw = consensus.group(),
+ from_op,
+ to_op,
+ "no replica offers op {from_op} for the merged partition log;
view change is \
+ stalled"
+ );
+ return;
+ };
+ let nonce = iggy_common::random_id::get_uuid();
+ let cluster = consensus.cluster();
+ let self_id = consensus.replica();
+ let namespace = consensus.group();
+ let view = consensus.view();
+ self.partition_repairs_inflight
+ .set(self.partition_repairs_inflight.get() + 1);
+ partition.repair = Some(partitions::RepairSession {
+ nonce,
+ view,
+ // Merged-log numbers, not the live frontier: `commit_to_op` is
what
+ // the walk must reach to finish the session, `fetch_to_op` the
head
+ // the view will announce.
+ commit_to_op: pending_commit_max(consensus),
+ fetch_to_op: to_op,
+ floor: None,
+ peer,
+ first_batch_offset: None,
+ idle_ticks: 0,
+ });
+ tracing::info!(
+ shard = self.id,
+ namespace_raw = namespace,
+ from_op,
+ to_op,
+ peer,
+ "repairing toward the merged partition log before starting the
view"
+ );
+ self.send_request_prepares(cluster, self_id, peer, nonce, from_op,
to_op, namespace)
+ .await;
+ }
+
/// Receiver side of a partition descriptor: accept the manifest, adopt
/// any reusable staged segments from an earlier attempt, and start
/// pulling, or fall back to journal repair when the peer cannot serve.
@@ -10004,6 +10376,111 @@ where
}
}
+/// Lowest op of a primary-elect's merged log this replica can be held to.
+///
+/// The merged commit point is what the cluster committed, `commit_min` what
this
+/// replica applied. They diverge whenever this replica has not applied the
merged
+/// commit point: a hole in the local prefix, or plain apply lag. Taking the
lower
+/// keeps coverage, repair scope and the stall retry asking about the same ops.
+fn merged_log_scan_floor(pending: &MergedLog, commit_min: u64) -> u64 {
+ pending.commit_max.min(commit_min + 1).max(1)
+}
+
+/// The two floors on a merged-log coverage scan.
+///
+/// Named, not positional: both are `u64`, they sit next to each other,
swapping
+/// them compiles, and the partition site passes the same value for both.
+#[derive(Debug, Clone, Copy)]
+struct ScanFloor {
+ /// Ops at or below this are gone AND already settled: compacted under a
+ /// snapshot (metadata) or at or below the local commit point (partitions).
+ /// Neither can diverge from the merged log and no repair puts the entry
back,
+ /// so demanding one parks the view change forever.
+ repair_floor: u64,
+ /// Highest op this replica has applied. See [`merged_log_scan_floor`].
+ commit_min: u64,
+}
+
+impl ScanFloor {
+ /// Lowest op the scan probes.
+ fn opens_at(self, pending: &MergedLog) -> u64 {
+ merged_log_scan_floor(pending, self.commit_min).max(self.repair_floor
+ 1)
+ }
+}
+
+/// Replicas a primary-elect can ask for `op` while its merged log is parked.
+///
+/// Offered bodies first: those senders proved they hold the entry. A gap below
+/// every sender's commit point has none, since a DVC suffix spans
`commit..=op`
+/// and says nothing underneath, so fall back to senders that committed the op.
+/// They hold it or compacted it, and `RangeEvicted` says which.
+///
+/// Both planes: a partition primary-elect parks a merged log the same way.
+fn view_repair_sources<B, P>(consensus: &VsrConsensus<B, P>, op: u64) ->
Vec<u8>
+where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+{
+ let offered = consensus.pending_view_body_sources(op);
+ if offered.is_empty() {
+ consensus.pending_view_commit_sources(op)
+ } else {
+ offered
+ }
+}
+
+/// Where a stalled repair session reopens its window.
+///
+/// A merged-log session reopens exactly where it was armed. Its floor is
COVERAGE,
+/// not the walk: `first_op_not_covered` reports ops whose journal entry is
absent
+/// or diverging, and an op can be applied (`commit_min` past it) while its
entry is
+/// gone. Raising the floor to `commit_min + 1` there skips the very op the
scan
+/// reported, and the view change parks on it forever.
+///
+/// A tail-repair session is the other way round. Its window IS the commit
gap, so
+/// ops the walk has since consumed must not be asked for again.
`session.from_op`
+/// still floors it, carrying the initial arm's snapshot clamp so no retry
asks for
+/// compacted ops.
+fn stalled_repair_from_op(session_from_op: u64, commit_min: u64,
repairing_view: bool) -> u64 {
+ if repairing_view {
+ session_from_op
+ } else {
+ session_from_op.max(commit_min + 1)
+ }
+}
+
+/// Walk a merged-log source list one step past `avoid`, wrapping.
+///
+/// A ring, not a filter. The list is `log_view`-ordered and identical on every
+/// call, so `find(|c| *c != avoid)` yields the head for every peer but the
head
+/// itself and a third sender is never reached.
+///
+/// `None` when the list is empty or `avoid` is its only entry.
+fn next_view_repair_peer(sources: &[u8], avoid: Option<u8>) -> Option<u8> {
+ let Some(avoid) = avoid else {
+ return sources.first().copied();
+ };
+ let Some(index) = sources.iter().position(|candidate| *candidate == avoid)
else {
+ // The peer that stalled is not in this list at all (the DVC quorum
moved
+ // under it), so nothing has been tried yet from where we now stand.
+ return sources.first().copied();
+ };
+ let next = sources[(index + 1) % sources.len()];
+ if next == avoid { None } else { Some(next) }
+}
+
+/// The merged log's commit point while a view change is parked; the live
frontier
+/// otherwise.
+fn pending_commit_max<B, P>(consensus: &VsrConsensus<B, P>) -> u64
+where
+ B: MessageBus,
+ P: Pipeline<Entry = consensus::PipelineEntry>,
+{
+ consensus
+ .with_pending_view_log(|pending| pending.commit_max)
+ .unwrap_or_else(|| consensus.commit_max())
+}
+
/// Whether a repaired prepare at `op` falls inside the range this replica is
/// currently repairing.
///
@@ -10024,7 +10501,7 @@ fn repair_op_in_scope(
pending
.filter(|_| is_primary_elect)
.map_or(op > commit_min, |pending| {
- (op >= pending.commit_max.max(1) && op <= pending.op_head)
+ (op >= merged_log_scan_floor(pending, commit_min) && op <=
pending.op_head)
|| pending
.committed_elsewhere
.iter()
@@ -10723,13 +11200,29 @@ const fn header_is_view_entry(local: &PrepareHeader,
canonical: &PrepareHeader)
/// repair cannot walk back to them. `repair_floor` drops the ops whose
journal entry
/// is legitimately gone AND whose identity is already settled: on the
metadata plane
/// ops compacted under a snapshot, on the partition plane ops at or below the
local
-/// commit point, whose flushed entries `evict_prefix` moves out of the header
vec.
-/// Neither can diverge from the merged log (a committed or compacted op is the
-/// quorum's op), and no repair puts the journal entry back, so demanding one
parks
-/// the view change forever.
+/// commit point. Neither can diverge from the merged log (a committed or
compacted
+/// op is the quorum's op), and no repair puts the journal entry back, so
demanding
+/// one parks the view change forever.
+///
+/// Not a residency bound. `evict_prefix` clears the header vec up to
`commit_max`
+/// (the cluster frontier), so ops above `repair_floor` can be non-resident and
+/// still serveable, from the evicted ring or the flushed segments. Callers
pass a
+/// `header_at` that reads both.
+///
+/// Opens at [`merged_log_scan_floor`]: the merged commit point alone would
declare
+/// the log serveable over a local gap, promoting a replica whose
`CommitJournal`
+/// gap-stops below where `RebuildPipeline` seeds.
+///
+/// Below the merged commit point, identity rests on the fault model rather
than on
+/// this scan. The merged log names headers only from the DVC suffixes, which
span
+/// `commit..=op` per sender, so an op the widened floor admits under
+/// `pending.commit_max` usually has no canonical header and `held` degrades to
+/// bare residency. Sound under crash-stop, where a committed op is the
quorum's
+/// op. Not a Byzantine or bit-rot guard: corruption is
`verify_prepare_integrity`'s
+/// job on the ingest side.
fn first_op_not_covered(
pending: &MergedLog,
- repair_floor: u64,
+ floor: ScanFloor,
header_at: impl Fn(u64) -> Option<PrepareHeader>,
) -> Option<u64> {
let held = |op: u64| {
@@ -10743,14 +11236,19 @@ fn first_op_not_covered(
.find(|header| header.op == op)
.is_none_or(|canonical| header_is_view_entry(&local, canonical))
};
- (pending.commit_max.max(1).max(repair_floor + 1)..=pending.op_head)
+ (floor.opens_at(pending)..=pending.op_head)
.find(|op| !held(*op))
.or_else(|| {
+ // NOT raised to `opens_at`: these ops sit outside the merged
window
+ // by construction, and dropping the ones below it would start the
+ // view over a committed op this replica cannot serve. The repair
+ // window is floored to match instead, via
+ // `MetadataRepairSession::from_op`.
pending
.committed_elsewhere
.iter()
.map(|header| header.op)
- .filter(|op| *op > repair_floor)
+ .filter(|op| *op > floor.repair_floor)
.find(|op| !held(*op))
})
}
@@ -11346,9 +11844,17 @@ mod repair_scope_tests {
mod view_coverage_tests {
//! Holding an op is not holding the view's op.
- use super::{MergedLog, first_op_not_covered};
+ use super::{MergedLog, ScanFloor, first_op_not_covered};
use iggy_binary_protocol::{Command, Operation, PrepareHeader};
+ /// What a caught-up replica passes: nothing compacted, nothing lagging.
+ fn caught_up(pending: &MergedLog) -> ScanFloor {
+ ScanFloor {
+ repair_floor: 0,
+ commit_min: pending.commit_max,
+ }
+ }
+
fn sealed(op: u64, request: u64) -> PrepareHeader {
let mut header = PrepareHeader {
command: Command::Prepare,
@@ -11373,7 +11879,7 @@ mod view_coverage_tests {
committed_elsewhere: Vec::new(),
};
let held = [sealed(100, 1), sealed(99, 7), sealed(98, 1)];
- let missing = first_op_not_covered(&pending, 0, |op| {
+ let missing = first_op_not_covered(&pending, caught_up(&pending), |op|
{
held.iter().find(|header| header.op == op).copied()
});
assert_eq!(missing, Some(99));
@@ -11397,16 +11903,187 @@ mod view_coverage_tests {
};
let nothing_resident = |_: u64| None;
assert_eq!(
- first_op_not_covered(&pending, 0, nothing_resident),
+ first_op_not_covered(&pending, caught_up(&pending),
nothing_resident),
Some(256),
"unfloored, the evicted committed op reads as an unfillable hole"
);
assert_eq!(
- first_op_not_covered(&pending, 256, nothing_resident),
+ first_op_not_covered(
+ &pending,
+ ScanFloor {
+ repair_floor: 256,
+ commit_min: 256,
+ },
+ nothing_resident
+ ),
None,
"floored at the local commit point, the view starts"
);
}
+
+ #[test]
+ fn
given_a_hole_below_the_merged_commit_point_when_scanning_should_report_it() {
+ // Missed op 7 and kept taking prepares above it: the cluster committed
+ // through 10 while this state machine stopped at 6. From the merged
commit
+ // point the view would start over the gap and the first quorum ack
would
+ // apply an op with 7..=10 never executed locally.
+ let pending = MergedLog {
+ op_head: 12,
+ commit_max: 10,
+ headers: (7..=12).rev().map(|op| sealed(op, 1)).collect(),
+ committed_elsewhere: Vec::new(),
+ };
+ let held: Vec<_> = (8..=12).map(|op| sealed(op, 1)).collect();
+ let missing = first_op_not_covered(
+ &pending,
+ ScanFloor {
+ repair_floor: 0,
+ commit_min: 6,
+ },
+ |op| held.iter().find(|header| header.op == op).copied(),
+ );
+ assert_eq!(
+ missing,
+ Some(7),
+ "a hole below the merged commit point must park the view change"
+ );
+ }
+
+ #[test]
+ fn
given_a_contiguous_prefix_when_scanning_should_open_at_the_merged_commit_point()
{
+ // Nothing missing below, so both bounds coincide. Op 9 is held but is
not
+ // the view's op 9, so the commit point itself is still
identity-checked.
+ let pending = MergedLog {
+ op_head: 12,
+ commit_max: 9,
+ headers: (9..=12).rev().map(|op| sealed(op, 1)).collect(),
+ committed_elsewhere: Vec::new(),
+ };
+ let held: Vec<_> = (9..=12)
+ .map(|op| sealed(op, if op == 9 { 7 } else { 1 }))
+ .collect();
+ let missing = first_op_not_covered(&pending, caught_up(&pending), |op|
{
+ held.iter().find(|header| header.op == op).copied()
+ });
+ assert_eq!(
+ missing,
+ Some(9),
+ "the merged commit point stays in scope when the prefix is
contiguous"
+ );
+ }
+
+ #[test]
+ fn given_a_held_run_below_the_hole_when_scanning_should_walk_to_the_hole()
{
+ // The span the widened floor buys, and the one that costs: the open
sits
+ // well below the merged commit point, the ops between are all held,
and the
+ // scan must walk them to reach 15. A scan that stopped on its first
probe
+ // would find op 4 covered and never look further.
+ let pending = MergedLog {
+ op_head: 22,
+ commit_max: 20,
+ headers: (4..=22).rev().map(|op| sealed(op, 1)).collect(),
+ committed_elsewhere: Vec::new(),
+ };
+ let held: Vec<_> = (4..=22)
+ .filter(|op| *op != 15)
+ .map(|op| sealed(op, 1))
+ .collect();
+ let missing = first_op_not_covered(
+ &pending,
+ ScanFloor {
+ repair_floor: 0,
+ commit_min: 3,
+ },
+ |op| held.iter().find(|header| header.op == op).copied(),
+ );
+ assert_eq!(
+ missing,
+ Some(15),
+ "the scan must walk the held run below the merged commit point,
not stop at its first covered probe"
+ );
+ }
+
+ #[test]
+ fn
given_a_committed_elsewhere_op_below_the_open_when_scanning_should_still_report_it()
{
+ // `committed_elsewhere` sits outside the merged window, so the
fallback is
+ // floored at `repair_floor` and not at the scan's open. Which is why
the
+ // repair window floors at the op reported here: a retry reopening at
+ // `opens_at` would skip op 5 forever.
+ let pending = MergedLog {
+ op_head: 12,
+ commit_max: 10,
+ headers: (7..=12).rev().map(|op| sealed(op, 1)).collect(),
+ committed_elsewhere: vec![sealed(5, 1)],
+ };
+ let held: Vec<_> = (7..=12).map(|op| sealed(op, 1)).collect();
+ let floor = ScanFloor {
+ repair_floor: 0,
+ commit_min: 6,
+ };
+ assert_eq!(floor.opens_at(&pending), 7);
+ let missing = first_op_not_covered(&pending, floor, |op| {
+ held.iter().find(|header| header.op == op).copied()
+ });
+ assert_eq!(
+ missing,
+ Some(5),
+ "an op committed elsewhere and below the open is still uncovered"
+ );
+ }
+
+ #[test]
+ fn given_a_hole_when_scoping_repair_should_admit_the_missing_op() {
+ // Coverage and scope must agree: the scan parks on op 7, so op 7's
repaired
+ // prepare must be ingested. From the merged commit point it would be
+ // requested and then refused.
+ let pending = MergedLog {
+ op_head: 12,
+ commit_max: 10,
+ headers: (7..=12).rev().map(|op| sealed(op, 1)).collect(),
+ committed_elsewhere: Vec::new(),
+ };
+ assert!(
+ super::repair_op_in_scope(Some(&pending), true, 6, 7),
+ "the op the coverage scan parked on must be in repair scope"
+ );
+ }
+
+ #[test]
+ fn given_a_source_list_when_rotating_should_walk_it_as_a_ring() {
+ use super::next_view_repair_peer;
+
+ let sources = [1u8, 2, 3];
+ assert_eq!(next_view_repair_peer(&sources, None), Some(1));
+ assert_eq!(next_view_repair_peer(&sources, Some(1)), Some(2));
+ assert_eq!(
+ next_view_repair_peer(&sources, Some(2)),
+ Some(3),
+ "a filter answers 1 here and never reaches the third sender"
+ );
+ assert_eq!(
+ next_view_repair_peer(&sources, Some(3)),
+ Some(1),
+ "the walk wraps"
+ );
+ }
+
+ #[test]
+ fn given_a_sole_or_absent_source_when_rotating_should_report_nobody_left()
{
+ use super::next_view_repair_peer;
+
+ assert_eq!(next_view_repair_peer(&[], None), None);
+ assert_eq!(next_view_repair_peer(&[], Some(1)), None);
+ assert_eq!(
+ next_view_repair_peer(&[1], Some(1)),
+ None,
+ "the only sender is the one that went quiet"
+ );
+ assert_eq!(
+ next_view_repair_peer(&[2, 3], Some(9)),
+ Some(2),
+ "a peer no longer in the list means nothing here has been tried
yet"
+ );
+ }
}
#[cfg(test)]
@@ -12286,13 +12963,14 @@ mod metadata_repair_session_tests {
use super::{
MetadataRepairSession, gap_repair_peer, metadata_repair_superseded,
next_transfer_peer,
- repair_chunk_walked,
+ repair_chunk_walked, stalled_repair_from_op,
};
- /// Armed at view 3, against a window ending at op 20.
+ /// Armed at view 3, against the window `11..=20`.
const fn session() -> MetadataRepairSession {
MetadataRepairSession {
nonce: 7,
+ from_op: 11,
to_op: 20,
view: 3,
peer: 0,
@@ -12300,6 +12978,31 @@ mod metadata_repair_session_tests {
}
}
+ /// A merged-log session must re-ask for the op the coverage scan
reported, even
+ /// once the walk has passed it. Coverage is about the journal ENTRY; an
op can
+ /// be applied and still have no entry to serve, which is exactly what
+ /// `committed_elsewhere` reports.
+ #[test]
+ fn
given_a_dropped_response_below_commit_min_when_retrying_should_still_ask_for_it()
{
+ assert_eq!(
+ stalled_repair_from_op(5, 6, true),
+ 5,
+ "clamping to commit_min + 1 would retry from 7 and skip the
reported hole"
+ );
+ }
+
+ /// The tail-repair session is the other way round: its window is the
commit gap,
+ /// so ops the walk consumed must not be re-requested.
+ #[test]
+ fn
given_a_walked_window_when_retrying_a_tail_session_should_open_above_it() {
+ assert_eq!(stalled_repair_from_op(5, 6, false), 7);
+ assert_eq!(
+ stalled_repair_from_op(11, 3, false),
+ 11,
+ "the arm floor still holds, so no retry asks for compacted ops"
+ );
+ }
+
#[test]
fn given_a_gap_stopped_backup_when_picking_a_peer_should_ask_the_primary()
{
assert_eq!(gap_repair_peer(2, 3, 0), Some(0));
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index b7465d2b7..1e46ec9c9 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -532,6 +532,15 @@ where
})
}
+ /// `first_partition_commit_fault` for the simulator's lost-wakeup
+ /// tripwire: a fenced pump and a missed wake both leave frames undrained,
and
+ /// only the second is a channel bug. Test/simulator only, like
`inbox_len`.
+ #[cfg(any(test, feature = "simulator"))]
+ #[must_use]
+ pub fn fenced_partition_fault(&self) -> Option<FatalCommit> {
+ self.first_partition_commit_fault()
+ }
+
/// Sanity check at pump entry: every Consensus frame routed through
/// [`Self::dispatch`] must land on the shard whose `id` matches the
/// `target_shard` the sender stamped on the frame. The ctor
diff --git a/core/simulator/src/bin/workload-fuzz.rs
b/core/simulator/src/bin/workload-fuzz.rs
index 8143016a7..fb6ab966a 100644
--- a/core/simulator/src/bin/workload-fuzz.rs
+++ b/core/simulator/src/bin/workload-fuzz.rs
@@ -132,11 +132,21 @@ struct Args {
/// empty shadow against empty committed state and agrees. `0` opts out.
#[arg(long, default_value_t = 1)]
min_commits: u64,
- /// Committed metadata ops that must have been witnessed by more than one
live
- /// replica, i.e. that exercised cross-replica agreement. Ignored below
two live
- /// replicas, where the property is untestable rather than untested. `0`
opts out.
+ /// Committed ops, on EITHER plane, that must have been witnessed by more
than
+ /// one live replica, i.e. that exercised cross-replica agreement. Ignored
below
+ /// two live replicas, where the property is untestable rather than
untested.
+ /// `0` opts out.
#[arg(long, default_value_t = 1)]
min_ops_compared: usize,
+ /// As `--min-ops-compared`, but METADATA ops only.
+ ///
+ /// Separate because a partition-only run satisfies the combined floor
while the
+ /// metadata oracle compares an empty chain against an empty chain and
agrees.
+ /// Defaults to `1`, which is the floor `--min-ops-compared` carried
before it
+ /// counted both planes; a campaign that wants no metadata coverage opts
out
+ /// with `0`.
+ #[arg(long, default_value_t = 1)]
+ min_metadata_ops_compared: usize,
/// Fail the run if crash or restart injection was requested but never
happened.
/// Off by default, since a short run at low probability may legitimately
draw
/// none; on for a campaign where such a seed is silently wasted.
@@ -562,9 +572,11 @@ fn run_quiesce_phase(
};
println!(
"quiesced and converged (leader-relative; entity oracle:
{entity_oracle}; \
- evictions={}; ops_compared={} replicas_compared={}
namespaces_checked={})",
+ evictions={}; ops_compared={} partition_ops_compared={}
replicas_compared={} \
+ namespaces_checked={})",
workload.evictions(),
convergence.ops_compared,
+ convergence.partition_ops_compared,
convergence.replicas_compared,
convergence.namespaces_checked,
);
@@ -574,13 +586,29 @@ fn run_quiesce_phase(
proved nothing about entity state (seed={seed:#x})"
);
let live = usize::from(replicas) - sim.crashed.len();
+ // Either plane satisfies it: a partition-plane run commits almost no
metadata,
+ // so the metadata count alone called every such run vacuous.
+ let compared = convergence.ops_compared +
convergence.partition_ops_compared;
assert!(
- args.min_ops_compared == 0 || live < 2 || convergence.ops_compared >=
args.min_ops_compared,
- "--min-ops-compared {}: {live} replicas live but only {} op(s)
witnessed \
- by more than one, so cross-replica agreement went untested \
- (seed={seed:#x})",
+ args.min_ops_compared == 0 || live < 2 || compared >=
args.min_ops_compared,
+ "--min-ops-compared {}: {live} replicas live but only {compared} op(s)
witnessed \
+ by more than one ({} metadata, {} partition), so cross-replica
agreement went \
+ untested (seed={seed:#x})",
args.min_ops_compared,
convergence.ops_compared,
+ convergence.partition_ops_compared,
+ );
+ // The metadata half on its own: summing the planes above lets a
partition-only
+ // run clear that floor while the metadata oracle compares nothing.
+ assert!(
+ args.min_metadata_ops_compared == 0
+ || live < 2
+ || convergence.ops_compared >= args.min_metadata_ops_compared,
+ "--min-metadata-ops-compared {}: {live} replicas live but only {}
committed metadata \
+ op(s) witnessed by more than one, so the metadata oracle compared an
empty chain \
+ (seed={seed:#x})",
+ args.min_metadata_ops_compared,
+ convergence.ops_compared,
);
// Again after the drain: the drain both answers outstanding requests and
// issues its own resends, so the pre-drain numbers are not the final ones.
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index d5c554023..51a2d3559 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -33,7 +33,7 @@ use deps::SimClock;
use deps::SimSuperblock;
use deps::{MemStorage, SimJournal};
use executor::{DetExecutor, RunOutcome, TaskId};
-use iggy_binary_protocol::{Command, GenericHeader, ReplyHeader};
+use iggy_binary_protocol::{Command, GenericHeader, PrepareHeader, ReplyHeader};
use iggy_common::IggyError;
use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
use metadata::impls::metadata::StreamsFrontend;
@@ -53,7 +53,7 @@ use server_common::sharding::{IggyNamespace,
PartitionLocation, ShardId};
use shard::shards_table::{ShardsTable, calculate_shard_assignment};
use shard::{CONSENSUS_TICK_INTERVAL, PartitionMaterialisation};
use std::cell::RefCell;
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::rc::Rc;
use std::sync::Arc;
@@ -135,6 +135,54 @@ pub(crate) struct PartitionConsensusState {
pub commit_min: u64,
}
+/// A pipeline head the commit walk is holding on: covered by the commit
frontier,
+/// but not the op the state machine is next owed.
+///
+/// What `drain_committable_prefix` / `peek_committable_head` refuse to drain.
Two
+/// different faults share that refusal and need different thresholds, so
+/// [`CommitPrefixHole::kind`] keeps them apart.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct CommitPrefixHole {
+ pub head_op: u64,
+ pub commit_min: u64,
+ pub commit_max: u64,
+ pub kind: CommitHoldKind,
+}
+
+/// Why a commit walk is holding below its pipeline head.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub(crate) enum CommitHoldKind {
+ /// `head_op > commit_min + 1`: the ops between never arrived. Legitimate
and
+ /// transient right after a promotion, while the bounded journal walk
clears the
+ /// apply backlog `RebuildPipeline` seeded above. Repair clears it, and
+ /// `commit_min` climbing is the proof repair is working.
+ MissingOps,
+ /// `head_op <= commit_min`: the walk applied this op and advanced past a
+ /// still-resident entry. Nothing missing, no repair owed, only a pop that
can
+ /// no longer happen -- so it never self-clears, and `commit_min` keeps
climbing
+ /// while the head stays frozen.
+ AppliedHead,
+}
+
+impl CommitPrefixHole {
+ fn read(head: Option<PrepareHeader>, commit_min: u64, commit_max: u64) ->
Option<Self> {
+ let head = head?;
+ if head.op > commit_max || head.op == commit_min + 1 {
+ return None;
+ }
+ Some(Self {
+ head_op: head.op,
+ commit_min,
+ commit_max,
+ kind: if head.op > commit_min {
+ CommitHoldKind::MissingOps
+ } else {
+ CommitHoldKind::AppliedHead
+ },
+ })
+ }
+}
+
pub struct Simulator {
/// All replicas, indexed by replica id. Always fully populated; crashed
replicas
/// stay alive but are skipped during dispatch.
@@ -966,6 +1014,21 @@ impl Simulator {
continue;
}
for shard in &replica.shards {
+ // First and unconditional. A fenced pump has exited, so its
frames
+ // pile up exactly as a missed wake does and every lane assert
below
+ // would misreport the cause -- and the pump's `FatalCommit`
return
+ // is dropped at the spawn, so nothing else sees it. Gated on a
+ // non-empty inbox, a fenced pump that happened to drain would
pass
+ // quiescence outright.
+ assert!(
+ shard.fenced_partition_fault().is_none(),
+ "fenced pump: replica {replica_id} shard {} exited on a
fatal commit ({:?}). \
+ Not a lost wakeup; fix the commit failure (seed {:#x},
schedule hash {:#x})",
+ shard.id,
+ shard.fenced_partition_fault(),
+ self.seed,
+ self.executor.schedule_hash(),
+ );
let pending = shard.inbox_len();
assert_eq!(
pending,
@@ -1364,6 +1427,30 @@ impl Simulator {
Some(partition.offsets())
}
+ /// A replica's journaled partition-plane prepare headers over `ops`, or
`None`
+ /// when it does not host the namespace.
+ ///
+ /// Repair headers, not resident ones. `evict_prefix` clears the resident
vec as
+ /// the committed prefix flushes to segments and moves those entries to the
+ /// repair ring, so a resident-only read compares nothing at all once a
run has
+ /// flushed. The ring is capacity-bounded, which makes this the recently
+ /// committed tail rather than the whole prefix -- and that tail is where
a bad
+ /// repair or a mis-decided view change lands.
+ ///
+ /// One pass per replica per namespace, not one per op: both lookups behind
+ /// `repair_headers_in` are linear.
+ #[must_use]
+ pub(crate) fn partition_journaled_headers(
+ &self,
+ replica_idx: usize,
+ namespace: IggyNamespace,
+ ops: std::ops::RangeInclusive<u64>,
+ ) -> Option<BTreeMap<u64, PrepareHeader>> {
+ let shard = self.replicas[replica_idx].partition_shard(namespace);
+ let partition = shard.plane.partitions().get_by_ns(&namespace)?;
+ Some(partition.log.journal().inner.repair_headers_in(ops))
+ }
+
/// Consensus view for a replica's partition-plane group, or `None` if that
/// replica does not host the namespace.
#[must_use]
@@ -1398,6 +1485,53 @@ impl Simulator {
})
}
+ /// A replica's metadata consensus handle, or `None` when it hosts no
metadata
+ /// plane. The one way to reach it; do not hand-walk the shard-0 / plane /
+ /// metadata chain.
+ #[must_use]
+ pub(crate) fn metadata_consensus(
+ &self,
+ replica_idx: usize,
+ ) -> Option<&consensus::VsrConsensus<crate::bus::SharedSimOutbox>> {
+ self.replicas[replica_idx].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ }
+
+ /// The metadata pipeline head the commit walk is holding on, if any. See
+ /// [`CommitPrefixHole`].
+ #[must_use]
+ pub(crate) fn metadata_commit_prefix_hole(
+ &self,
+ replica_idx: usize,
+ ) -> Option<CommitPrefixHole> {
+ let consensus = self.metadata_consensus(replica_idx)?;
+ CommitPrefixHole::read(
+ consensus.pipeline_head_header(),
+ consensus.commit_min(),
+ consensus.commit_max(),
+ )
+ }
+
+ /// Partition-plane twin of [`Self::metadata_commit_prefix_hole`].
+ #[must_use]
+ pub(crate) fn partition_commit_prefix_hole(
+ &self,
+ replica_idx: usize,
+ namespace: IggyNamespace,
+ ) -> Option<CommitPrefixHole> {
+ let shard = self.replicas[replica_idx].partition_shard(namespace);
+ let partition = shard.plane.partitions().get_by_ns(&namespace)?;
+ let consensus = partition.consensus();
+ CommitPrefixHole::read(
+ consensus.pipeline_head_header(),
+ consensus.commit_min(),
+ consensus.commit_max(),
+ )
+ }
+
/// Index of the current primary for `namespace`, as seen by the first live
/// replica hosting it, or `None` if no live replica hosts it.
///
diff --git a/core/simulator/src/workload/invariants.rs
b/core/simulator/src/workload/invariants.rs
index 4f6dd3a1d..d2011bcb5 100644
--- a/core/simulator/src/workload/invariants.rs
+++ b/core/simulator/src/workload/invariants.rs
@@ -23,18 +23,49 @@
//! trace and the determinism baseline (`workload_replay_is_deterministic`)
//! unchanged.
-use crate::Simulator;
use crate::workload::state_checker::StateChecker;
use crate::workload::{CLIENT_REQUEST_QUEUE_MAX, Workload};
+use crate::{CommitHoldKind, CommitPrefixHole, Simulator};
+use consensus::Consensus;
use server_common::sharding::IggyNamespace;
use std::collections::HashMap;
+/// Ticks a `Normal` metadata primary may sit behind its own recovery barrier
+/// before the run is called wedged.
+///
+/// A primary below the barrier admits nothing. Transient while a resumed
primary
+/// re-pipelines its suffix, a handful of round trips, so two orders of
magnitude
+/// of slack: only a barrier nothing will ever lower trips it.
+const RECOVERY_BARRIER_WEDGE_TICKS: u32 = 2_000;
+
+/// Ticks a replica may hold its commit walk below a MISSING op.
+///
+/// A promotion holds here legitimately while the journal walk clears its apply
+/// backlog 64 ops per sweep, so this sits past any backlog a run generates.
The
+/// count also resets when `commit_min` advances, which is repair working.
+const COMMIT_PREFIX_HOLE_WEDGE_TICKS: u32 = 2_000;
+
+/// Ticks a replica may hold its commit walk below an ALREADY APPLIED head.
+///
+/// Reachable without a bug: a `set_commit_floor` jump can raise `commit_min`
past
+/// a live pipeline. Nothing pops a head below the floor, so only a view change
+/// clears it, and this has to outlast the view-change escalation to avoid
failing
+/// a run on the window in between.
+const COMMIT_PREFIX_APPLIED_HEAD_TICKS: u32 = 2_000;
+
/// Per-(replica, namespace) high-water marks carried across ticks so each new
/// reading can be compared against the last.
#[derive(Debug, Default)]
pub struct Invariants {
commit_offset: HashMap<(u8, IggyNamespace), u64>,
view: HashMap<(u8, IggyNamespace), u64>,
+ /// Consecutive ticks a replica has been a `Normal` metadata primary still
+ /// gated by its recovery barrier. Reset as soon as any of that stops
holding.
+ barrier_gated_ticks: HashMap<u8, u32>,
+ /// Consecutive ticks a plane's commit walk has held below a committable
+ /// pipeline head, keyed by replica and namespace (`None` = metadata),
paired
+ /// with the `commit_min` the count started from.
+ commit_hole_ticks: HashMap<(u8, Option<IggyNamespace>), (u32, u64)>,
/// Cross-replica committed-log agreement. Runs every tick like the rest,
so a
/// divergence is reported where it appears rather than at the next
quiesce.
state_checker: StateChecker,
@@ -71,9 +102,25 @@ impl Invariants {
for replica_idx in 0..sim.replica_count {
if sim.is_crashed(replica_idx) {
+ self.barrier_gated_ticks.remove(&replica_idx);
+ self.commit_hole_ticks
+ .retain(|(replica, _), _| *replica != replica_idx);
continue;
}
+ self.check_recovery_barrier(sim, seed, replica_idx);
+ self.check_commit_prefix_contiguity(
+ seed,
+ replica_idx,
+ None,
+ sim.metadata_commit_prefix_hole(usize::from(replica_idx)),
+ );
for &ns in &workload.options.namespaces {
+ self.check_commit_prefix_contiguity(
+ seed,
+ replica_idx,
+ Some(ns),
+ sim.partition_commit_prefix_hole(usize::from(replica_idx),
ns),
+ );
if let Some(offsets) = sim.offsets(usize::from(replica_idx),
ns) {
let cur = offsets.commit_offset;
if let Some(&prev) = self.commit_offset.get(&(replica_idx,
ns)) {
@@ -102,6 +149,121 @@ impl Invariants {
self.state_checker.check(sim, seed);
}
+ /// Catch a metadata primary permanently shut behind its own recovery
barrier.
+ ///
+ /// The shape `VsrConsensus::redecide_recovery_barrier` fixes, caught from
the
+ /// outside: a primary that can never clear its barrier drops every
request as
+ /// `NotReady`, which otherwise surfaces only as an unexplained stall.
+ ///
+ /// # Panics
+ /// When a `Normal` metadata primary sits below its barrier for
+ /// [`RECOVERY_BARRIER_WEDGE_TICKS`] consecutive ticks.
+ fn check_recovery_barrier(&mut self, sim: &Simulator, seed: u64,
replica_idx: u8) {
+ let Some(consensus) = sim.metadata_consensus(usize::from(replica_idx))
else {
+ return;
+ };
+ // Mirrors `consensus::is_caught_up_primary`, `!is_transferring`
included:
+ // a primary in state transfer is shut by the transfer, not the
barrier, so
+ // blaming the barrier is a false panic.
+ //
+ // Both gates, because they read different counters: admission compares
+ // `commit_max`, the HTTP read gate (`server/src/http/reads.rs`)
compares
+ // `commit_min`. A barrier met by one and never the other 503s every
read
+ // while admission looks open, which is the half
+ // `redecide_recovery_barrier` exists for.
+ let gated = consensus.is_primary()
+ && !consensus.has_ceded_primaryship()
+ && consensus.is_normal()
+ && !consensus.is_transferring()
+ && consensus.recovery_barrier() > 0
+ && (consensus.commit_max() < consensus.recovery_barrier()
+ || consensus.commit_min() < consensus.recovery_barrier());
+ if !gated {
+ self.barrier_gated_ticks.remove(&replica_idx);
+ return;
+ }
+ let ticks = self
+ .barrier_gated_ticks
+ .entry(replica_idx)
+ .and_modify(|ticks| *ticks += 1)
+ .or_insert(1);
+ assert!(
+ *ticks < RECOVERY_BARRIER_WEDGE_TICKS,
+ "replica {replica_idx} has been a Normal metadata primary gated by
its recovery \
+ barrier for {ticks} ticks: barrier={} commit={}..{} view={}.
Nothing lowers the \
+ barrier, so this primary drops every client request (commit_max
gate) or 503s \
+ every local read (commit_min gate) from here on (seed={seed:#x})",
+ consensus.recovery_barrier(),
+ consensus.commit_min(),
+ consensus.commit_max(),
+ consensus.view(),
+ );
+ }
+
+ /// Catch a commit walk permanently held below a committable pipeline head.
+ ///
+ /// The state `drain_committable_prefix` and `peek_committable_head`
refuse to
+ /// drain. Two faults share that refusal and they are not the same wait,
so they
+ /// are counted apart (see [`CommitHoldKind`]):
+ ///
+ /// - [`CommitHoldKind::MissingOps`] is legitimate while repair runs, so
the
+ /// count restarts when `commit_min` advances and only a hole nothing
refills
+ /// trips it.
+ /// - [`CommitHoldKind::AppliedHead`] never self-clears; only a view change
+ /// does. Resetting it on `commit_min` would be exactly wrong, since the
head
+ /// is frozen while `commit_min` climbs.
+ ///
+ /// # Panics
+ /// When one plane holds past the threshold for its hold kind.
+ fn check_commit_prefix_contiguity(
+ &mut self,
+ seed: u64,
+ replica_idx: u8,
+ namespace: Option<IggyNamespace>,
+ hole: Option<CommitPrefixHole>,
+ ) {
+ let key = (replica_idx, namespace);
+ let Some(hole) = hole else {
+ self.commit_hole_ticks.remove(&key);
+ return;
+ };
+ let (limit, diagnosis) = match hole.kind {
+ CommitHoldKind::MissingOps => (
+ COMMIT_PREFIX_HOLE_WEDGE_TICKS,
+ "the ops between are missing and nothing is refilling them, so
every reply \
+ above the hole is owed forever",
+ ),
+ CommitHoldKind::AppliedHead => (
+ COMMIT_PREFIX_APPLIED_HEAD_TICKS,
+ "the commit point moved past this entry, so nothing is missing
and no repair \
+ is owed: only a view change can clear it, and none has",
+ ),
+ };
+ let entry = self
+ .commit_hole_ticks
+ .entry(key)
+ .or_insert((0, hole.commit_min));
+ // A missing-op hold whose `commit_min` moved is repair landing, so
the run
+ // starts over. An applied-head hold is measured while `commit_min`
climbs,
+ // so it must not.
+ if hole.kind == CommitHoldKind::MissingOps && entry.1 !=
hole.commit_min {
+ *entry = (0, hole.commit_min);
+ }
+ entry.0 += 1;
+ let ticks = entry.0;
+ assert!(
+ ticks < limit,
+ "replica {replica_idx} has held its commit walk below a
committable pipeline head \
+ for {ticks} ticks on {} ({:?}): head_op={} commit={}..{}.
{diagnosis} \
+ (seed={seed:#x})",
+ namespace.map_or_else(|| "the metadata plane".to_owned(), |ns|
format!("{ns:?}")),
+ hole.kind,
+ hole.head_op,
+ hole.commit_min,
+ hole.commit_max,
+ );
+ }
+
/// The canonical committed chain built so far. Tests read it to prove the
/// equality check compared replicas against each other rather than passing
/// over an empty chain.
diff --git a/core/simulator/src/workload/oracle.rs
b/core/simulator/src/workload/oracle.rs
index 4c84bdfe6..18b4b714d 100644
--- a/core/simulator/src/workload/oracle.rs
+++ b/core/simulator/src/workload/oracle.rs
@@ -435,10 +435,20 @@ pub fn assert_converged(sim: &Simulator, workload: &mut
Workload) -> Convergence
}
}
+ // Partition-plane twin of the walk above: that one bounds how much each
+ // replica committed, this compares what they committed.
+ let partition_ops_compared =
+ state_checker::assert_partition_prefixes_agree(sim,
&workload.options.namespaces, seed);
+ tracing::info!(
+ partition_ops_compared,
+ "committed partition entries agree across every live replica"
+ );
+
let replicas_compared = assert_committed_metadata_agrees(sim, &live, seed);
let report = ConvergenceReport {
ops_compared,
+ partition_ops_compared,
replicas_compared,
namespaces_checked,
};
@@ -491,6 +501,10 @@ pub fn assert_converged(sim: &Simulator, workload: &mut
Workload) -> Convergence
pub struct ConvergenceReport {
/// Committed metadata ops witnessed by more than one live replica.
pub ops_compared: usize,
+ /// Committed partition ops witnessed by more than one live replica,
summed over
+ /// every namespace. Separate from `ops_compared`: a partition-plane run
commits
+ /// almost no metadata and would otherwise read as having compared nothing.
+ pub partition_ops_compared: usize,
/// Live replicas whose committed metadata CONTENT was compared against a
peer
/// sharing its commit point. Zero on a solo cluster.
pub replicas_compared: usize,
diff --git a/core/simulator/src/workload/state_checker.rs
b/core/simulator/src/workload/state_checker.rs
index dd7e91729..53995a6ec 100644
--- a/core/simulator/src/workload/state_checker.rs
+++ b/core/simulator/src/workload/state_checker.rs
@@ -32,9 +32,9 @@
//! non-vacuous: a chain nothing was compared against passes silently.
use crate::Simulator;
-use consensus::MetadataHandle;
use iggy_binary_protocol::PrepareHeader;
use journal::Journal;
+use server_common::sharding::IggyNamespace;
use std::collections::{BTreeMap, BTreeSet};
/// One op of the canonical committed chain.
@@ -104,7 +104,7 @@ impl StateChecker {
continue;
}
let replica = &sim.replicas[usize::from(replica_idx)];
- let Some(consensus) =
replica.shards[0].plane.metadata().consensus.as_ref() else {
+ let Some(consensus) =
sim.metadata_consensus(usize::from(replica_idx)) else {
continue;
};
let committed = consensus.commit_min();
@@ -254,7 +254,7 @@ pub fn assert_committed_prefixes_agree(sim: &Simulator,
seed: u64) -> usize {
continue;
}
let replica = &sim.replicas[usize::from(replica_idx)];
- let Some(consensus) =
replica.shards[0].plane.metadata().consensus.as_ref() else {
+ let Some(consensus) = sim.metadata_consensus(usize::from(replica_idx))
else {
continue;
};
let committed = consensus.commit_min();
@@ -291,6 +291,71 @@ pub fn assert_committed_prefixes_agree(sim: &Simulator,
seed: u64) -> usize {
witnesses.values().filter(|&&count| count > 1).count()
}
+/// Assert every live replica's committed partition prefix agrees, op for op.
+///
+/// Partition-plane counterpart of [`assert_committed_prefixes_agree`], and
the only
+/// check comparing what replicas committed on this plane rather than how
much. The
+/// quiesce oracle otherwise bounds only that no backup committed past its
leader.
+///
+/// Two differences from the metadata twin, both from the partition journal
evicting
+/// its committed prefix as it flushes to segments:
+///
+/// * A missing header is ordinary, not a hole, so this compares only the ops
two
+/// replicas both still hold. `evict_prefix` moves a flushed entry to the
repair
+/// ring rather than dropping it, so that set is the ring plus whatever is
still
+/// resident: the recently committed tail, where a bad repair or a
mis-decided
+/// view change lands.
+/// * Identity, not the sealed checksum. `restamp_prepare_view` rewrites the
view on
+/// retransmit, so `identity_checksum` compares the entry, not the delivery.
+///
+/// Returns ops witnessed by more than one replica, summed over every
namespace.
+///
+/// # Panics
+/// If two live replicas hold different entries at the same committed
partition op.
+#[must_use]
+pub fn assert_partition_prefixes_agree(
+ sim: &Simulator,
+ namespaces: &[IggyNamespace],
+ seed: u64,
+) -> usize {
+ let mut compared = 0;
+ for &namespace in namespaces {
+ let mut canonical: BTreeMap<u64, (u128, u8)> = BTreeMap::new();
+ let mut witnesses: BTreeMap<u64, usize> = BTreeMap::new();
+ for replica_idx in 0..sim.replica_count {
+ if sim.is_crashed(replica_idx) {
+ continue;
+ }
+ let idx = usize::from(replica_idx);
+ let Some(state) = sim.partition_consensus_state(idx, namespace)
else {
+ continue;
+ };
+ let Some(headers) =
+ sim.partition_journaled_headers(idx, namespace,
1..=state.commit_min)
+ else {
+ continue;
+ };
+ for (op, header) in headers {
+ let identity = header.identity_checksum();
+ if let Some(&(expected, owner)) = canonical.get(&op) {
+ assert_eq!(
+ identity, expected,
+ "at quiesce replica {replica_idx} and replica {owner}
disagree on \
+ committed partition op {op} of ns {namespace:?}:
{identity:#x} vs \
+ {expected:#x} (seed={seed:#x})",
+ );
+ *witnesses.entry(op).or_insert(1) += 1;
+ } else {
+ canonical.insert(op, (identity, replica_idx));
+ witnesses.insert(op, 1);
+ }
+ }
+ }
+ compared += witnesses.values().filter(|&&count| count > 1).count();
+ }
+ compared
+}
+
/// Whether a committed op having no journal header is legitimate rather than a
/// hole.
///
@@ -320,6 +385,7 @@ mod tests {
use super::*;
use crate::client::SimClient;
use crate::packet::PacketSimulatorOptions;
+ use consensus::PartitionsHandle;
const SEED: u64 = 0x5C11;
@@ -359,11 +425,8 @@ mod tests {
#[should_panic(expected = "a hole in the committed log")]
fn a_missing_committed_head_above_the_snapshot_floor_is_a_hole() {
let sim = cluster_with_committed_ops();
- let committed = sim.replicas[1].shards[0]
- .plane
- .metadata()
- .consensus
- .as_ref()
+ let committed = sim
+ .metadata_consensus(1)
.expect("shard 0 owns metadata consensus")
.commit_min();
assert!(
@@ -406,6 +469,97 @@ mod tests {
checker.check(&sim, SEED);
}
+ /// A three-replica cluster with committed partition ops flushed to
segments.
+ ///
+ /// The state the partition comparison has to work in: `evict_prefix`
clears the
+ /// resident header vec on flush, so a resident-only read sees nothing.
+ fn cluster_with_flushed_partition_ops() -> (Simulator, IggyNamespace) {
+
server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+ let client_id: u128 = 1;
+ let mut sim = Simulator::new(
+ 3,
+ std::iter::once(client_id),
+ PacketSimulatorOptions {
+ node_count: 3,
+ client_count: 1,
+ seed: SEED,
+ ..PacketSimulatorOptions::default()
+ },
+ );
+ let client = SimClient::new(client_id);
+ let namespace = IggyNamespace::new(1, 1, 0);
+ sim.init_partition(namespace);
+ sim.register_client_with_primary(&client);
+ for sequence in 0..6u32 {
+ let msg = client.send_messages(
+ namespace,
+ &[bytes::Bytes::from(format!("wl-part-{sequence}"))],
+ );
+ sim.submit_request(client_id, 0, msg.into_generic());
+ for _ in 0..60 {
+ sim.step();
+ }
+ }
+
+ for replica_idx in 0..3usize {
+ let shard = sim.replicas[replica_idx].partition_shard(namespace);
+ let partitions = shard.plane.partitions();
+ let config = partitions.config();
+ let Some(partition) = partitions.get_mut_by_ns(&namespace) else {
+ continue;
+ };
+
futures::executor::block_on(partition.flush_committed_messages(config))
+ .expect("the in-memory flush must succeed");
+ }
+ (sim, namespace)
+ }
+
+ /// The partition comparison must survive a flush.
+ ///
+ /// `header_by_op` reads the resident header vec, which `evict_prefix`
clears as
+ /// the committed prefix flushes, so a resident-only read compares zero
ops at
+ /// quiescence and the check passes over nothing.
+ #[test]
+ fn a_flushed_partition_prefix_is_still_compared() {
+ let (sim, namespace) = cluster_with_flushed_partition_ops();
+
+ let committed = sim
+ .partition_consensus_state(1, namespace)
+ .expect("replica 1 hosts the namespace")
+ .commit_min;
+ assert!(
+ committed > 1,
+ "the cluster committed {committed} partition op(s), so there is
nothing \
+ for two replicas to agree on"
+ );
+
+ let resident = sim.replicas[1]
+ .partition_shard(namespace)
+ .plane
+ .partitions()
+ .get_by_ns(&namespace)
+ .expect("replica 1 hosts the namespace")
+ .log
+ .journal()
+ .inner
+ .header_by_op(1);
+ assert!(
+ resident.is_none(),
+ "op 1 is still resident, so this test does not exercise the
flushed path"
+ );
+
+ let compared = assert_partition_prefixes_agree(&sim, &[namespace],
SEED);
+ assert!(
+ compared > 0,
+ "no committed partition op was witnessed by more than one replica
after \
+ the flush, so the check passed over an empty set"
+ );
+ }
+
/// An undamaged prefix passes the recheck: the reset must turn a restart
into a
/// comparison, not into a failure.
#[test]
