krishvishal commented on code in PR #4073:
URL: https://github.com/apache/iggy/pull/4073#discussion_r3947950925
##########
core/shard/src/lib.rs:
##########
@@ -5532,18 +5542,29 @@ where
})
};
if let Some((peer, nonce, to_op)) = stalled {
- // 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.
+ // Primary-elect only, and floored so a retry re-requests the
window the
+ // initial arm did. A backup's parked `StartView` suffix is a
+ // verification reference: 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)))
+ .then(|| {
+ consensus.with_pending_view_log(|pending| {
+ merged_log_scan_floor(pending, consensus.commit_min())
+ })
+ })
.flatten()
.unwrap_or_else(|| consensus.commit_min() + 1);
- if from_op <= to_op {
+ if from_op > to_op {
+ // Satisfied. 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.
+ *self.metadata_repair.borrow_mut() = None;
+ } else {
+ // The quiet peer may be the thing that died, and nothing else
+ // re-targets a journal-repair session, so retrying it forever
pins
+ // the walk while the rest of the cluster is serveable.
+ let peer = next_repair_peer(consensus.replica_count(),
consensus.replica(), peer);
Review Comment:
Fixed. `rotate_stalled_metadata_repair` writes `session.peer`, fires only
past `REPAIR_MAX_STALL_RETRIES` with the budget on the shard so a re-armed
session can't reset it, and picks via `next_transfer_peer` for a Normal backup
and the DVC source set for a primary-elect; `next_repair_peer` is gone.
`RangeEvicted` also stops converting when `retained_from <= commit_min + 1`,
which is what a peer behind the window sends.
##########
core/shard/src/lib.rs:
##########
@@ -5753,7 +5785,7 @@ 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 missing = first_op_not_covered(&pending, repair_floor,
consensus.commit_min(), |op| {
Review Comment:
Fixed. `pending_view_commit_sources` (senders with `commit >= op`, freshest
`log_view` first) covers the range `index_of` can't see, and a primary-elect
now lets the view-change timeout escalate on `RangeEvicted` instead of arming a
transfer. Tested in `impls::view_source_tests` against a recorded DVC quorum.
##########
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.
+ pub fn redecide_recovery_barrier(&self, head: u64) {
+ let barrier = self.recovery_barrier.get();
+ if barrier == 0 {
+ return;
+ }
+ let barrier = barrier.min(head);
+ self.recovery_barrier
+ .set(if barrier <= self.commit_max.get() {
Review Comment:
Fixed: `barrier.min(head)`, no collapse. `make_start_view(7, 105, ...)` sets
`commit = op`, which is why the barrier zeroed at `commit_min` 100; the test
now expects 105 and asserts the `commit_min` gate still holds.
##########
core/consensus/src/plane_helpers.rs:
##########
@@ -482,15 +510,37 @@ 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 stalls
rather
+/// than panicking for the same one: 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 next = consensus.commit_min() + 1;
+ let head = consensus
.pipeline_head_header()
- .filter(|header| header.op <= commit)
+ .filter(|header| header.op <= commit)?;
+ if head.op != next {
+ // Unreachable in debug and the simulator; release reports and waits.
Review Comment:
Fixed. Both are warn-and-hold, and a hold that never clears trips a new
simulator invariant (`check_commit_prefix_contiguity`, 2000 ticks, both
planes). 529's only caller is the metadata plane, whose walk is unbounded, so
that one changed for symmetry rather than reachability.
##########
core/simulator/src/lib.rs:
##########
@@ -1319,6 +1332,24 @@ impl Simulator {
Some(partition.offsets())
}
+ /// A replica's journaled partition-plane prepare header at `op`, or
`None` when
+ /// it does not host the namespace or no longer holds the entry.
+ ///
+ /// Absence is ordinary, unlike on the metadata plane: the partition
journal
+ /// evicts its committed prefix as it flushes to segments. The quiesce
oracle
+ /// compares only the ops two replicas both still hold.
+ #[must_use]
+ pub(crate) fn partition_journaled_header(
+ &self,
+ replica_idx: usize,
+ namespace: IggyNamespace,
+ op: u64,
+ ) -> Option<PrepareHeader> {
+ let shard = self.replicas[replica_idx].partition_shard(namespace);
+ let partition = shard.plane.partitions().get_by_ns(&namespace)?;
+ partition.log.journal().inner.header_by_op(op)
Review Comment:
Fixed. `repair_headers_in(1..=commit_min)`, one pass per replica per
namespace; `partition_ops_compared` went 503 to 1516 on seed 1.
`a_flushed_partition_prefix_is_still_compared` asserts op 1 is non-resident
after a real flush yet `compared > 0`.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -2081,10 +2081,27 @@ where
.is_some_and(|&high_water| offset <= high_water)
}
- fn apply_consumer_offset_commit(
- &self,
- pending: PendingConsumerOffsetCommit,
- ) -> Result<(), IggyError> {
+ /// Note a committed delete that found no offset to remove.
+ ///
+ /// Expected wherever the paired `AckLevel::NoAck` store never replicated,
so a
+ /// diagnostic and not a fault. Still logged: on a replica that did serve
the
+ /// store it is the first symptom of a lost apply.
+ fn log_absent_offset_delete(&self, kind: &str, id: u64) {
Review Comment:
Fixed.
`given_an_absent_offset_when_committing_a_delete_should_apply_without_fencing`
commits the delete on a primary for both `Consumer` and `ConsumerGroup` and
asserts `fatal().is_none()`, plus the present-offset counterpart.
--
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]
