This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch fix_partition_replica_commit in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 00b57ffdec854be299541176e97a72930252bf34 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Wed Oct 7 16:54:52 2026 +0200 fix(partitions): don't commit a stale prepare after adopting a newer view --- core/consensus/src/dvc_merge.rs | 30 +- core/consensus/src/impls.rs | 17 +- core/partitions/src/iggy_partition.rs | 668 ++++++++++++++++++++++- core/partitions/src/journal.rs | 22 +- core/partitions/src/lib.rs | 2 + core/partitions/src/view_fence.rs | 153 ++++++ core/shard/src/lib.rs | 293 +++++++++- core/simulator/src/lib.rs | 3 + core/simulator/src/stale_prepare_commit_tests.rs | 325 +++++++++++ 9 files changed, 1474 insertions(+), 39 deletions(-) diff --git a/core/consensus/src/dvc_merge.rs b/core/consensus/src/dvc_merge.rs index 0f8dfca4d..56aae9e42 100644 --- a/core/consensus/src/dvc_merge.rs +++ b/core/consensus/src/dvc_merge.rs @@ -27,8 +27,10 @@ #[cfg(test)] use crate::view_change_quorum::DvcSuffix; -use crate::view_change_quorum::{DvcQuorumArray, StoredDvc, dvc_count, dvc_iter}; -use iggy_binary_protocol::{CHECKSUM_UNSEALED, PrepareHeader}; +use crate::view_change_quorum::{ + DvcQuorumArray, DvcSuffixError, StoredDvc, dvc_count, dvc_iter, dvc_suffix_decode, +}; +use iggy_binary_protocol::{CHECKSUM_UNSEALED, PrepareHeader, StartViewHeader}; /// Sizes the merge needs from the replica. #[derive(Debug, Clone, Copy)] @@ -88,6 +90,30 @@ pub struct MergedLog { pub committed_elsewhere: Vec<PrepareHeader>, } +impl MergedLog { + /// The log a `StartView` announces: its suffix slots verbatim, blanks included, + /// or `None` when it carries numbers only. + /// + /// # Errors + /// Returns the decode error when the suffix body is malformed. + pub fn from_start_view( + header: &StartViewHeader, + suffix_body: &[u8], + ) -> Result<Option<Self>, DvcSuffixError> { + let suffix = dvc_suffix_decode(suffix_body, header.op, 0, 0)?; + let headers = suffix.headers(); + if headers.is_empty() { + return Ok(None); + } + Ok(Some(Self { + op_head: header.op, + commit_max: header.commit, + headers: headers.to_vec(), + committed_elsewhere: Vec::new(), + })) + } +} + /// Highest op the quorum proves committed. /// /// Three independent lower bounds, because a single sender's view of the commit diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index cd17643cc..78e31db4a 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -3442,8 +3442,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { /// Falls back to the announced `op` on an empty body (probe answer, stale-view /// correction). fn adopt_start_view_suffix(&self, header: &StartViewHeader, suffix_body: &[u8]) -> u64 { - let suffix = match dvc_suffix_decode(suffix_body, header.op, 0, 0) { - Ok(suffix) => suffix, + match MergedLog::from_start_view(header, suffix_body) { + Ok(Some(log)) => *self.pending_view_log.borrow_mut() = Some(log), + Ok(None) => {} Err(error) => { tracing::warn!( replica = self.replica, @@ -3452,20 +3453,8 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { op = header.op, "start_view suffix did not decode, falling back to the announced op: {error}" ); - return header.op; } - }; - let headers = suffix.headers(); - if headers.is_empty() { - return header.op; } - - *self.pending_view_log.borrow_mut() = Some(MergedLog { - op_head: header.op, - commit_max: header.commit, - headers: headers.to_vec(), - committed_elsewhere: Vec::new(), - }); header.op } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 9260bb383..7a07a600f 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -38,6 +38,7 @@ use crate::poll_plan::{ use crate::segment::Segment; use crate::state_transfer::{PartitionTransferSession, PendingTransferRearm}; use crate::types::{COMMIT_WALK_OPS_MAX, FatalCommit, RepairConclusion, RepairSession}; +use crate::view_fence::{ViewFence, ViewWindowVerdict, view_window_verdict}; use crate::{ AppendResult, Partition, PartitionOffsets, PartitionsConfig, PollFragments, PollQueryResult, PollingArgs, PollingConsumer, @@ -64,7 +65,8 @@ use iggy_binary_protocol::responses::messages::{ SendMessagesConfirmationResponse, SendMessagesResponse, }; use iggy_binary_protocol::{ - AckLevel, Command, Operation, PrepareHeader, WireDecode, WireEncode, WireIdentifier, + AckLevel, CHECKSUM_UNSEALED, Command, Operation, PrepareHeader, WireDecode, WireEncode, + WireIdentifier, }; use iggy_binary_protocol::{PrepareOkHeader, ReplyHeader, RoutedRequestHeader}; use iggy_common::{ @@ -100,7 +102,7 @@ use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; use tokio::sync::Mutex as TokioMutex; -use tracing::{debug, error, warn}; +use tracing::{debug, error, info, warn}; /// Which of a partition's offset counters are live. /// @@ -231,6 +233,12 @@ where /// server down without the partition moving again in the meantime. fatal: Option<FatalCommit>, pub(crate) pending_consumer_offset_commits: HashMap<u64, PendingConsumerOffsetCommit>, + /// Entries journaled before this replica adopted the current view, held back + /// from commit and acknowledgement until proven. See [`Self::fence_view_window`]. + view_fence: Option<ViewFence>, + /// Set at boot when the WAL replayed history: the first view adoption after a + /// restart fences it even though `log_view` came back unchanged. + recovered_unproven: bool, /// Identity shared with pending polls and replaced when their history retires. poll_history: PollHistoryId, /// Committed consumer-offset membership and values. This is deliberately @@ -653,6 +661,8 @@ where installed_frontier: None, fatal: None, pending_consumer_offset_commits: HashMap::new(), + view_fence: None, + recovered_unproven: false, poll_history: PollHistoryId::default(), durable_consumer_offsets: DurableConsumerOffsets::default(), consumer_offset_capacities: ConsumerKind::ALL @@ -1090,6 +1100,7 @@ where } } } + self.recovered_unproven = persistence.head() > 0; Ok(()) } @@ -1372,11 +1383,30 @@ where return false; } self.persist_repaired_prefix(); - let ready = persistence.certify_log_view( - self.consensus.log_view(), - self.consensus.sequencer().current_sequence(), - self.consensus.last_prepare_checksum(), - ); + let (op, checksum) = if let Some(fence) = self.view_fence.filter(|_| self.view_fenced()) { + // The certificate covers the whole WAL, so while fenced it may only + // vouch for a WAL holding nothing above `commit_min`, and only when + // the window sits below the suffix a `DoViewChange` reports. + let floor = self.consensus.commit_min(); + if persistence.head() != floor || fence.through >= self.consensus.commit_max() { + return false; + } + let checksum = if floor == 0 { + 0 + } else { + let Some(checksum) = persistence.checksum(floor) else { + return false; + }; + checksum + }; + (floor, checksum) + } else { + ( + self.consensus.sequencer().current_sequence(), + self.consensus.last_prepare_checksum(), + ) + }; + let ready = persistence.certify_log_view(self.consensus.log_view(), op, checksum); self.start_persistence(); ready } @@ -1434,6 +1464,13 @@ where let Some(persistence) = &self.persistence else { return; }; + // A held entry can sit at the WAL head, where a repaired successor that + // chains to the view's entry instead would fail the WAL's parent check. + let through = if self.view_fenced() { + through.min(self.consensus.commit_min()) + } else { + through + }; while let Some(op) = persistence.head().checked_add(1) { if op > through || op > self.consensus.sequencer().current_sequence() { break; @@ -5204,6 +5241,12 @@ where // whole chain has it committed. Safe to drop entirely. return; } + // A Break here truncates below `current_op` without rewinding the + // sequencer, so this prepare may land above a hole; the fence still holds + // and repair refills it before anything is applied. + if self.consensus().is_follower() && self.offer_view_anchor(&header) { + self.prove_view_window().await; + } let journal_holds_op = self.log.journal().inner.holds_op(header.op); if journal_holds_op { @@ -5607,6 +5650,12 @@ where return; } self.resynchronize_consumer_offset_reservations(); + if self.view_fenced() { + self.prove_view_window().await; + if self.view_fenced() { + return; + } + } // The primary commits inline via `on_ack` (it drains its own pipeline). // Backups never populate the pipeline - they journal replicated prepares @@ -5705,6 +5754,9 @@ where max_ops: usize, config: &PartitionsConfig, ) -> Vec<PipelineEntry> { + if self.view_fenced() { + return Vec::new(); + } let from_op = self.consensus.commit_min() + 1; // Stop below the pipeline head. The drain above holds rather than pops // when the head is not the op owed next; walking that op out of the @@ -5963,6 +6015,254 @@ where }) } + /// Hold `(commit_min, through]` back from commit and acknowledgement until + /// [`Self::prove_view_window`] links it into `anchor`, the current view's header + /// at `through + 1`. Without an anchor the first current-view prepare at that op + /// supplies one (`offer_view_anchor`). + /// + /// For a replica that adopted a newer view: it can hold an entry from a log the + /// view discarded under the view's commit point, which nothing else compares + /// against the view's log. + pub fn fence_view_window(&mut self, through: u64, anchor: Option<PrepareHeader>) { + if through <= self.consensus.commit_min() { + self.view_fence = None; + return; + } + let view = self.consensus.view(); + let previous = self + .view_fence + .filter(|fence| fence.view == view && fence.through == through); + // An unsealed header names no identity, and a zero `parent` above op 1 + // would link anything below it. + let anchor = anchor + .filter(|anchor| { + anchor.checksum != CHECKSUM_UNSEALED + && (anchor.parent != CHECKSUM_UNSEALED || anchor.op == 1) + }) + .or_else(|| previous.and_then(|fence| fence.anchor)); + if previous.is_none() { + info!( + namespace_raw = self.namespace().inner(), + view, + commit_min = self.consensus.commit_min(), + through, + anchored = anchor.is_some(), + "holding partition entries journaled before this view until its log proves them" + ); + } + self.view_fence = Some(ViewFence { + view, + through, + anchor, + hole: previous.and_then(|fence| fence.hole), + unanchored_ticks: previous.map_or(0, |fence| fence.unanchored_ticks), + }); + } + + /// Whether a fenced window is still unproven. + #[must_use] + pub fn view_fenced(&self) -> bool { + self.view_fence + .as_ref() + .is_some_and(|fence| fence.through > self.consensus.commit_min()) + } + + /// Whether the fence is still waiting for its anchor. + #[must_use] + pub fn view_fence_unanchored(&self) -> bool { + self.view_fenced() + && self + .view_fence + .as_ref() + .is_some_and(|fence| fence.anchor.is_none()) + } + + /// Count one tick without an anchor; true every `every` ticks, when the caller + /// should ask the view's primary for its headers again. + pub fn view_fence_probe_due(&mut self, every: u32) -> bool { + if !self.view_fence_unanchored() { + return false; + } + let Some(fence) = self.view_fence.as_mut() else { + return false; + }; + fence.unanchored_ticks = fence.unanchored_ticks.saturating_add(1); + if fence.unanchored_ticks < every.max(1) { + return false; + } + fence.unanchored_ticks = 0; + true + } + + /// Take `header` as the fence's anchor when it is the op above the window and + /// was prepared in the fence's view. Only for a header from that view's primary + /// (a prepare that passed `replicate_preflight`, or its `StartView`). Returns + /// whether the anchor was taken. + pub(crate) const fn offer_view_anchor(&mut self, header: &PrepareHeader) -> bool { + if let Some(fence) = self.view_fence.as_mut() + && fence.anchor.is_none() + && fence.view == header.view + && header.op == fence.through + 1 + && header.checksum != CHECKSUM_UNSEALED + && (header.parent != CHECKSUM_UNSEALED || header.op == 1) + { + fence.anchor = Some(*header); + fence.hole = None; + return true; + } + false + } + + /// Lowest op of the fenced window, at or under `commit_max`, with no entry: + /// what repair has to fetch before the chain can reach below it. + #[must_use] + pub fn view_fence_hole(&self) -> Option<u64> { + let fence = self.view_fence.as_ref()?; + let floor = self.consensus.commit_min(); + let top = fence.through.min(self.consensus.commit_max()); + (floor + 1..=top).find(|op| !self.holds_view_entry(*op)) + } + + /// Residency of a window op. A forced flush evicts past `commit_min` only on a + /// caught-up primary or at shutdown, and the flush is capped while fenced, so + /// a window op is never ring-only and the index alone answers. + fn holds_view_entry(&self, op: u64) -> bool { + self.log.journal().inner.holds_op(op) + } + + /// Check the fenced window against its anchor. `Proven` lifts the fence, a + /// `Break` drops the entries from the broken op so repair refetches the view's, + /// a `Hole` waits for repair. `None` while there is no fence or no anchor. + /// + /// A `Break` truncates the whole suffix from the broken op, entries the anchor + /// already proved and current-view prepares included; none of them was acked. + pub async fn prove_view_window(&mut self) -> Option<ViewWindowVerdict> { + let fence = self.view_fence?; + let floor = self.consensus.commit_min(); + if fence.through <= floor { + self.view_fence = None; + self.persist_repaired_prefix(); + return Some(ViewWindowVerdict::Proven); + } + let anchor = fence.anchor?; + if let Some(hole) = fence.hole + && !self.holds_view_entry(hole) + { + return Some(ViewWindowVerdict::Hole { op: hole }); + } + let present = self + .log + .journal() + .inner + .walked_headers_in(floor + 1..=fence.through); + let verdict = view_window_verdict(floor, fence.through, anchor.parent, |op| { + present.get(&op).copied() + }); + match verdict { + ViewWindowVerdict::Proven => { + // The WAL flush below reads the last append per op; a duplicate + // whose last append is not the entry just proven breaks there. + if let Some(op) = present.iter().find_map(|(op, header)| { + let flushed = self.log.journal().inner.repair_entry(*op)?; + let flushed = bytemuck::checked::try_from_bytes::<PrepareHeader>( + &flushed.as_slice()[..size_of::<PrepareHeader>()], + ) + .ok()?; + (flushed.checksum != header.checksum).then_some(*op) + }) { + self.drop_unproven_from(op).await; + return Some(ViewWindowVerdict::Break { op }); + } + debug!( + namespace_raw = self.namespace().inner(), + through = fence.through, + "the adopted view's log proves the held partition entries" + ); + self.view_fence = None; + // A Break earlier in this fence left it naming a dropped entry. + let head = self.consensus.sequencer().current_sequence(); + if let Some(header) = self.log.journal().inner.repair_header(head) { + self.consensus.set_last_prepare_checksum(header.checksum); + } + self.persist_repaired_prefix(); + } + ViewWindowVerdict::Hole { op } => { + if fence.hole != Some(op) { + debug!( + namespace_raw = self.namespace().inner(), + op, "held partition entries wait on op {op} before they can be proven" + ); + } + if let Some(fence) = self.view_fence.as_mut() { + fence.hole = Some(op); + } + } + ViewWindowVerdict::Break { op } => self.drop_unproven_from(op).await, + } + Some(verdict) + } + + async fn drop_unproven_from(&mut self, op: u64) { + // The evicted ring holds only the applied prefix, so a broken entry above + // `commit_min` is always resident. + debug_assert!(self.log.journal().inner.holds_op(op)); + let result = if self.log.journal().inner.holds_op(op) { + self.truncate_uncommitted_from(op).await.map(Some) + } else { + Ok(None) + }; + let removed = match result { + Ok(Some(removed)) => removed, + Ok(None) => { + error!( + namespace_raw = self.namespace().inner(), + op, + "partition entry at op {op} is not the adopted view's and is already \ + flushed; marking the partition fatal and shutting down" + ); + self.mark_view_fence_fatal(op); + return; + } + Err(error) => { + error!( + namespace_raw = self.namespace().inner(), + op, + %error, + "cannot drop partition entries from op {op} the adopted view's log does not \ + chain to; marking the partition fatal and shutting down" + ); + self.mark_view_fence_fatal(op); + return; + } + }; + warn!( + namespace_raw = self.namespace().inner(), + op, + removed, + view = self.consensus.view(), + "dropped {removed} partition entries from op {op}: the adopted view's log does not \ + chain to them" + ); + if let Some(fence) = self.view_fence.as_mut() { + fence.hole = None; + } + self.persist_repaired_prefix(); + } + + fn mark_view_fence_fatal(&mut self, op: u64) { + self.fatal = Some(FatalCommit { + namespace_raw: self.namespace().inner(), + op, + operation: None, + }); + } + + /// Whether the WAL replayed history at boot, consumed by the first view + /// adoption so it fences that history even with `log_view` unchanged. + pub fn take_recovered_unproven(&mut self) -> bool { + std::mem::take(&mut self.recovered_unproven) + } + /// Drop an uncommitted view-divergent suffix and restore every append cursor /// from the retained prefix as one write-locked operation. /// @@ -5976,6 +6276,11 @@ where let mut entries = self.log.journal().inner.resident_entries(); entries.sort_unstable_by_key(peek_op); + let removed_ops: HashSet<u64> = entries + .iter() + .map(peek_op) + .filter(|op| *op >= from_op) + .collect(); let mut retained_info = JournalInfo::default(); let mut retained_next_offset = 0; let mut rewind_next_offset = None; @@ -6062,8 +6367,11 @@ where } self.consensus.note_journal_mutation(); let commit_max = self.consensus.commit_max(); + // A staged offset outlives its entry only when the entry was never + // resident here: a removed op is refetched, and its stale staging would + // otherwise be applied in place of the view's. self.pending_consumer_offset_commits - .retain(|op, _| *op < from_op || *op <= commit_max); + .retain(|op, _| *op < from_op || (*op <= commit_max && !removed_ops.contains(op))); self.offset_reservations_need_resync.set(true); Ok(removed) } @@ -6147,6 +6455,7 @@ where } /// Returns false while the requested physical prefix is still pending in the WAL. + /// While a view fence is up it flushes no further than `commit_min`. #[allow(clippy::too_many_lines)] async fn commit_messages_inner( &mut self, @@ -6184,7 +6493,12 @@ where // the commit path panics the shard pump instead. All segment range / // stats / durable-offset accounting below is computed from the committed // entries, not the resident-journal snapshot above. - let commit_max = self.consensus.commit_max().min(through_op); + let mut commit_max = self.consensus.commit_max().min(through_op); + // Forced flushes (shutdown, transfer offers) reach here without the walk: + // nothing above an unproven window may land in a segment. + if self.view_fenced() { + commit_max = commit_max.min(self.consensus.commit_min()); + } let committed_entries = self.log.journal().inner.committed_prefix(commit_max); if committed_entries.is_empty() { if force { @@ -8612,7 +8926,7 @@ where } /// Ingest one repaired prepare: journal + stage it exactly like a live - /// replicated op, minus the view fence, the gap check, and the ack (the + /// replicated op, minus the view-number check, the gap check, and the ack (the /// op is already committed cluster-wide; there is nobody to ack to). The /// commit walk runs at `RepairDone`, after the floor is known. pub async fn apply_repaired_prepare(&mut self, message: Message<PrepareHeader>) { @@ -8970,6 +9284,11 @@ where if self.fatal.is_some() || self.materialization_missing { return false; } + // An ack above an unproven window can be wiped when the window turns out + // to break; the primary's retransmit re-drives it once the fence lifts. + if self.view_fenced() { + return false; + } // Durable-before-send: a PrepareOk implies this replica's // (view, log_view), so it must not leave until they are durable, or a // crash could recover an older view than the one this ack helped @@ -17492,6 +17811,303 @@ mod tests { .expect("journal append"); } + /// A sealed prepare at `op` chained to `parent`, journaled. `request` tells two + /// entries at the same op apart. + async fn journal_sealed( + partition: &IggyPartition<IggyMessageBus>, + op: u64, + parent: u128, + request: u64, + ) -> PrepareHeader { + let prepare = sealed_prepare(op, parent, request); + let header = *prepare.header(); + partition + .log + .journal() + .inner + .append(prepare.into_frozen()) + .await + .expect("journal append"); + header + } + + fn sealed_prepare(op: u64, parent: u128, request: u64) -> Message<PrepareHeader> { + let size = std::mem::size_of::<PrepareHeader>(); + Message::<PrepareHeader>::new(size).transmute_header(|_, header: &mut PrepareHeader| { + header.command = Command::Prepare; + header.client = message_bus::AUTO_COMMIT_CLIENT_ID; + header.session = 1; + header.request = request; + header.op = op; + header.parent = parent; + header.operation = Operation::CreateStream; + header.size = u32::try_from(size).expect("prepare header size fits in u32"); + header.checksum = header.identity_checksum(); + }) + } + + fn walked_ops(partition: &IggyPartition<IggyMessageBus>) -> Vec<u64> { + partition + .collect_committable_from_journal(COMMIT_WALK_OPS_MAX, &repair_config()) + .into_iter() + .map(|entry| entry.header.op) + .collect() + } + + #[compio::test] + async fn given_a_fenced_window_when_the_anchor_proves_it_should_release_the_walk() { + let mut partition = test_partition(); + let first = journal_sealed(&partition, 1, 0, 1).await; + let second = journal_sealed(&partition, 2, first.checksum, 2).await; + partition.consensus.restore_commit_state(0, 2); + partition.fence_view_window(2, None); + assert!( + walked_ops(&partition).is_empty(), + "the walk applied an entry the view has not vouched for" + ); + + partition.offer_view_anchor(sealed_prepare(3, second.checksum, 3).header()); + assert_eq!( + partition.prove_view_window().await, + Some(ViewWindowVerdict::Proven) + ); + assert!(!partition.view_fenced()); + assert_eq!(walked_ops(&partition), vec![1, 2]); + } + + #[compio::test] + async fn given_an_old_views_entry_under_the_anchor_when_proving_should_drop_it_and_stay_fenced() + { + let mut partition = test_partition(); + let first = journal_sealed(&partition, 1, 0, 1).await; + // The view's op 2, which the anchor chains to, is not what is held. + let canonical_second = sealed_prepare(2, first.checksum, 2); + journal_sealed(&partition, 2, first.checksum, 99).await; + partition.consensus.restore_commit_state(0, 2); + partition.fence_view_window( + 2, + Some(*sealed_prepare(3, canonical_second.header().checksum, 3).header()), + ); + + assert_eq!( + partition.prove_view_window().await, + Some(ViewWindowVerdict::Break { op: 2 }) + ); + assert!(!partition.log.journal().inner.holds_op(2)); + assert!(partition.log.journal().inner.holds_op(1)); + assert!( + walked_ops(&partition).is_empty(), + "op 1 is not proven until op 2 is refetched, so the walk must still hold" + ); + assert!(partition.view_fenced()); + } + + #[compio::test] + async fn given_a_hole_in_the_fenced_window_when_proving_should_truncate_nothing() { + let mut partition = test_partition(); + journal_sealed(&partition, 1, 0, 1).await; + let third = journal_sealed(&partition, 3, 7, 3).await; + partition.consensus.restore_commit_state(0, 3); + partition.fence_view_window(3, Some(*sealed_prepare(4, third.checksum, 4).header())); + + assert_eq!( + partition.prove_view_window().await, + Some(ViewWindowVerdict::Hole { op: 2 }) + ); + assert!( + partition.log.journal().inner.holds_op(1) && partition.log.journal().inner.holds_op(3), + "a hole is no evidence against the entries around it" + ); + assert_eq!(partition.view_fence_hole(), Some(2)); + assert!(partition.view_fenced()); + } + + #[compio::test] + async fn given_an_anchor_from_another_view_when_offered_should_refuse_it() { + let mut partition = test_partition(); + let first = journal_sealed(&partition, 1, 0, 1).await; + partition.consensus.restore_commit_state(0, 1); + partition.fence_view_window(1, None); + let mut other_view = *sealed_prepare(2, first.checksum, 2).header(); + other_view.view = partition.consensus.view() + 1; + assert!( + !partition.offer_view_anchor(&other_view), + "an anchor from a view the fence was not installed in vouches for the wrong log" + ); + assert!(partition.view_fence_unanchored()); + } + + #[compio::test] + async fn given_an_unsealed_anchor_when_fencing_should_stay_unanchored() { + let mut partition = test_partition(); + journal_sealed(&partition, 1, 0, 1).await; + partition.consensus.restore_commit_state(0, 1); + // What a `StartView` carries at a commit point the primary holds no + // header for: checksum and parent both zero. + let blank = PrepareHeader { + command: Command::Prepare, + op: 2, + ..Default::default() + }; + partition.fence_view_window(1, Some(blank)); + assert!( + partition.view_fence_unanchored(), + "a blank anchor links any window below it" + ); + assert!(!partition.offer_view_anchor(&blank)); + } + + /// A Break leaves `last_prepare_checksum` naming the entry it dropped, and the + /// refetch fills the op without moving the head; unless the proof refreshes + /// it, certification keeps failing on a quiet partition. + #[compio::test] + async fn given_a_stale_last_prepare_checksum_when_the_window_proves_should_certify() { + let (_directory, mut partition, held) = partition_with_pending_durable_ack().await; + let persistence = partition.persistence.as_ref().unwrap().clone(); + let view = partition.consensus().view() + 1; + partition.consensus.set_view(view); + partition.consensus().set_log_view(view); + partition.consensus().restore_commit_state(0, 1); + partition + .consensus() + .set_last_prepare_checksum(held.checksum ^ 1); + let anchor = checksummed_segment_prepare(2, held.checksum, 0, b"anchor"); + partition.fence_view_window(1, Some(*anchor.header())); + assert_eq!( + partition.prove_view_window().await, + Some(ViewWindowVerdict::Proven) + ); + partition.ensure_wal_view(); + persistence.drain_with_timeout().await.unwrap(); + assert_eq!(persistence.certified_log_view(), Some(view)); + } + + #[compio::test] + async fn given_a_fenced_wal_when_certifying_should_vouch_only_for_the_applied_prefix() { + let (_directory, mut partition, held) = partition_with_pending_durable_ack().await; + let persistence = partition.persistence.as_ref().unwrap().clone(); + let view = partition.consensus().view() + 1; + partition.consensus.set_view(view); + partition.consensus().set_log_view(view); + partition.consensus().sequencer().set_sequence(4); + + // A window reaching the suffix the next `DoViewChange` reports: muted. + partition.consensus().restore_commit_state(held.op, 3); + partition.fence_view_window(3, None); + assert!(!partition.ensure_wal_view()); + persistence.drain_with_timeout().await.unwrap(); + assert_ne!(persistence.certified_log_view(), Some(view)); + + // WAL holds nothing above `commit_min`, window below the suffix: certified. + partition.consensus().advance_commit_max(4); + partition.ensure_wal_view(); + persistence.drain_with_timeout().await.unwrap(); + assert_eq!(persistence.certified_log_view(), Some(view)); + } + + #[compio::test] + async fn given_an_unanchored_fence_when_ticking_should_ask_for_headers_at_the_pace_given() { + let mut partition = test_partition(); + journal_sealed(&partition, 1, 0, 1).await; + partition.consensus.restore_commit_state(0, 1); + partition.fence_view_window(1, None); + let due: Vec<bool> = (0..6).map(|_| partition.view_fence_probe_due(3)).collect(); + assert_eq!(due, vec![false, false, true, false, false, true]); + } + + #[compio::test] + async fn given_a_cached_hole_when_the_missing_op_arrives_should_bypass_the_cache() { + let mut partition = test_partition(); + journal_sealed(&partition, 1, 0, 1).await; + let third = journal_sealed(&partition, 3, 7, 3).await; + partition.consensus.restore_commit_state(0, 3); + partition.fence_view_window(3, Some(*sealed_prepare(4, third.checksum, 4).header())); + assert_eq!( + partition.prove_view_window().await, + Some(ViewWindowVerdict::Hole { op: 2 }) + ); + assert_eq!(partition.view_fence.and_then(|fence| fence.hole), Some(2)); + // Op 3 chains to parent 7, which the arriving op 2 is not. + journal_sealed(&partition, 2, 0, 99).await; + assert_eq!( + partition.prove_view_window().await, + Some(ViewWindowVerdict::Break { op: 2 }), + "the cache must be bypassed once the missing op arrives" + ); + } + + /// The fence holds entries back from the WAL too: a stale entry can sit at the + /// WAL head, and appending the view's successor on top of it fails the WAL's + /// parent check, which fences the whole partition. + #[compio::test] + async fn given_a_stale_wal_head_when_fenced_should_not_persist_repaired_successors() { + let (_directory, mut partition, stale) = partition_with_pending_durable_ack().await; + let persistence = partition.persistence.as_ref().unwrap().clone(); + assert_eq!(persistence.head(), stale.op); + let canonical = checksummed_segment_prepare(1, 0, 7, b"canonical"); + let successor = + checksummed_segment_prepare(2, canonical.header().checksum, 0, b"successor"); + partition + .log + .journal() + .inner + .append(successor.clone().into_frozen()) + .await + .unwrap(); + partition.consensus().sequencer().set_sequence(2); + partition.consensus().restore_commit_state(0, 2); + partition.fence_view_window(1, Some(*successor.header())); + + partition.persist_repaired_prefix(); + assert_eq!( + persistence.head(), + stale.op, + "a repaired successor reached the WAL above an unproven entry" + ); + assert!(persistence.failure().is_none()); + } + + #[compio::test] + async fn given_a_fenced_window_when_acking_should_withhold_the_ack() { + let (mut partition, _) = recording_partition_at(1, 3); + let prepare = sealed_prepare(1, 0, 1); + let first = *prepare.header(); + partition + .log + .journal() + .inner + .append(prepare.into_frozen()) + .await + .expect("journal append"); + partition.consensus.sequencer().set_sequence(1); + partition.consensus.restore_commit_state(0, 1); + assert!( + partition.send_prepare_ok(&first).await, + "an unfenced backup acks what it holds" + ); + partition.fence_view_window(1, None); + assert!( + !partition.send_prepare_ok(&first).await, + "an ack above an unproven window can be wiped when the window breaks" + ); + } + + #[compio::test] + async fn given_a_staged_offset_under_commit_max_when_its_entry_is_truncated_should_drop_the_staging() + { + let mut partition = test_partition(); + journal_prepare(&partition, 1, Operation::StoreConsumerOffset).await; + journal_prepare(&partition, 2, Operation::StoreConsumerOffset).await; + partition.consensus.sequencer().set_sequence(2); + partition.consensus.restore_commit_state(0, 2); + partition.stage_consumer_offset_upsert(2, ConsumerKind::Consumer, 8, 0, false); + partition.truncate_uncommitted_from(2).await.unwrap(); + assert!( + !partition.pending_consumer_offset_commits.contains_key(&2), + "the stale cursor would be applied in place of the view's op 2" + ); + } + /// Walking through the head advances `commit_min` past a resident entry only /// `on_ack` can pop and answer, after which every later ack finds the drain /// empty and no reply is ever shipped. @@ -18701,6 +19317,38 @@ mod tests { ); } + #[compio::test] + async fn given_a_fenced_window_when_flushing_should_write_nothing_above_commit_min() { + let dir = tempfile::tempdir().expect("temp dir"); + let log_path = dir.path().join("segment.log"); + let index_path = dir.path().join("segment.index"); + let mut fixture = PersistFixture::new( + log_path.to_str().expect("utf-8 path"), + index_path.to_str().expect("utf-8 path"), + ) + .await; + let partition = &mut fixture.partition; + partition.log.journal().inner.set_repair_retention(true); + partition.repair = Some(armed_session(3, 0, None)); + for op in 1..=3 { + partition + .apply_repaired_prepare(repaired_send_prepare(op, 0, u128::from(op))) + .await; + } + partition.consensus().advance_commit_max(3); + partition.fence_view_window(3, None); + + partition + .flush_committed_messages(&repair_config()) + .await + .expect("flush"); + assert!( + std::fs::read(&log_path).unwrap().is_empty(), + "a forced flush wrote unproven entries to the segment" + ); + assert!((1..=3).all(|op| partition.log.journal().inner.holds_op(op))); + } + #[compio::test] async fn given_flushed_repair_ahead_of_commit_min_when_replayed_should_persist_only_new_batches() { diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index 545377cd4..8b723c557 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -426,6 +426,22 @@ impl PartitionJournal<PartitionJournalMemStorage> { /// width of `ops`. Resident entries take precedence over retained repairs. #[must_use] pub fn repair_headers_in(&self, ops: RangeInclusive<u64>) -> BTreeMap<u64, PrepareHeader> { + self.headers_in(ops, false) + } + + /// Like [`Self::repair_headers_in`], but the FIRST resident header wins on a + /// duplicated op, as in [`Self::committed_headers_from`]: the entry proven is + /// the entry the commit walk would apply. + #[must_use] + pub fn walked_headers_in(&self, ops: RangeInclusive<u64>) -> BTreeMap<u64, PrepareHeader> { + self.headers_in(ops, true) + } + + fn headers_in( + &self, + ops: RangeInclusive<u64>, + first_wins: bool, + ) -> BTreeMap<u64, PrepareHeader> { let mut found = BTreeMap::new(); if ops.is_empty() { return found; @@ -433,7 +449,11 @@ impl PartitionJournal<PartitionJournalMemStorage> { { let headers = unsafe { &*self.headers.get() }; for header in headers.iter().filter(|header| ops.contains(&header.op)) { - found.insert(header.op, *header); + if first_wins { + found.entry(header.op).or_insert(*header); + } else { + found.insert(header.op, *header); + } } } let ring = unsafe { &*self.evicted_ring.get() }; diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs index 9ff71a20b..51950964f 100644 --- a/core/partitions/src/lib.rs +++ b/core/partitions/src/lib.rs @@ -42,6 +42,7 @@ pub mod segment_anchor; mod segment_recovery; pub mod state_transfer; mod types; +mod view_fence; pub use consumer_offset_capacity::{AutoCommitReservation, ConsumerOffsetCapacityError}; use iggy_binary_protocol::PrepareHeader; @@ -52,6 +53,7 @@ pub use iggy_index_writer::IggyIndexWriter; pub use iggy_partition::{IggyPartition, PurgeError, SegmentRemoval}; pub use iggy_partitions::IggyPartitions; pub use journal::{EVICTED_RING_BYTES_MAX, EVICTED_RING_CAPACITY}; +pub use view_fence::ViewWindowVerdict; /// Offsets a partition claims in its superblock ahead of the mint counter /// before it will append, so a crash-restarted replica resumes above every diff --git a/core/partitions/src/view_fence.rs b/core/partitions/src/view_fence.rs new file mode 100644 index 000000000..1ed971504 --- /dev/null +++ b/core/partitions/src/view_fence.rs @@ -0,0 +1,153 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Entries a replica journaled before adopting a newer view, held back from +//! commit and acknowledgement until the view's log proves them. + +use iggy_binary_protocol::{CHECKSUM_UNSEALED, PrepareHeader}; + +/// An unproven window `(commit_min, through]` installed in `view` and, once known, +/// that view's header at `through + 1`, whose `parent` the window must chain into. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ViewFence { + pub view: u32, + pub through: u64, + pub anchor: Option<PrepareHeader>, + /// The op the last proof stopped at for lack of an entry. Until it arrives + /// nothing below it can change, so a proof does not rescan the window. + pub hole: Option<u64>, + /// Ticks spent without an anchor, paced by [`crate::IggyPartition::view_fence_probe_due`]. + pub unanchored_ticks: u32, +} + +/// What the `parent` chain says about a fenced window. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ViewWindowVerdict { + /// Every op in the window links into the anchor. + Proven, + /// No entry at `op`, so nothing at or below it can be linked yet. + Hole { op: u64 }, + /// The entry at `op` belongs to a log the view discarded: it is not the one + /// the op above chains to, or it is not the copy the WAL flush would write. + Break { op: u64 }, +} + +/// Walk `(floor, through]` top-down from `anchor_parent`, the checksum the view's +/// entry at `through` must carry. +/// +/// An unsealed checksum on either side counts as a link, as in the shard's +/// `header_is_view_entry`: a pre-seal WAL entry carries no identity to compare. +#[must_use] +pub fn view_window_verdict( + floor: u64, + through: u64, + anchor_parent: u128, + header_at: impl Fn(u64) -> Option<PrepareHeader>, +) -> ViewWindowVerdict { + let mut expected = anchor_parent; + for op in (floor + 1..=through).rev() { + let Some(local) = header_at(op) else { + return ViewWindowVerdict::Hole { op }; + }; + let linked = local.checksum == expected + || local.checksum == CHECKSUM_UNSEALED + || expected == CHECKSUM_UNSEALED; + if !linked { + return ViewWindowVerdict::Break { op }; + } + expected = local.parent; + } + ViewWindowVerdict::Proven +} + +#[cfg(test)] +mod tests { + use super::{ViewWindowVerdict, view_window_verdict}; + use iggy_binary_protocol::{Command, Operation, PrepareHeader}; + use std::collections::BTreeMap; + + const FLOOR: u64 = 3; + const THROUGH: u64 = 6; + + fn sealed(op: u64, parent: u128, request: u64) -> PrepareHeader { + let mut header = PrepareHeader { + command: Command::Prepare, + operation: Operation::SendMessages, + op, + parent, + request, + ..Default::default() + }; + header.checksum = header.identity_checksum(); + header + } + + /// The view's log over `FLOOR + 1 ..= THROUGH + 1`, each entry chained to the + /// one below. The last entry is the anchor. + fn view_log() -> BTreeMap<u64, PrepareHeader> { + let mut parent = 0; + (FLOOR + 1..=THROUGH + 1) + .map(|op| { + let header = sealed(op, parent, op); + parent = header.checksum; + (op, header) + }) + .collect() + } + + fn verdict(log: &BTreeMap<u64, PrepareHeader>) -> ViewWindowVerdict { + let anchor = log[&(THROUGH + 1)]; + view_window_verdict(FLOOR, THROUGH, anchor.parent, |op| log.get(&op).copied()) + } + + #[test] + fn given_the_views_own_entries_when_proving_should_report_proven() { + assert_eq!(verdict(&view_log()), ViewWindowVerdict::Proven); + } + + #[test] + fn given_an_old_views_entry_under_the_anchor_when_proving_should_report_its_op() { + let mut log = view_log(); + // Same op, different request: what an isolated old primary journals. + log.insert(FLOOR + 1, sealed(FLOOR + 1, 0, 99)); + assert_eq!( + verdict(&log), + ViewWindowVerdict::Break { op: FLOOR + 1 }, + "a stale entry under the anchor was accepted as the view's" + ); + } + + #[test] + fn given_a_hole_under_the_anchor_when_proving_should_report_the_hole_not_a_break() { + let mut log = view_log(); + log.remove(&(FLOOR + 2)); + log.insert(FLOOR + 1, sealed(FLOOR + 1, 0, 99)); + assert_eq!( + verdict(&log), + ViewWindowVerdict::Hole { op: FLOOR + 2 }, + "a hole is missing data, not evidence against the entry below it" + ); + } + + #[test] + fn given_an_empty_window_when_proving_should_report_proven() { + assert_eq!( + view_window_verdict(THROUGH, THROUGH, 0, |_| None), + ViewWindowVerdict::Proven + ); + } +} diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index eff6cf5da..905c687b3 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -4799,6 +4799,7 @@ where return; }; partition.ensure_materialization_recovery(); + let prior_log_view = partition.consensus().log_view(); let actions = partition .consensus() @@ -4814,8 +4815,13 @@ where // like it, pending-less adoptions (empty StartView suffix) still sweep // the relics above the adopted head. let pending = partition.consensus().pending_view_log(); - reconcile_partition_view_divergence(self.id, partition, pending.as_ref()).await; + // Consumed on every adoption, so a restart's flag never outlives the first. + let recovered = partition.take_recovered_unproven(); + let fence_below = recovered || prior_log_view < partition.consensus().view(); + reconcile_partition_view_divergence(self.id, partition, pending.as_ref(), fence_below) + .await; } + anchor_view_fence_from_start_view(self.id, partition, &header, suffix_body).await; let consensus = partition.consensus(); let (local_actions, wire_actions) = split_local_actions(actions); // Locals go to the partition dispatcher ONLY: `RebuildPipeline` @@ -4961,6 +4967,7 @@ where if !partition.requires_state_transfer() && partition.persist_superblock_if_needed().await { + refresh_partition_dvc_suffix(partition); respond_start_view::<B, _, MJ>(consensus).await; } } @@ -5000,6 +5007,9 @@ where else { return; }; + // The answer carries this suffix: empty, it names no header the prober can + // check its own entries against. + refresh_partition_dvc_suffix(partition); let consensus = partition.consensus(); let actions = consensus.handle_request_start_view(PlaneKind::Partitions, &header); let (local_actions, wire_actions) = split_local_actions(actions); @@ -5144,7 +5154,13 @@ where let Some(partition) = planes.1.0.get_mut_by_ns(&namespace) else { return; }; - if !partition.consensus().is_normal() { + if !partition_serves_repair( + partition.consensus().status(), + header.replica, + partition + .consensus() + .primary_index(partition.consensus().view()), + ) { return; } let cluster = partition.consensus().cluster(); @@ -5183,10 +5199,15 @@ where // until the body is journaled, and the primary's retransmit is dropped // by the backup gap check (adoption already advanced its sequencer to // the head), so repair is the only channel that can deliver them. - let to_op = repair_serve_ceiling( - header.to_op, - partition.consensus().commit_max(), - partition.consensus().sequencer().current_sequence(), + let to_op = fenced_serve_ceiling( + repair_serve_ceiling( + header.to_op, + partition.consensus().commit_max(), + partition.consensus().sequencer().current_sequence(), + ), + partition + .view_fenced() + .then(|| partition.consensus().commit_min()), ); // `None` means the journal holds NOTHING, not "nothing was evicted": // the partition journal is memory-only and `clear_all` wipes the @@ -5960,7 +5981,13 @@ where // already holds a header for, so the scan would report a gap nothing // fills. Backups reach this on StartView adoption; a primary-elect has no // adoption to hang it off. - reconcile_partition_view_divergence(self.id, partition, Some(&pending)).await; + // `log_view` still names the old log here: it is raised only when the + // view starts. + let recovered = partition.take_recovered_unproven(); + let fence_below = + recovered || partition.consensus().log_view() < partition.consensus().view(); + reconcile_partition_view_divergence(self.id, partition, Some(&pending), fence_below) + .await; let consensus = partition.consensus(); // Identity, not presence: see the metadata twin. The floor is the local // commit point, the partition twin of the metadata snapshot floor: @@ -6007,6 +6034,11 @@ where .await; return; } + // Everything is held, but not yet proven to be the view's: starting + // now commits whatever this replica journaled in an older view. + if partition.view_fenced() { + return; + } let actions = consensus.start_pending_view(PlaneKind::Partitions); let (local_actions, wire_actions) = split_local_actions(actions); @@ -8029,7 +8061,9 @@ where repairs_live += 1; } let probe = partition_gap_probe(partition); - let walk_stalled = group_is_walk_stalled(&probe); + // A fenced walk takes nothing until the window is proven, and the + // events that prove it (repair, an anchor) re-drive it themselves. + let walk_stalled = group_is_walk_stalled(&probe) && !partition.view_fenced(); // The RATE cap only. The concurrency cap lives in the arm fn, // which is the funnel every arming site goes through; resolved // before the debounce either way, so a refusal keeps the group @@ -8131,6 +8165,25 @@ where } } + // A fence with no anchor stays shut until the view's primary names the + // header above it; ask again rather than wait for the next prepare. + if let Some(partition) = partitions.get_mut_by_ns(&namespace) + && !partition + .consensus() + .is_primary_for_view(partition.consensus().view()) + && partition.view_fence_probe_due(self.repair_retry_ticks.get()) + { + // A probe promises nothing about this replica's view, so it skips the + // superblock gate, which cannot certify while the fence holds the + // WAL back. + let consensus = partition.consensus(); + let probe = [VsrAction::SendRequestStartView { + view: consensus.view(), + group: consensus.group(), + }]; + dispatch_vsr_actions::<B, _, MJ>(consensus, None, &probe).await; + } + // Transfer stall retry: descriptor and chunk frames are // fire-and-forget, so a lost one must not wedge the session (and // the rejoin behind it) forever. Budget-bounded: a peer that died @@ -10835,6 +10888,23 @@ fn repair_op_in_scope( }) } +/// Whether a partition replica in `status` answers a repair request from +/// `requester`. `ViewChange` only for this view's primary-elect, whose fence +/// proves what it is served: the senders it repairs from wait on its `StartView`. +const fn partition_serves_repair(status: Status, requester: u8, view_primary: u8) -> bool { + match status { + Status::Normal => true, + Status::ViewChange => requester == view_primary, + Status::Recovering => false, + } +} + +/// A fenced replica never hands out entries it has not proven: its serve stops at +/// `fenced_floor`, its own commit point. +fn fenced_serve_ceiling(ceiling: u64, fenced_floor: Option<u64>) -> u64 { + fenced_floor.map_or(ceiling, |floor| ceiling.min(floor)) +} + /// Ceiling on the op range a repair request may ask this replica to walk. /// /// Not `commit_max` alone: a new primary repairing toward a merged log needs the @@ -11228,7 +11298,10 @@ where .log .journal() .inner - .holds_op(commit_min.saturating_add(1)); + .holds_op(commit_min.saturating_add(1)) + // The fenced commit walk cannot reach a hole above the next op, so only + // repair fills it. + && partition.view_fence_hole().is_none(); // Same discipline, one guard deeper: the suffix test walks the header vec, // so it runs only for a group that HAS an unfinished suffix and already // owes nothing else. @@ -11422,11 +11495,16 @@ fn build_dvc_suffix( /// duplicate header and rewrites `op_to_storage_offset`, and `committed_prefix` walks /// positionally, so the stale entry is what `evict_prefix` flushes to the segment: /// durable divergent bytes, no error anywhere. +/// +/// Below the view's canonical headers it cannot compare anything, so with +/// `fence_below` (an adoption that replaces this replica's log) it fences that +/// window instead; see [`fence_unproven_view_window`]. #[allow(clippy::future_not_send)] async fn reconcile_partition_view_divergence<B, SB>( shard: u16, partition: &mut IggyPartition<B, SB>, pending: Option<&MergedLog>, + fence_below: bool, ) where B: MessageBus, SB: journal::superblock::SuperblockStore, @@ -11481,9 +11559,26 @@ async fn reconcile_partition_view_divergence<B, SB>( repairable_from = Some(repairable_from.map_or(above_head, |op| op.min(above_head))); } - let Some(from_op) = repairable_from else { - return; - }; + if let Some(from_op) = repairable_from { + drop_diverging_partition_suffix(shard, partition, from_op, op_head).await; + } + if fence_below { + fence_unproven_view_window(partition, pending).await; + } +} + +/// Truncate the entries from `from_op` that the view's log names differently or +/// does not name at all. +#[allow(clippy::future_not_send)] +async fn drop_diverging_partition_suffix<B, SB>( + shard: u16, + partition: &mut IggyPartition<B, SB>, + from_op: u64, + op_head: u64, +) where + B: MessageBus, + SB: journal::superblock::SuperblockStore, +{ match partition.truncate_uncommitted_from(from_op).await { Ok(removed) => { tracing::warn!( @@ -11511,6 +11606,101 @@ async fn reconcile_partition_view_divergence<B, SB>( } } +/// The lowest header in a view's suffix that can anchor a fence. A blank slot, or +/// an unsealed one, names no identity: its zero `parent` would link anything +/// below it. Op 1 is the one header whose real parent is zero. +fn view_fence_anchor(headers: &[PrepareHeader]) -> Option<PrepareHeader> { + headers + .iter() + .filter(|header| { + matches!(dvc_header_kind(header), DvcHeaderKind::Valid) + && header.checksum != CHECKSUM_UNSEALED + && (header.parent != CHECKSUM_UNSEALED || header.op == 1) + }) + .min_by_key(|header| header.op) + .copied() +} + +/// Fence what this replica holds between its commit point and the view's lowest +/// canonical header (or the adopted head, when the `StartView` named none), and +/// try to prove it right away. +/// +/// Those entries can come from a log the view discarded, and nothing else +/// compares them before the commit walk applies them. Fencing, not truncating: a +/// hole or a missing anchor is no evidence against an entry, and truncating on +/// either drops entries this replica may be the only one to hold. An empty window +/// is fenced too, so whatever repair puts there is proven before it is applied. +#[allow(clippy::future_not_send)] +async fn fence_unproven_view_window<B, SB>( + partition: &mut IggyPartition<B, SB>, + pending: Option<&MergedLog>, +) where + B: MessageBus, + SB: journal::superblock::SuperblockStore, +{ + let anchor = pending.and_then(|pending| view_fence_anchor(&pending.headers)); + // Without an anchor the window runs to the adopted head, which the pending log + // names; the sequencer can sit above it after an above-head truncation. + let through = match (anchor, pending) { + (Some(anchor), _) => anchor.op.saturating_sub(1), + (None, Some(pending)) => pending.op_head, + (None, None) => partition.consensus().sequencer().current_sequence(), + }; + partition.fence_view_window(through, anchor); + partition.prove_view_window().await; +} + +/// From this view's primary, and not behind what this replica already holds: an +/// older frame would truncate above its own head. +const fn start_view_may_anchor( + (frame_view, sender, frame_op): (u32, u8, u64), + view: u32, + primary: u8, + head: u64, +) -> bool { + frame_view == view && sender == primary && frame_op >= head +} + +/// Hand a fence still waiting for its anchor the headers a same-view `StartView` +/// carries: the probe answer a fenced replica asked for. +#[allow(clippy::future_not_send)] +async fn anchor_view_fence_from_start_view<B, SB>( + shard: u16, + partition: &mut IggyPartition<B, SB>, + header: &StartViewHeader, + suffix_body: &[u8], +) where + B: MessageBus, + SB: journal::superblock::SuperblockStore, +{ + let consensus = partition.consensus(); + if !partition.view_fence_unanchored() + || !start_view_may_anchor( + (header.view, header.replica, header.op), + consensus.view(), + consensus.primary_index(header.view), + consensus.sequencer().current_sequence(), + ) + { + return; + } + let pending = match MergedLog::from_start_view(header, suffix_body) { + Ok(Some(pending)) => pending, + Ok(None) => return, + Err(error) => { + tracing::warn!( + shard, + namespace_raw = consensus.group(), + view = header.view, + %error, + "start_view suffix did not decode, so the fence keeps waiting for an anchor" + ); + return; + } + }; + reconcile_partition_view_divergence(shard, partition, Some(&pending), true).await; +} + /// Whether a locally journaled header IS the entry the view's log names at that op. /// /// Identity, not presence: otherwise a stale prepare at the right op reads as @@ -12439,6 +12629,85 @@ mod view_coverage_tests { } } +#[cfg(test)] +mod view_fence_gate_tests { + //! The gates around the partition view fence, apart from the partition. + + use super::{ + fenced_serve_ceiling, partition_serves_repair, start_view_may_anchor, view_fence_anchor, + }; + use consensus::{Status, dvc_blank}; + use iggy_binary_protocol::{Command, Operation, PrepareHeader}; + + fn sealed(op: u64, parent: u128) -> PrepareHeader { + let mut header = PrepareHeader { + command: Command::Prepare, + operation: Operation::SendMessages, + op, + parent, + ..Default::default() + }; + header.checksum = header.identity_checksum(); + header + } + + #[test] + fn given_a_blank_commit_point_when_choosing_an_anchor_should_take_the_lowest_sealed_header() { + let above = sealed(6, 7); + let headers = [sealed(7, above.checksum), above, dvc_blank(5)]; + assert_eq!( + view_fence_anchor(&headers).map(|anchor| anchor.op), + Some(6), + "a blank's zero parent links any window below it" + ); + assert_eq!(view_fence_anchor(&[dvc_blank(5)]), None); + } + + #[test] + fn given_op_one_when_choosing_an_anchor_should_accept_its_zero_parent() { + assert_eq!( + view_fence_anchor(&[sealed(1, 0)]).map(|anchor| anchor.op), + Some(1) + ); + assert_eq!( + view_fence_anchor(&[sealed(2, 0)]), + None, + "a zero parent above op 1 vouches for nothing" + ); + } + + #[test] + fn given_a_start_view_when_deciding_whether_it_anchors_should_require_this_views_primary_at_or_above_the_head() + { + let start_view = |view: u32, replica: u8, op: u64| (view, replica, op); + assert!(start_view_may_anchor(start_view(2, 1, 9), 2, 1, 9)); + assert!(!start_view_may_anchor(start_view(1, 1, 9), 2, 1, 9)); + assert!(!start_view_may_anchor(start_view(2, 0, 9), 2, 1, 9)); + assert!( + !start_view_may_anchor(start_view(2, 1, 8), 2, 1, 9), + "an older frame truncates above its own head" + ); + } + + #[test] + fn given_a_view_change_when_asked_for_repair_should_serve_only_the_primary_elect() { + assert!(partition_serves_repair(Status::Normal, 2, 1)); + assert!(partition_serves_repair(Status::ViewChange, 1, 1)); + assert!( + !partition_serves_repair(Status::ViewChange, 2, 1), + "an unfenced requester would take this replica's unproven entries" + ); + assert!(!partition_serves_repair(Status::Recovering, 1, 1)); + } + + #[test] + fn given_a_fenced_server_when_capping_a_repair_should_stop_at_its_commit_point() { + assert_eq!(fenced_serve_ceiling(9, None), 9); + assert_eq!(fenced_serve_ceiling(9, Some(4)), 4); + assert_eq!(fenced_serve_ceiling(3, Some(4)), 3); + } +} + #[cfg(test)] mod dvc_suffix_window_tests { //! The suffix window's floor is a scan bound, not a commit point. diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index c176dd5d2..d73c3172f 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -1642,6 +1642,9 @@ fn materialise_partition( #[cfg(test)] mod consumer_offset_history_tests; +#[cfg(test)] +mod stale_prepare_commit_tests; + #[cfg(test)] mod tests { use super::partition_repair_driver_tests::cluster; diff --git a/core/simulator/src/stale_prepare_commit_tests.rs b/core/simulator/src/stale_prepare_commit_tests.rs new file mode 100644 index 000000000..b7b4c8b0f --- /dev/null +++ b/core/simulator/src/stale_prepare_commit_tests.rs @@ -0,0 +1,325 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! A replica must never commit a prepare the cluster did not commit. +//! +//! The old primary journals an op, is cut off before any peer sees it, and the +//! other two replicas elect a new view that commits a DIFFERENT prepare at the +//! same op. Whether the old primary rejoins that view or is later elected over +//! it, its stale prepare sits below the view's canonical headers, and it must +//! end up serving the view's entry at that offset, not its own. + +use bytes::Bytes; +use consensus::{PartitionsHandle, Status}; +use futures::FutureExt; +use iggy_common::PollingStrategy; +use partitions::{PollingArgs, PollingConsumer}; +use server_common::send_messages::{BatchIntegrity, decode_batch_slice_with}; +use server_common::sharding::IggyNamespace; + +use crate::Simulator; +use crate::client::SimClient; +use crate::packet::{PacketSimulatorOptions, ProcessId}; + +const REPLICAS: u8 = 3; +const CLIENT_ID: u128 = 1; +const OLD_PRIMARY: u8 = 0; +const STEPS_PER_SEND: usize = 12; +/// Well past `NORMAL_HEARTBEAT_TICKS` (500), so the majority elects. +const ELECTION_STEPS: usize = 5_000; +const SETTLE_STEPS: usize = 5_000; +/// Long enough for a lone survivor's view changes to rotate onto the old primary. +const REJOIN_STEPS: usize = 40_000; + +fn cluster(seed: u64) -> (Simulator, SimClient) { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + let options = PacketSimulatorOptions { + node_count: REPLICAS, + client_count: 1, + seed, + ..PacketSimulatorOptions::default() + }; + let sim = Simulator::new(usize::from(REPLICAS), std::iter::once(CLIENT_ID), options); + (sim, SimClient::new(CLIENT_ID)) +} + +fn group_state(sim: &Simulator, replica: u8, namespace: IggyNamespace) -> (Status, u32, u64, u64) { + let partitions = sim.replicas[usize::from(replica)] + .partition_shard(namespace) + .plane + .partitions(); + let partition = partitions + .get_by_ns(&namespace) + .expect("replica hosts the group"); + let consensus = partition.consensus(); + ( + consensus.status(), + consensus.view(), + consensus.commit_min(), + consensus.commit_max(), + ) +} + +fn send(sim: &mut Simulator, client: &SimClient, namespace: IggyNamespace, to: u8, payload: &str) { + let request = client.send_messages(namespace, &[Bytes::from(payload.to_owned())]); + sim.submit_request(client.client_id(), to, request.into_generic()); + for _ in 0..STEPS_PER_SEND { + sim.step(); + } +} + +fn isolate(sim: &mut Simulator, replica: u8, isolated: bool) { + for peer in (0..REPLICAS).filter(|peer| *peer != replica) { + sim.network.set_link_filter( + ProcessId::Replica(replica), + ProcessId::Replica(peer), + !isolated, + ); + sim.network.set_link_filter( + ProcessId::Replica(peer), + ProcessId::Replica(replica), + !isolated, + ); + } +} + +/// `(offset, payload)` for every message a consumer can poll from `replica`, +/// read through the ordinary client poll path. +fn polled(sim: &mut Simulator, replica: u8, namespace: IggyNamespace) -> Vec<(u64, String)> { + let mut out = Vec::new(); + loop { + let next = out.last().map_or(0, |(offset, _)| offset + 1); + let args = PollingArgs::new(PollingStrategy::offset(next), 1_000, false); + let poll = sim.poll_messages( + usize::from(replica), + namespace, + PollingConsumer::Consumer(1, 0), + &args, + ); + let (tx, rx) = shard::channel(1); + sim.executor.spawn(async move { + let _ = tx.try_send(poll.await); + }); + sim.run_pumps(); + let fragments = rx + .recv() + .now_or_never() + .expect("poll completes") + .expect("reply channel open") + .expect("poll accepted"); + let before = out.len(); + for fragment in &fragments { + let mut bytes = fragment.as_slice(); + while let Ok(batch) = decode_batch_slice_with(bytes, BatchIntegrity::LayoutOnly) { + for message in &batch { + out.push(( + batch.header.base_offset + u64::from(message.header.offset_delta), + String::from_utf8_lossy(message.payload).into_owned(), + )); + } + let Some(rest) = bytes.get(batch.header.total_size()..) else { + break; + }; + bytes = rest; + } + } + if out.len() == before { + return out; + } + } +} + +#[test] +fn given_an_old_primary_holding_an_uncommitted_prepare_when_it_rejoins_a_view_that_committed_another_prepare_at_that_op_should_not_commit_its_own() + { + let (mut sim, client) = cluster(0x5EED_57A1); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + + for index in 0..3 { + send( + &mut sim, + &client, + namespace, + OLD_PRIMARY, + &format!("warmup-{index}"), + ); + } + let (_, _, warm_commit, _) = group_state(&sim, OLD_PRIMARY, namespace); + assert_eq!(warm_commit, 3, "warmup did not commit on the old primary"); + + // Cut the old primary off, then hand it a request: it journals op 4 and + // can never replicate it. + isolate(&mut sim, OLD_PRIMARY, true); + send(&mut sim, &client, namespace, OLD_PRIMARY, "stale"); + + // The majority elects a new view without it. + let new_view = (0..ELECTION_STEPS) + .find_map(|_| { + sim.step(); + let (status, view, _, _) = group_state(&sim, 1, namespace); + (status == Status::Normal && view > 0).then_some(view) + }) + .expect("replicas 1 and 2 never elected a new view"); + let new_primary = primary_of(&sim, 1, namespace, new_view); + assert_ne!(new_primary, OLD_PRIMARY); + + // The new view commits different prepares at op 4 and beyond. + for index in 0..4 { + send( + &mut sim, + &client, + namespace, + new_primary, + &format!("fresh-{index}"), + ); + } + let (_, _, majority_commit, _) = group_state(&sim, new_primary, namespace); + assert!( + majority_commit >= 5, + "the new view committed only through {majority_commit}" + ); + + isolate(&mut sim, OLD_PRIMARY, false); + for _ in 0..SETTLE_STEPS { + sim.step(); + } + let (_, _, rejoined_commit, _) = group_state(&sim, OLD_PRIMARY, namespace); + assert!( + rejoined_commit >= majority_commit, + "the old primary never caught up: committed {rejoined_commit} of {majority_commit}" + ); + + let reference = polled(&mut sim, new_primary, namespace); + let rejoined = polled(&mut sim, OLD_PRIMARY, namespace); + assert!( + !reference.iter().any(|(_, payload)| payload == "stale"), + "the new view committed the isolated primary's request: {reference:?}" + ); + assert_eq!( + rejoined, reference, + "the rejoined replica serves a committed log that differs from the view's" + ); +} + +/// The primary-elect twin: the old primary never adopts the majority's view. It is +/// elected later, with its stale op under the merged log's commit point, where the +/// merged headers do not reach. +#[test] +fn given_an_old_primary_holding_an_uncommitted_prepare_when_it_is_elected_over_a_log_that_committed_another_prepare_should_not_commit_its_own() + { + let (mut sim, client) = cluster(0x5EED_57A1); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + for index in 0..3 { + send( + &mut sim, + &client, + namespace, + OLD_PRIMARY, + &format!("warmup-{index}"), + ); + } + isolate(&mut sim, OLD_PRIMARY, true); + send(&mut sim, &client, namespace, OLD_PRIMARY, "stale"); + let new_view = (0..ELECTION_STEPS) + .find_map(|_| { + sim.step(); + let (status, view, _, _) = group_state(&sim, 1, namespace); + (status == Status::Normal && view > 0).then_some(view) + }) + .expect("replicas 1 and 2 never elected a new view"); + let new_primary = primary_of(&sim, 1, namespace, new_view); + for index in 0..4 { + send( + &mut sim, + &client, + namespace, + new_primary, + &format!("fresh-{index}"), + ); + } + let (_, _, majority_commit, _) = group_state(&sim, new_primary, namespace); + assert!( + majority_commit >= 5, + "the new view committed only through {majority_commit}" + ); + + // Cut the new primary off and keep the old one apart until the survivor's view + // changes reach a view the old primary leads; then let those two meet. + let survivor = (0..REPLICAS) + .find(|replica| *replica != OLD_PRIMARY && *replica != new_primary) + .expect("three replicas"); + isolate(&mut sim, new_primary, true); + (0..REJOIN_STEPS) + .find(|_| { + sim.step(); + let (_, view, _, _) = group_state(&sim, survivor, namespace); + primary_of(&sim, survivor, namespace, view) == OLD_PRIMARY + }) + .expect("the survivor never reached a view the old primary leads"); + connect(&mut sim, OLD_PRIMARY, survivor); + let led_view = (0..SETTLE_STEPS) + .find_map(|_| { + sim.step(); + let (status, view, _, _) = group_state(&sim, OLD_PRIMARY, namespace); + (status == Status::Normal + && view > new_view + && primary_of(&sim, OLD_PRIMARY, namespace, view) == OLD_PRIMARY) + .then_some(view) + }) + .expect("the old primary never led a view: the primary-elect path did not run"); + + isolate(&mut sim, new_primary, false); + for _ in 0..SETTLE_STEPS { + sim.step(); + } + let reference = polled(&mut sim, survivor, namespace); + let elected = polled(&mut sim, OLD_PRIMARY, namespace); + assert!( + !reference.iter().any(|(_, payload)| payload == "stale"), + "view {led_view} committed the isolated primary's request: {reference:?}" + ); + assert_eq!( + elected, reference, + "the elected old primary serves a committed log that differs from the survivor's" + ); +} + +fn primary_of(sim: &Simulator, replica: u8, namespace: IggyNamespace, view: u32) -> u8 { + sim.replicas[usize::from(replica)] + .partition_shard(namespace) + .plane + .partitions() + .get_by_ns(&namespace) + .expect("replica hosts the group") + .consensus() + .primary_index(view) +} + +fn connect(sim: &mut Simulator, left: u8, right: u8) { + sim.network + .set_link_filter(ProcessId::Replica(left), ProcessId::Replica(right), true); + sim.network + .set_link_filter(ProcessId::Replica(right), ProcessId::Replica(left), true); +}
