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
commit f7c3338914274f6a1ac1fb828bcfb81a94c74caf Author: Krishna Vishal <[email protected]> AuthorDate: Sat Sep 5 16:56:48 2026 +0530 fix(consensus): keep a replica off a hole in its committed prefix --- core/consensus/src/impls.rs | 147 ++++++++++++++++++++++ core/consensus/src/plane_helpers.rs | 62 ++++++++- core/partitions/src/iggy_partition.rs | 72 ++++++----- core/shard/src/lib.rs | 180 ++++++++++++++++++++++++--- core/shard/src/router.rs | 9 ++ core/simulator/src/bin/workload-fuzz.rs | 16 ++- core/simulator/src/lib.rs | 33 ++++- core/simulator/src/workload/invariants.rs | 57 +++++++++ core/simulator/src/workload/oracle.rs | 14 +++ core/simulator/src/workload/state_checker.rs | 62 +++++++++ 10 files changed, 586 insertions(+), 66 deletions(-) diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 159e0ca5e..f41f06e6d 100644 --- a/core/consensus/src/impls.rs +++ b/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() { + 0 + } else { + barrier + }); + } + /// Deadline paired with [`Self::recovery_barrier`]; only meaningful while the /// barrier is armed (non-zero). #[must_use] @@ -3363,6 +3390,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 { @@ -3703,6 +3733,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. @@ -4544,6 +4576,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( + view: u32, + op: u64, + commit: u64, + replica: u8, + ) -> 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.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 @@ -4722,6 +4780,92 @@ mod timestamp_clamp_tests { assert_eq!(consensus.status(), Status::Normal); } + /// 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() { + // Recovered at head 120, proven committed only through 100. + 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); + + // 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, 1, 0).header(), + &[] + ) + .is_empty(), + "the StartView at the commit floor must be adopted" + ); + assert_eq!( + consensus.recovery_barrier(), + 0, + "the adopted head settled the suffix's fate and commit_max covers it, \ + so the barrier must clear rather than latch" + ); + assert!( + is_caught_up_primary_barrier_open(&consensus), + "a cleared barrier must stop gating admission" + ); + } + + /// 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 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); + + // Head 120, commit still 100: 101..=120 re-replicate under the new view. + let start_view = make_start_view_with_commit(7, 120, 100, 1); + assert!( + !consensus + .handle_start_view(PlaneKind::Metadata, start_view.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!( + !is_caught_up_primary_barrier_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); + } + + /// The `commit_max >= recovery_barrier` clause of `is_caught_up_primary`, read + /// directly so these tests need not satisfy the primary/status clauses. + fn is_caught_up_primary_barrier_open<B: MessageBus, P>(consensus: &VsrConsensus<B, P>) -> bool + where + P: Pipeline<Entry = PipelineEntry>, + { + consensus.commit_max() >= consensus.recovery_barrier() + } + /// The split-brain gate's predicate: `view` and `log_view` each independently /// make the superblock stale, and only a matching `mark_superblock_durable` /// clears it. The simulator proves the withheld-send behavior end to end; this @@ -5075,6 +5219,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 +5228,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" diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs index 195120af8..0d38def00 100644 --- a/core/consensus/src/plane_helpers.rs +++ b/core/consensus/src/plane_helpers.rs @@ -435,8 +435,14 @@ where /// 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, and a backstop +/// only: reaching it means committing over a hole coverage should have caught. +/// Loud in debug and the simulator, hold-and-repair in release. /// /// # Panics /// If `head()` returns `Some` but `pop()` returns `None` (unreachable). @@ -446,18 +452,39 @@ where P: Pipeline<Entry = PipelineEntry>, { let commit = consensus.commit_max(); + let commit_min = consensus.commit_min(); + let replica = consensus.replica(); 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 { + debug_assert_eq!( + head_op, next, + "pipeline head must be the next op owed to the state machine" + ); + tracing::error!( + replica, + head_op, + expected_op = next, + commit_min, + commit_max = commit, + drained = drained.len(), + "committable head sits above a hole in the committed prefix; holding the \ + commit walk until repair refills it" + ); + break; + } let entry = pipeline .pop() .expect("drain_committable_prefix: head exists"); drained.push(entry); + next += 1; } }); @@ -473,7 +500,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 +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. + debug_assert_eq!( + head.op, next, + "pipeline head must be the next op owed to the state machine" + ); + tracing::error!( + replica = consensus.replica(), + head_op = head.op, + commit_min = consensus.commit_min(), + commit_max = commit, + "committable head sits above a hole in the committed prefix; holding the \ + commit walk until repair refills it" + ); + return None; + } + Some(head) } /// Build reply for a committed prepare. @@ -1995,6 +2045,8 @@ mod tests { 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/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 2128ae517..bdfa0bc2f 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -2008,7 +2008,7 @@ where // durably stored; the in-memory update is idempotent on replay // because we look up by (kind, id). self.persist_consumer_offset_commit(pending).await?; - self.apply_consumer_offset_commit(pending)?; + self.apply_consumer_offset_commit(pending); self.pending_consumer_offset_commits.remove(&op); Ok(()) } @@ -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) { + debug!( + target: "iggy.partitions.diag", + plane = "partitions", + replica_id = self.consensus.replica(), + namespace_raw = self.namespace().inner(), + kind, + id, + follower = self.consensus.is_follower(), + "committed consumer offset delete found no offset to remove" + ); + } + + /// Infallible: idempotent map mutations. A committed op must apply on every + /// replica, so there is no way to refuse one. + fn apply_consumer_offset_commit(&self, pending: PendingConsumerOffsetCommit) { match pending.mutation { PendingConsumerOffsetMutation::Upsert(offset) if pending.kind == ConsumerKind::Consumer => @@ -2104,7 +2121,6 @@ where pending.auto_commit, create, ); - Ok(()) } PendingConsumerOffsetMutation::Upsert(offset) if pending.kind == ConsumerKind::ConsumerGroup => @@ -2133,28 +2149,20 @@ where pending.auto_commit, create, ); - Ok(()) } - // Commit-time apply keeps its invariant check on the PRIMARY: - // admission verified the offset exists there, so a miss on the - // primary is real divergence (log corruption / out-of-order apply) - // and must surface rather than silently mask a split state. A - // FOLLOWER may legitimately miss the offset: `AckLevel::NoAck` - // stores apply on the primary only and are never replicated, - // so a later quorum delete finds nothing on the backups -- erroring - // there would fail the committed apply, panic the replica as - // divergent, and crash-loop on every journal replay. The - // prepare-time race is handled by not re-checking existence at - // staging (see `stage_consumer_offset_delete`). + // Deleting an absent offset is a no-op on every role: `AckLevel::NoAck` + // stores never replicate, so which replicas hold an offset is not + // agreed and presence cannot be an invariant of a committed delete. + // Erroring would fence, then crash-loop on replay, over a state the + // design permits. The prepare-time race is handled by not re-checking + // existence at staging (see `stage_consumer_offset_delete`). PendingConsumerOffsetMutation::Delete if pending.kind == ConsumerKind::Consumer => { let id = pending.consumer_id; let guard = self.consumer_offsets.pin(); let key = usize::try_from(id).expect("u32 consumer id must fit usize"); - let removed = guard.remove(&key).is_some(); - if !removed && !self.consensus.is_follower() { - return Err(IggyError::ConsumerOffsetNotFound(key)); + if guard.remove(&key).is_none() { + self.log_absent_offset_delete("consumer", u64::from(id)); } - Ok(()) } PendingConsumerOffsetMutation::Delete if pending.kind == ConsumerKind::ConsumerGroup => @@ -2164,13 +2172,11 @@ where let key = ConsumerGroupId( usize::try_from(group_id).expect("u32 group id must fit usize"), ); - let removed = guard.remove(&key).is_some(); - if !removed && !self.consensus.is_follower() { - return Err(IggyError::ConsumerOffsetNotFound(key.0)); + if guard.remove(&key).is_none() { + self.log_absent_offset_delete("consumer_group", u64::from(group_id)); } - Ok(()) } - _ => Ok(()), + _ => {} } } @@ -2269,15 +2275,7 @@ where ); return; } - if let Err(error) = self.apply_consumer_offset_commit(pending) { - emit_partition_diag( - tracing::Level::WARN, - &PartitionDiagEvent::new(self.diag_ctx(), "no_ack offset apply failed") - .with_operation(request_header.operation) - .with_error(error.to_string()), - ); - return; - } + self.apply_consumer_offset_commit(pending); let reply = build_reply_from_request( &self.consensus, @@ -2587,7 +2585,7 @@ where offset: u64, ) -> Result<(), IggyError> { let pending = PendingConsumerOffsetCommit::try_from_polling_consumer(consumer, offset)?; - self.apply_consumer_offset_commit(pending)?; + self.apply_consumer_offset_commit(pending); Ok(()) } diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 8181831fd..cf979d73b 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -5459,9 +5459,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), + ) }; if let Some(missing_op) = missing { tracing::debug!( @@ -5500,10 +5503,17 @@ where } } - /// Re-request the remaining repair window when the stream has gone quiet. + /// Re-request the remaining repair window when the stream has gone quiet, and + /// keep the session honest about whether repair is still owed. /// /// Repair frames are fire-and-forget, so a lost one leaves the session armed /// forever with the commit walk pinned below the frontier. + /// + /// The session is edge-armed but serves a level condition. A window can be + /// satisfied without the walk reaching `to_op` (ops above it arrive live and + /// are dropped by `on_replicate`'s gap check, which arms no repair), leaving a + /// session no completion clears and an `is_none` gate that refuses a new one. + /// So: clear on satisfaction, re-arm on the level. #[allow(clippy::future_not_send)] async fn retry_stalled_metadata_repair<P>(&self, consensus: &VsrConsensus<B, P>) where @@ -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); tracing::info!( shard = self.id, from_op, @@ -5563,6 +5584,17 @@ where .await; } } + + // Level trigger: covers every way a gap opens without arming repair, the + // clear above and `on_replicate` dropping a non-contiguous prepare. + let unserved_gap = consensus.commit_min() < consensus.commit_max() + && self.metadata_repair.borrow().is_none(); + if unserved_gap { + let primary = consensus.primary_index(consensus.view()); + if primary != consensus.replica() { + self.maybe_request_metadata_repair(consensus, primary).await; + } + } } /// Compare this replica's log against the headers the view decided, and drop or @@ -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| { usize::try_from(op) .ok() .and_then(|slot| journal.handle().header(slot)) @@ -9570,6 +9602,27 @@ where } } +/// Next target for a repair session whose peer has gone quiet. +/// +/// Round-robin over the other replicas, so a dead peer is left behind within one +/// retry interval. Falls back to `current` on a solo cluster. +fn next_repair_peer(replica_count: u8, self_id: u8, current: u8) -> u8 { + (1..replica_count) + .map(|step| (current + step) % replica_count) + .find(|candidate| *candidate != self_id) + .unwrap_or(current) +} + +/// 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 exactly when the local prefix has a hole. The +/// lower of the two keeps coverage, repair scope and the stall retry answering +/// the same question about that hole. +fn merged_log_scan_floor(pending: &MergedLog, commit_min: u64) -> u64 { + pending.commit_max.min(commit_min + 1).max(1) +} + /// Whether a repaired prepare at `op` falls inside the range this replica is /// currently repairing. /// @@ -9590,7 +9643,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() @@ -10201,9 +10254,14 @@ const fn header_is_view_entry(local: &PrepareHeader, canonical: &PrepareHeader) /// 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. +/// +/// 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. fn first_op_not_covered( pending: &MergedLog, repair_floor: u64, + commit_min: u64, header_at: impl Fn(u64) -> Option<PrepareHeader>, ) -> Option<u64> { let held = |op: u64| { @@ -10217,7 +10275,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(|| { pending @@ -10818,7 +10876,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, 0, pending.commit_max, |op| { held.iter().find(|header| header.op == op).copied() }); assert_eq!(missing, Some(99)); @@ -10842,16 +10900,102 @@ mod view_coverage_tests { }; let nothing_resident = |_: u64| None; assert_eq!( - first_op_not_covered(&pending, 0, nothing_resident), + first_op_not_covered(&pending, 0, pending.commit_max, 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, 256, pending.commit_max, 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, 0, 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, 0, pending.commit_max, |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_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" + ); + } +} + +#[cfg(test)] +mod repair_peer_tests { + use super::next_repair_peer; + + #[test] + fn given_a_quiet_peer_when_rotating_should_skip_self_and_cycle() { + // Replica 1 of 3 must reach the only other peer, and never itself. + assert_eq!(next_repair_peer(3, 1, 0), 2); + assert_eq!(next_repair_peer(3, 1, 2), 0); + assert_eq!( + next_repair_peer(3, 1, 1), + 2, + "rotating off self must still land on a real peer" + ); + } + + #[test] + fn given_a_solo_cluster_when_rotating_should_hold_the_current_peer() { + // Nowhere to rotate to; keep the current peer over a nonexistent id. + assert_eq!(next_repair_peer(1, 0, 0), 0); + } } #[cfg(test)] diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 4a3a0d214..8ce6c2cab 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -532,6 +532,15 @@ where }) } + /// [`Self::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..381203746 100644 --- a/core/simulator/src/bin/workload-fuzz.rs +++ b/core/simulator/src/bin/workload-fuzz.rs @@ -562,9 +562,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 +576,17 @@ 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, ); // 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 8a9d7c631..dd06323d9 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; @@ -930,6 +930,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(), + self.seed, + self.executor.schedule_hash(), + ); assert_eq!( pending, 0, @@ -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) + } + /// Consensus view for a replica's partition-plane group, or `None` if that /// replica does not host the namespace. #[must_use] diff --git a/core/simulator/src/workload/invariants.rs b/core/simulator/src/workload/invariants.rs index 4f6dd3a1d..4efa62a7e 100644 --- a/core/simulator/src/workload/invariants.rs +++ b/core/simulator/src/workload/invariants.rs @@ -26,15 +26,27 @@ use crate::Simulator; use crate::workload::state_checker::StateChecker; use crate::workload::{CLIENT_REQUEST_QUEUE_MAX, Workload}; +use consensus::{Consensus, MetadataHandle}; 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; + /// 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>, /// 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,8 +83,10 @@ impl Invariants { for replica_idx in 0..sim.replica_count { if sim.is_crashed(replica_idx) { + self.barrier_gated_ticks.remove(&replica_idx); continue; } + self.check_recovery_barrier(sim, seed, replica_idx); for &ns in &workload.options.namespaces { if let Some(offsets) = sim.offsets(usize::from(replica_idx), ns) { let cur = offsets.commit_offset; @@ -102,6 +116,49 @@ 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() + && !consensus.has_ceded_primaryship() + && consensus.is_normal() + && consensus.commit_max() < 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 from here on (seed={seed:#x})", + consensus.recovery_barrier(), + consensus.commit_min(), + consensus.commit_max(), + consensus.view(), + ); + } + /// 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..d1426850a 100644 --- a/core/simulator/src/workload/state_checker.rs +++ b/core/simulator/src/workload/state_checker.rs @@ -35,6 +35,7 @@ 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. @@ -291,6 +292,67 @@ 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: 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; + }; + for op in 1..=state.commit_min { + let Some(header) = sim.partition_journaled_header(idx, namespace, op) else { + continue; + }; + 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. ///
