spetz commented on code in PR #4073:
URL: https://github.com/apache/iggy/pull/4073#discussion_r3945308691
##########
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:
`header_by_op` reads resident headers only. `evict_prefix` clears them on
flush and moves the entries to the repair ring. After every replica flushes,
`assert_partition_prefixes_agree` compares zero ops and the check is vacuous at
quiescence.
Use `repair_headers_in(1..=commit_min)` once per replica per namespace,
since state_checker.rs:333 probes op by op and both lookups are linear. Add a
post-flush comparison test.
##########
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:
Patch coverage on this file is 20%. Add a unit test that commits
DeleteConsumerOffset for an absent offset on a primary and asserts the
partition stays unfenced.
##########
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:
This scan now reports a `missing_op` below `pending.commit_max`, but
`pending_view_body_sources(missing_op)` (5829) only looks at DVC suffixes,
which span `commit..=op` per sender. `index_of` returns None below that, the
source list comes back empty, and the view change stalls until the timeout
escalates. Every sender with `commit >= missing_op` holds or has compacted the
op.
For `missing_op < pending.commit_max`, select DVC senders with `commit >=
missing_op`, most recent `log_view` first. A RangeEvicted from such a peer
means this replica cannot serve the committed prefix, so let the view-change
timeout escalate rather than arm a state transfer as primary-elect. Test
against a real 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:
Collapsing the barrier to 0 when `barrier <= commit_max` opens the HTTP read
gate early. `await_recovery_barrier` (core/server/src/http/reads.rs:279) gates
on `commit_min` because adoption advances `commit_max` before applying the
suffix. With head 105 and `commit_min` 100, new readers see barrier 0 and serve
state from before ops 101..=105 apply. `is_caught_up_primary` compares
`commit_max >= barrier` and needs no zero.
Set `barrier.min(head)` and drop the collapse. Update
`given_a_discarded_suffix_when_adopting_a_view_should_lower_the_barrier` to
expect 105 and assert `commit_max() >= recovery_barrier()`.
##########
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:
"Unreachable in debug and the simulator" is false on the partition plane.
`IggyPartition::commit_journal` walks at most `COMMIT_WALK_OPS_MAX` (64) ops
per call, and a promoted primary's pipeline is seeded from `merged.commit_max +
1`. A primary-elect whose journal covers the merged log but whose apply lags by
more than 64 ops passes the coverage scan, starts the view, and receives quorum
acks for `commit_max + 1` before the sweep drains the backlog.
`drain_committable_prefix` then sees `head_op > commit_min + 1` and the assert
fires. Release logs an error per ack until the walk catches up.
Finish the journal walk before the pipeline drain on partition promotion, or
downgrade the asserts at 466 and 529 to a log.
##########
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:
`session.peer` is never updated. The rotated value is a shadowed local, so
every retry rotates from the original peer and the RepairDone continuation
(line 5097) still sends to it.
Rotation is also blind and immediate. A Normal backup rotates on the first
stall to any replica, including one lagging below `from_op`.
`on_request_prepares` (4683-4708) answers a range it never held with
RangeEvicted and RepairDone, and the RangeEvicted arm (5106) arms a state
transfer against that peer without checking `retained_from` against `commit_min
+ 1`. On the primary-elect path the rotation leaves the
`pending_view_body_sources` set and hits the same conversion mid view change.
Write `session.peer`. Reuse `next_transfer_peer` (2518, prefers the primary)
instead of adding `next_repair_peer`. Rotate only after a retry budget, as
`tick_partitions` does (7148). Keep primary-elect rotation inside
`pending_view_body_sources`.
--
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]
