hubcio commented on code in PR #4073:
URL: https://github.com/apache/iggy/pull/4073#discussion_r3949612851
##########
core/shard/src/lib.rs:
##########
@@ -4972,6 +4975,9 @@ where
if !in_scope {
return;
}
+ // The serving peer is answering. Clears the stall clock and the
budget
+ // so a window served in chunks cannot rotate off a healthy stream.
+ self.note_metadata_repair_progress();
Review Comment:
warning: repair-prepare frames carry no sender identity or session nonce, so
a peer that was already rotated away can still clear the new session's clock
and budget with in-flight frames. carry the nonce on the frame.
##########
core/shard/src/lib.rs:
##########
@@ -10646,7 +10815,7 @@ 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)
+ (merged_log_scan_floor(pending, commit_min).max(repair_floor +
1)..=pending.op_head)
.find(|op| !held(*op))
.or_else(|| {
Review Comment:
warning: this fallback filters only by `repair_floor`, so `missing_op` can
land below `merged_log_scan_floor` while the retry's `from_op` sits above it -
the arm asks for that op once and the retry never asks again. store `from_op`
on the session and floor the retry at it.
##########
core/shard/src/lib.rs:
##########
@@ -5163,6 +5160,51 @@ 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 has just been told it
+ // cannot serve; installing under it would start the view
over
+ // a log the new state no longer matches. The honest
answer is
+ // that another replica holds the committed prefix, so
leave
+ // the session for the stall rotation to re-target and let
the
Review Comment:
warning: this returns without touching the session or its `idle_ticks`, so
every `RangeEvicted` costs a full `repair_retry_ticks` interval before rotation
looks at another sender. force the stall here, or call
`rotate_stalled_metadata_repair` directly.
##########
core/shard/src/lib.rs:
##########
@@ -10646,7 +10815,7 @@ 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)
+ (merged_log_scan_floor(pending, commit_min).max(repair_floor +
1)..=pending.op_head)
Review Comment:
warning: this floor collapses to `commit_min + 1` at the partition site, so
the walk length now tracks the apply backlog against the linear `header_by_op`
probe, quadratic while a view change is parked. building the window once into a
map fixes the cost - though note a ring-aware probe on its own would report
coverage for ops `collect_committable_from_journal` still cannot walk.
##########
core/simulator/src/lib.rs:
##########
@@ -930,6 +955,19 @@ impl Simulator {
}
for shard in &replica.shards {
let pending = shard.inbox_len();
+ // A fenced pump has exited, so its frames pile up exactly as a
+ // missed wake does. Reported apart: the fix is a failed
commit,
+ // not a channel bug.
+ assert!(
+ pending == 0 || shard.fenced_partition_fault().is_none(),
Review Comment:
warning: the fenced check sits behind `pending == 0 ||`, so a fenced
partition with a drained inbox passes quiescence - and the pump's
`Option<FatalCommit>` is dropped at line 477, so nothing else sees it. assert
`fenced_partition_fault().is_none()` first, unconditionally.
##########
core/simulator/src/lib.rs:
##########
@@ -930,6 +955,19 @@ impl Simulator {
}
for shard in &replica.shards {
let pending = shard.inbox_len();
+ // A fenced pump has exited, so its frames pile up exactly as a
+ // missed wake does. Reported apart: the fix is a failed
commit,
+ // not a channel bug.
+ assert!(
+ pending == 0 || shard.fenced_partition_fault().is_none(),
+ "fenced pump: replica {replica_id} shard {} holds
{pending} frame(s) at \
+ quiescence because its pump exited on a fatal commit
({:?}). Not a lost \
+ wakeup; fix the commit failure (seed {:#x}, schedule hash
{:#x})",
+ shard.id,
+ shard.fenced_partition_fault(),
Review Comment:
warning: the reply-lane assert below has no such guard, so a fenced pump
still trips "lost wakeup: reply lane holds N frame(s)" - the exact misdiagnosis
this is meant to prevent. add the same guard there.
##########
core/shard/src/lib.rs:
##########
@@ -5596,51 +5646,44 @@ where
})
};
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!(
+ // Primary-elect only, and floored at `merged_log_scan_floor` so a
Review Comment:
nit: the retry doesn't re-request the window the initial arm did - it opens
at the floor instead of `missing_op`, and drops the `repair_floor` raise. fix
the comment.
##########
core/consensus/src/impls.rs:
##########
@@ -1729,6 +1729,33 @@ 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.
+ pub fn redecide_recovery_barrier(&self, head: u64) {
Review Comment:
simplification: `pub` with no callers outside this file - only 3395, 3761
and the test at 4885. make it private.
##########
core/shard/src/lib.rs:
##########
@@ -5596,51 +5646,44 @@ where
})
};
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!(
+ // Primary-elect only, and floored at `merged_log_scan_floor` so a
+ // retry re-requests the window the initial arm did. The merged
log's
+ // commit point 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, and above it when the local
+ // prefix has a hole, so the commit point alone would skip that. 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| {
+ merged_log_scan_floor(pending, consensus.commit_min())
+ })
+ })
+ .flatten()
Review Comment:
warning: the retry `from_op` has no snapshot clamp, unlike the initial arm's
`.max(repair_floor + 1)`, so it can request compacted ops no repair puts back -
and rotation picks its sources off the same value. floor it at `snapshot_op() +
1`.
##########
core/shard/src/lib.rs:
##########
@@ -5499,9 +5541,12 @@ where
// from the evicted ring or the flushed segments.
let missing = {
let journal = partition.log.journal();
- first_op_not_covered(&pending, consensus.commit_min(), |op| {
- journal.inner.header_by_op(op)
- })
+ first_op_not_covered(
+ &pending,
+ consensus.commit_min(),
+ consensus.commit_min(),
+ |op| journal.inner.header_by_op(op),
Review Comment:
warning: for ops the new floor admits below `pending.commit_max` the
`disagrees` check finds no canonical header and passes, and `held` falls back
to bare residency, so a forked prepare gets journaled and applied. extend
`committed_elsewhere` down to the scan floor, or state the crash-fault-only
assumption.
##########
core/simulator/src/workload/invariants.rs:
##########
@@ -102,6 +141,91 @@ 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.replicas[usize::from(replica_idx)].shards[0]
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ else {
+ return;
+ };
+ let gated = consensus.is_primary()
Review Comment:
warning: this omits `!consensus.is_transferring()`, which
`is_caught_up_primary` has - a primary in state transfer is shut by the
transfer, not the barrier, and 2000 ticks of it panics blaming the barrier.
mirror the predicate.
##########
core/consensus/src/plane_helpers.rs:
##########
@@ -1995,6 +2041,8 @@ mod tests {
fn drains_only_up_to_commit_frontier_even_without_quorum_flags() {
Review Comment:
warning: no test covers `head_op < commit_min + 1` on either gate - the only
thing that catches it is the simulator invariant, and that needs 2000
consecutive ticks. add a unit test per gate.
##########
core/consensus/src/impls.rs:
##########
@@ -3668,6 +3698,30 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
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> {
Review Comment:
simplification: this duplicates `pending_view_body_sources` apart from one
predicate, including the sort and the self-exclusion. one private selector
taking `impl Fn(&StoredDvc) -> bool` keeps those in one place.
##########
core/consensus/src/plane_helpers.rs:
##########
@@ -446,18 +456,35 @@ where
P: Pipeline<Entry = PipelineEntry>,
{
let commit = consensus.commit_max();
+ let commit_min = consensus.commit_min();
+ let replica = consensus.replica();
Review Comment:
nit: `replica` is only read inside the warn arm, and the twin at 530 inlines
it. fold it into the macro args for consistency.
##########
core/shard/src/lib.rs:
##########
@@ -5658,22 +5701,108 @@ 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.
+ #[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;
+ self.metadata_repair_attempts.set(0);
+
+ if consensus.view_log_is_pending() &&
consensus.is_primary_for_view(consensus.view()) {
+ let sources = metadata_view_repair_sources(consensus, from_op);
+ let Some(next_peer) = sources.into_iter().find(|candidate|
*candidate != 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;
+ };
+ let nonce = iggy_common::random_id::get_uuid();
Review Comment:
simplification: third place that mints a `MetadataRepairSession` and calls
`send_request_prepares`, after 6049 and 6107. one `arm_metadata_repair_session`
helper saves about 15 lines and stops the nonce/view/`idle_ticks` invariant
living in three spots.
##########
core/consensus/src/impls.rs:
##########
@@ -4722,6 +4804,97 @@ mod timestamp_clamp_tests {
assert_eq!(consensus.status(), Status::Normal);
}
+ /// The wedge `redecide_recovery_barrier` exists for: a view
Review Comment:
nit: these recovery-barrier tests land in `timestamp_clamp_tests`, whose
module doc covers the monotonic-floor contract only. give them their own module.
##########
core/simulator/src/lib.rs:
##########
@@ -1353,6 +1415,42 @@ impl Simulator {
})
}
+ /// The metadata pipeline head the commit walk is holding on, if any. See
+ /// [`CommitPrefixHole`].
+ #[must_use]
+ pub(crate) fn metadata_commit_prefix_hole(
Review Comment:
simplification: this and `partition_commit_prefix_hole` share an identical
tail, and `check_recovery_barrier` hand-walks the same chain a third way. a
`metadata_consensus(idx)` accessor gives one way to reach the handle - it's
line-neutral, so buy it for consistency, not lines.
##########
core/consensus/src/impls.rs:
##########
@@ -5404,3 +5580,130 @@ 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 message_bus::BusMessage;
+ use server_common::MESSAGE_ALIGN;
+ use server_common::iobuf::Frozen;
+
+ struct NoopBus;
Review Comment:
simplification: third identical `NoopBus` plus `MessageBus` impl in this
file, after 4501 and 5518. one shared `#[cfg(test)] mod test_bus` saves about
45 lines.
##########
core/consensus/src/impls.rs:
##########
@@ -4544,6 +4600,32 @@ mod timestamp_clamp_tests {
msg
}
+ /// [`make_start_view`] with commit decoupled from head, for a view that
keeps
+ /// an uncommitted suffix.
+ #[allow(clippy::cast_possible_truncation)]
+ fn make_start_view_with_commit(
Review Comment:
simplification: `make_start_view_with_commit` copies `make_start_view`
verbatim apart from `commit` being a parameter. give the original the parameter
and update its four call sites, about 24 lines.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
