This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch consensus-prefix-contiguity
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/consensus-prefix-contiguity by
this push:
new 0100a561a fix: address review comments round 4
0100a561a is described below
commit 0100a561a864896be356884dcb37cfd50f539fe4
Author: Krishna Vishal <[email protected]>
AuthorDate: Tue Sep 8 22:22:52 2026 +0530
fix: address review comments round 4
---
core/consensus/src/plane_helpers.rs | 38 +++++++++++++----
core/partitions/src/iggy_partition.rs | 12 ++++--
core/shard/src/lib.rs | 69 ++++++++++++++++++++-----------
core/simulator/src/workload/invariants.rs | 16 +++----
4 files changed, 93 insertions(+), 42 deletions(-)
diff --git a/core/consensus/src/plane_helpers.rs
b/core/consensus/src/plane_helpers.rs
index 45951379e..a72995cf0 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -433,6 +433,25 @@ 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 from the head, while covered by the commit frontier,
and
@@ -453,8 +472,9 @@ where
/// `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. Both journal walks stop below
the
-/// head to keep it unreachable, so it is logged apart.
+/// 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).
@@ -560,16 +580,20 @@ fn report_uncommittable_head(
walk until the ops below it are journaled"
);
} else {
- // Unreachable while both journal walks stop below the head, so this
is a
- // defect and not a state to wait out: nothing will pop or answer the
entry.
- tracing::error!(
+ // 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; the
commit walk \
- advanced past a resident pipeline entry and its reply can no
longer be sent"
+ "committable head sits at or below the applied commit point;
nothing can pop \
+ or answer this entry until a view change clears the pipeline"
);
}
}
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index e7d1f9599..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,
@@ -6729,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;
}
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index e408541c8..fb544cf90 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -1607,7 +1607,14 @@ where
/// 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.
- metadata_repair_attempts: Cell<u32>,
+ ///
+ /// 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
@@ -1762,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),
})
}
@@ -2249,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),
}
}
@@ -5293,7 +5300,7 @@ where
// 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() {
+ if self.burn_metadata_repair_attempt(consensus.view())
{
*self.metadata_repair.borrow_mut() = None;
tracing::warn!(
shard = self.id,
@@ -5337,7 +5344,7 @@ where
// 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() {
+ if self.burn_metadata_repair_attempt(consensus.view())
{
self.rotate_stalled_metadata_repair(
consensus,
header.replica,
@@ -5394,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;
}
@@ -5824,7 +5836,7 @@ where
);
*self.metadata_repair.borrow_mut() = None;
self.note_metadata_repair_walked();
- } else if self.burn_metadata_repair_attempt() {
+ } else if self.burn_metadata_repair_attempt(consensus.view()) {
self.rotate_stalled_metadata_repair(consensus, peer, from_op,
to_op)
.await;
} else {
@@ -6173,12 +6185,15 @@ where
// 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.
- if self.metadata_repair_attempts.get() >
partitions::REPAIR_MAX_STALL_RETRIES {
+ //
+ // 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,
- attempts = self.metadata_repair_attempts.get(),
- "merged-log repair is out of retries; not re-arming"
+ view = consensus.view(),
+ "merged-log repair is out of retries for this view; not
re-arming"
);
return;
}
@@ -6828,12 +6843,23 @@ where
/// or the rotation that mints a new one would reset the count and
re-target
/// forever without giving up on a peer. Only
/// [`Self::note_metadata_repair_walked`] clears it.
- fn burn_metadata_repair_attempt(&self) -> bool {
- let attempts = self.metadata_repair_attempts.get() + 1;
- self.metadata_repair_attempts.set(attempts);
+ 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
}
+ /// Whether this view has already spent its merged-log repair budget.
+ 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.
fn burn_metadata_transfer_attempt(&self) -> bool {
let attempts = self.metadata_transfer_attempts.get() + 1;
@@ -6881,7 +6907,7 @@ where
/// 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);
+ self.metadata_repair_attempts.set((0, 0));
self.note_metadata_repair_clock();
}
@@ -7527,20 +7553,15 @@ 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();
- // The one session that legitimately runs outside `Normal`
- // (`request_partition_view_repair`). Treated as superseded it
would
- // be dropped on the tick after it was armed, leaving the
coverage
- // scan with no fetcher; refused a stall retry it would pin
the view
- // change behind one lost frame. Mirrors
- // `retry_stalled_metadata_repair`'s `repairing_view`.
- let repairing_view =
partition.consensus().view_log_is_pending()
- &&
partition.consensus().is_primary_for_view(consensus_view);
- let session_live = consensus_normal || repairing_view;
+ // 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 !session_live || consensus_view != session.view {
return true;
diff --git a/core/simulator/src/workload/invariants.rs
b/core/simulator/src/workload/invariants.rs
index c9912b5a3..d2011bcb5 100644
--- a/core/simulator/src/workload/invariants.rs
+++ b/core/simulator/src/workload/invariants.rs
@@ -47,9 +47,11 @@ const COMMIT_PREFIX_HOLE_WEDGE_TICKS: u32 = 2_000;
/// Ticks a replica may hold its commit walk below an ALREADY APPLIED head.
///
-/// Near zero: nothing clears this one, so there is no legitimate window to
wait
-/// out. The slack only covers a reading taken between an apply and its pop.
-const COMMIT_PREFIX_APPLIED_HEAD_TICKS: u32 = 5;
+/// 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.
@@ -207,8 +209,8 @@ impl Invariants {
/// - [`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, so it gets a
near-zero
- /// threshold. Resetting it on `commit_min` would be exactly wrong: the
head
+ /// - [`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
@@ -233,8 +235,8 @@ impl Invariants {
),
CommitHoldKind::AppliedHead => (
COMMIT_PREFIX_APPLIED_HEAD_TICKS,
- "the commit walk advanced past this entry, so nothing is
missing and nothing \
- will ever pop it: its reply can no longer be built and its
awaiter never wakes",
+ "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
