address review
diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs index afccb0c..1c1a8d8 100644 --- a/core/binary_protocol/src/consensus/header.rs +++ b/core/binary_protocol/src/consensus/header.rs
@@ -1237,6 +1237,14 @@ covered.view = 0; u128::from(twox_hash::XxHash3_64::oneshot(bytemuck::bytes_of(&covered))) } + + /// Whether a parent chain can be proven down from this header. An unsealed + /// header names no identity, and a zero `parent` would link anything below it; + /// op 1 is the one prepare whose real parent is zero. + #[must_use] + pub const fn anchors_chain(&self) -> bool { + self.checksum != CHECKSUM_UNSEALED && (self.parent != CHECKSUM_UNSEALED || self.op == 1) + } } // RepairPrepareHeader - repair peer -> recovering replica (journal repair)
diff --git a/core/consensus/src/dvc_merge.rs b/core/consensus/src/dvc_merge.rs index 56aae9e..888a0f6 100644 --- a/core/consensus/src/dvc_merge.rs +++ b/core/consensus/src/dvc_merge.rs
@@ -92,25 +92,21 @@ impl MergedLog { /// The log a `StartView` announces: its suffix slots verbatim, blanks included, - /// or `None` when it carries numbers only. + /// and no headers 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> { + ) -> Result<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 { + Ok(Self { op_head: header.op, commit_max: header.commit, - headers: headers.to_vec(), + headers: suffix.headers().to_vec(), committed_elsewhere: Vec::new(), - })) + }) } }
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index b2f16c3..1329f05 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs
@@ -3538,18 +3538,12 @@ /// correction). fn adopt_start_view_suffix(&self, header: &StartViewHeader, suffix_body: &[u8]) -> u64 { match MergedLog::from_start_view(header, suffix_body) { - Ok(Some(log)) => *self.pending_view_log.borrow_mut() = Some(log), // A numbers-only StartView still has to replace the pending log while // the held history is unproven, so the proof waits on this view's head. - Ok(None) if self.view_log_unproven() => { - *self.pending_view_log.borrow_mut() = Some(MergedLog { - op_head: header.op, - commit_max: header.commit, - headers: Vec::new(), - committed_elsewhere: Vec::new(), - }); + Ok(log) if !log.headers.is_empty() || self.view_log_unproven() => { + *self.pending_view_log.borrow_mut() = Some(log); } - Ok(None) => {} + Ok(_) => {} Err(error) => { tracing::warn!( replica = self.replica,
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 7d8bfc4..1c9a2c1 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs
@@ -66,8 +66,7 @@ SendMessagesConfirmationResponse, SendMessagesResponse, }; use iggy_binary_protocol::{ - AckLevel, CHECKSUM_UNSEALED, Command, Operation, PrepareHeader, WireDecode, WireEncode, - WireIdentifier, + AckLevel, Command, Operation, PrepareHeader, WireDecode, WireEncode, WireIdentifier, }; use iggy_binary_protocol::{PrepareOkHeader, ReplyHeader, RoutedRequestHeader}; use iggy_common::{ @@ -102,7 +101,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use tokio::sync::Mutex as TokioMutex; -use tracing::{debug, error, info, warn}; +use tracing::{debug, error, warn}; /// Which of a partition's offset counters are live. /// @@ -5677,6 +5676,17 @@ self.notify_io(); } + /// Run a view adoption's `actions` once the pending truncation lands, so the + /// pipeline rebuild reads the journal after the cut. + pub fn defer_view_adoption(&mut self, adoption: Vec<consensus::VsrAction>, from: u8) { + if let Some(PendingPartitionTransition::Truncate { actions, peer, .. }) = + &mut self.transition + { + actions.extend(adoption); + *peer = Some(from); + } + } + #[must_use] pub const fn view_truncation_pending(&self) -> bool { matches!( @@ -5902,8 +5912,15 @@ else { unreachable!("truncate transition exists"); }; - let unproven_drop = self.unproven_drop_from.take() == Some(*ops.start()); let from_op = *ops.start(); + // Another cut that lands first leaves the marked entries resident, so the + // marker stays for the proof that finds them again. + let unproven_drop = self + .unproven_drop_from + .is_some_and(|marked| from_op <= marked); + if unproven_drop { + self.unproven_drop_from = None; + } let removed = self.remove_resident_range(ops).await?; if unproven_drop { self.note_unproven_dropped(from_op, removed); @@ -8444,7 +8461,9 @@ } // After the in-flight commit: a proof that breaks truncates the journal. if self.view_fenced() { - self.prove_view_window().await; + if !self.view_proof_deferred() { + self.prove_view_window().await; + } if self.view_fenced() && self.recovered_commit <= self.consensus.commit_min() { return; } @@ -8859,16 +8878,11 @@ 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) - }) + .filter(PrepareHeader::anchors_chain) .or_else(|| previous.and_then(|fence| fence.anchor)); if previous.is_none() { - info!( + debug!( namespace_raw = self.namespace().inner(), view, commit_min = self.consensus.commit_min(), @@ -8906,8 +8920,12 @@ self.view_fence.as_ref().is_some_and(|fence| { fence.anchor.is_some() && fence.through > self.consensus.commit_min() - && self.log.journal().inner.holds_op(fence.through) - && self.held_through() >= fence.through + && self + .log + .journal() + .inner + .held_through_until(self.consensus.commit_min(), fence.through) + >= fence.through }) } @@ -8917,13 +8935,20 @@ self.fatal.is_none() && !self.materialization_missing && !self.persistence_checkpoint_pending() - // A Break found while busy cuts nothing (`drop_unproven_from`), so - // walking the window then only rebuilds its header map to break - // again. A pending cut is history work too. - && !self.history_is_busy() + // The commit walk returns before the proof while the view's log is unproven. + && !self.consensus.view_log_unproven() + && !self.view_proof_deferred() && self.view_fence_held() } + /// Whether a proof now would only find the same `Break` again: a Break found + /// while history work runs or a transition is queued cuts nothing + /// (`drop_unproven_from`), so walking the window rebuilds its header map for + /// nothing. + fn view_proof_deferred(&self) -> bool { + self.history_is_busy() || self.transition.is_some() + } + /// An installed snapshot replaces the history the recovered commit point named. pub(crate) const fn forget_recovered_commit(&mut self) { self.recovered_commit = 0; @@ -8986,8 +9011,7 @@ && fence.anchor.is_none() && fence.view == header.view && header.op == fence.through + 1 - && header.checksum != CHECKSUM_UNSEALED - && (header.parent != CHECKSUM_UNSEALED || header.op == 1) + && header.anchors_chain() { fence.anchor = Some(*header); fence.hole = None; @@ -9108,9 +9132,9 @@ } if self.has_io_dispatcher() { // The I/O lane owns a mounted partition's WAL. While other history work - // is in flight the fence keeps holding, and the next proof finds the - // same break. - if !self.history_is_busy() { + // is in flight or a transition is queued the fence keeps holding, and + // the next proof finds the same break. + if !self.view_proof_deferred() { self.unproven_drop_from = Some(op); self.defer_view_truncation(op..=u64::MAX, Vec::new(), None); } @@ -9143,9 +9167,6 @@ if let Some(fence) = self.view_fence.as_mut() { fence.hole = None; } - if let Some(session) = self.repair.as_mut() { - session.requested_from = None; - } self.persist_repaired_prefix(); } @@ -9312,6 +9333,18 @@ } } self.consensus.note_journal_mutation(); + if *ops.start() <= self.recovered_commit { + let lowered = ops.start().saturating_sub(1); + error!( + namespace_raw = self.namespace().inner(), + from_op = *ops.start(), + recovered_commit = self.recovered_commit, + lowered, + "cut partition entries the WAL recovered as committed; lowering the recovered \ + commit point to {lowered} so repair refetches them" + ); + self.recovered_commit = lowered; + } self.pending_consumer_offset_commits .retain(|op, _| !ops.contains(op)); // Nothing is in flight for a cut range, so the next sweep re-requests it
diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs index db27bd8..1c095fa 100644 --- a/core/partitions/src/types.rs +++ b/core/partitions/src/types.rs
@@ -315,7 +315,7 @@ /// First op of the latest request. The next chunk is pulled once the held /// run reaches it, which works while a fenced commit point cannot move. /// - /// `None` after a proof dropped held entries, so the next sweep asks for + /// `None` after any removal of resident entries, so the next sweep asks for /// the cut window from the held run without waiting out /// [`REPAIR_RETRY_TICKS`]. pub requested_from: Option<u64>,
diff --git a/core/server/config.toml b/core/server/config.toml index 6b120ae..b1ebf81 100644 --- a/core/server/config.toml +++ b/core/server/config.toml
@@ -517,6 +517,9 @@ # on a single dropped frame. Paces both the metadata and partition repair loops; # must be nonzero. Paces streams that are already OPEN only: how long a hole # waits before one is opened for it is repair_gap_debounce_interval below. +# Also paces the RequestStartView probe a partition backup sends while its +# entries from before a newer view wait for a header to prove them against; +# those probes have no attempt cap. repair_retry_interval = "1s" # How long a backup must hold committed ops it cannot walk to before the shard's
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index b578298..a0179e0 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs
@@ -4953,9 +4953,10 @@ .consensus() .handle_start_view(PlaneKind::Partitions, &header, suffix_body); let adopted = !actions.is_empty(); - // Consumed on every adoption, so a restart's flag never outlives the first. - let recovered = adopted && partition.take_recovered_unproven(); let transferring = partition.consensus().is_transferring(); + // Consumed only by an adoption that fences, so a transfer that aborts still + // leaves the recovered suffix to the next one. + let recovered = adopted && !transferring && partition.take_recovered_unproven(); let fence_below = adopted && !transferring && (recovered || prior_log_view < partition.consensus().view()); @@ -4980,7 +4981,13 @@ } reconcile_partition_view_divergence(self.id, partition, pending.as_ref()).await; if fence_below { + let truncating = partition.view_truncation_pending(); partition.prove_view_window().await; + // A Break queued a cut, and the pipeline rebuild must read the journal after it. + if !truncating && partition.view_truncation_pending() { + partition.defer_view_adoption(actions, header.replica); + return; + } } if partition.consensus().view_log_unproven() { self.maybe_request_partition_repair(partition, header.replica) @@ -4989,6 +4996,10 @@ } } anchor_view_fence_from_start_view(self.id, partition, &header, suffix_body).await; + if partition.view_truncation_pending() { + partition.defer_view_adoption(actions, header.replica); + return; + } self.finish_partition_view_adoption( IggyNamespace::from_raw(header.group), header.replica, @@ -8438,6 +8449,8 @@ && !partition .consensus() .is_primary_for_view(partition.consensus().view()) + // The answer would re-adopt the view and drop the live repair session. + && !partition.consensus().view_log_unproven() && partition.view_fence_probe_due(self.repair_retry_ticks.get()) { // A probe promises nothing about this replica's view, so it skips the @@ -11112,7 +11125,7 @@ } /// A fenced replica never hands out entries it has not proven: its serve stops at -/// `fenced_floor`, its own commit point. +/// `fenced_floor`, the higher of its `commit_min` and its recovered commit point. fn fenced_serve_ceiling(ceiling: u64, fenced_floor: Option<u64>) -> u64 { fenced_floor.map_or(ceiling, |floor| ceiling.min(floor)) } @@ -11750,15 +11763,17 @@ { return None; } - // Not below the recovered commit point either: the WAL names those ops committed. - let applied_floor = partition + // This cut stays above the recovered commit point, which the WAL names committed. + let committed_floor = partition .fence_floor() .max(partition.log.journal().inner.evicted_through()); let op_head = pending.map_or_else( || partition.consensus().sequencer().current_sequence(), |pending| pending.op_head, ); - let above_head = op_head.max(applied_floor).checked_add(1)?; + // An adopted head never omits a known commit (`handle_start_view`). + debug_assert!(op_head >= partition.fence_floor()); + let above_head = op_head.max(committed_floor).checked_add(1)?; partition .log .journal() @@ -11830,16 +11845,13 @@ } } -/// 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. +/// The lowest header in a view's suffix that can anchor a fence. A blank slot +/// names no identity, so its zero `parent` would link anything below it. 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) + matches!(dvc_header_kind(header), DvcHeaderKind::Valid) && header.anchors_chain() }) .min_by_key(|header| header.op) .copied() @@ -11881,8 +11893,8 @@ } /// Hand a fence still waiting for its anchor the headers a same-view `StartView` -/// carries: the probe answer a fenced replica asked for. A proof that breaks -/// queues its own truncation, as on adoption. +/// carries: the probe answer a fenced replica asked for. A held entry the answer +/// names differently, or a proof that breaks, queues a truncation, as on adoption. #[allow(clippy::future_not_send)] async fn anchor_view_fence_from_start_view<B, SB>( shard: u16, @@ -11905,8 +11917,8 @@ return; } let pending = match MergedLog::from_start_view(header, suffix_body) { - Ok(Some(pending)) => pending, - Ok(None) => return, + Ok(pending) if !pending.headers.is_empty() => pending, + Ok(_) => return, Err(error) => { tracing::warn!( shard, @@ -11918,10 +11930,56 @@ return; } }; + if let Some(op) = first_view_divergence(partition, &pending) { + tracing::warn!( + shard, + namespace_raw = partition.consensus().group(), + op, + view = header.view, + "held partition entry at op {op} disagrees with the view's log, dropping from there" + ); + partition.consensus().fence_view_log(op, op); + if partition.has_io_dispatcher() { + partition.defer_view_truncation(op..=u64::MAX, Vec::new(), Some(header.replica)); + } else if let Err(error) = partition.truncate_uncommitted_from(op).await { + tracing::error!(shard, op, %error, "could not truncate diverging partition entries"); + } else { + reconcile_partition_view_divergence(shard, partition, Some(&pending)).await; + } + } fence_view_window_below(partition, Some(&pending)); partition.prove_view_window().await; } +/// The lowest op above the fence floor where this replica holds an entry the view's +/// log names differently. The fence's chain proof stops below its anchor, so held +/// entries at or above it are compared here. +fn first_view_divergence<B, SB>( + partition: &IggyPartition<B, SB>, + pending: &MergedLog, +) -> Option<u64> +where + B: MessageBus, + SB: journal::superblock::SuperblockStore, +{ + let journal = &partition.log.journal().inner; + let floor = partition.fence_floor(); + let top = pending.op_head.min(journal.last_op()?); + pending + .headers + .iter() + .filter(|canonical| { + canonical.op > floor + && canonical.op <= top + && matches!(dvc_header_kind(canonical), DvcHeaderKind::Valid) + && journal + .header_by_op(canonical.op) + .is_some_and(|local| !header_is_view_entry(&local, canonical)) + }) + .map(|canonical| canonical.op) + .min() +} + /// 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 @@ -12487,14 +12545,16 @@ mod repair_scope_tests { //! Who parked the log decides what it means. - use consensus::{LocalPipeline, Sequencer, VsrConsensus}; - use iggy_binary_protocol::{Command, PrepareHeader, RequestPreparesHeader}; + use consensus::{LocalPipeline, PlaneKind, Sequencer, Status, VsrConsensus}; + use iggy_binary_protocol::{ + Command, ConsensusHeader, PrepareHeader, RequestPreparesHeader, StartViewChangeHeader, + }; use message_bus::IggyMessageBus; use server_common::Message; use super::{ - MergedLog, adopted_suffix_head, pending_covers_suffix, repair_op_in_scope, - repair_serve_ceiling, + MergedLog, adopted_suffix_head, fenced_serve_ceiling, pending_covers_suffix, + repair_op_in_scope, repair_serve_ceiling, }; fn header(op: u64) -> PrepareHeader { @@ -12587,6 +12647,58 @@ } #[test] + fn given_a_fenced_view_change_donor_when_serving_should_stop_at_its_fence_floor() { + const COMMIT_MIN: u64 = 20; + const FENCE_FLOOR: u64 = 30; + const HEAD: u64 = 90; + const NEXT_VIEW: u32 = 1; + let consensus = VsrConsensus::new(1, 0, 3, 0, IggyMessageBus::new(0), LocalPipeline::new()); + consensus.init(); + consensus.restore_commit_state(COMMIT_MIN, 40); + consensus.sequencer().set_sequence(HEAD); + let change = Message::<StartViewChangeHeader>::new(std::mem::size_of::< + StartViewChangeHeader, + >()) + .transmute_header(|_, header: &mut StartViewChangeHeader| { + header.command = Command::StartViewChange; + header.size = u32::try_from(std::mem::size_of::<StartViewChangeHeader>()).unwrap(); + header.cluster = consensus.cluster(); + header.replica = 2; + header.view = NEXT_VIEW; + header.seal(); + }); + consensus.handle_start_view_change(PlaneKind::Partitions, change.header()); + assert_eq!(consensus.status(), Status::ViewChange); + let next_primary = consensus.primary_index(NEXT_VIEW); + let backup = (0..3) + .find(|replica| *replica != next_primary && *replica != consensus.replica()) + .unwrap(); + let ceiling = |requester: u8, fenced_floor: Option<u64>| { + let request = Message::<RequestPreparesHeader>::new(std::mem::size_of::< + RequestPreparesHeader, + >()) + .transmute_header(|_, header: &mut RequestPreparesHeader| { + header.command = Command::RequestPrepares; + header.cluster = consensus.cluster(); + header.size = u32::try_from(std::mem::size_of::<RequestPreparesHeader>()).unwrap(); + header.replica = requester; + header.view = NEXT_VIEW; + header.from_op = 1; + header.to_op = u64::MAX; + }); + fenced_serve_ceiling( + repair_serve_ceiling(request.header(), &consensus), + fenced_floor, + ) + }; + assert_eq!(ceiling(next_primary, None), HEAD); + // The fence floor counts the recovered commit point, above `commit_min`. + assert_eq!(ceiling(next_primary, Some(FENCE_FLOOR)), FENCE_FLOOR); + assert_eq!(ceiling(backup, None), COMMIT_MIN); + assert_eq!(ceiling(backup, Some(FENCE_FLOOR)), COMMIT_MIN); + } + + #[test] fn given_an_adopted_suffix_when_live_head_advances_should_preserve_its_repair_boundary() { let pending = parked(); assert_eq!(adopted_suffix_head(&pending, 98, 100), Some(100));
diff --git a/core/shard/src/partition_io.rs b/core/shard/src/partition_io.rs index ca06aac..3e46528 100644 --- a/core/shard/src/partition_io.rs +++ b/core/shard/src/partition_io.rs
@@ -5028,64 +5028,180 @@ #[compio::test] async fn given_a_blank_start_view_head_when_repair_arrives_should_preserve_and_prove_history() { - const HEAD: u64 = 1; - const NEW_VIEW: u32 = 3; + const HEAD: u64 = 3; for held in [false, true] { - let bus = Rc::new(IggyMessageBus::new(0)); - let (owner, _sender) = test_owner(&bus, None); - let namespace = IggyNamespace::new(0, 0, 0); - let prepares = primary_prepares(&bus, namespace, HEAD).await; - let prepare = prepares.into_iter().next().unwrap(); - let head_header = *prepare.header(); - let mut partition = - repairing_replica(&bus, namespace, (REPAIR_BACKUP, 3), REPAIR_PRIMARY, 0, 0); - partition.repair = None; - if held { + for canonical in [true, false] { + let (bus, probes) = probe_capturing_bus(); + let (owner, _sender) = test_owner(&bus, None); + owner.set_repair_retry_ticks(2); + let namespace = IggyNamespace::new(0, 0, 0); + let prepares = + Box::pin(blank_head_backup(&owner, &bus, namespace, HEAD, held)).await; + let partition = owner.plane.partitions().get_mut_by_ns(&namespace).unwrap(); + assert!(!partition.consensus().view_log_unproven()); + assert!((1..=HEAD).all(|op| partition.log.journal().inner.holds_op(op))); + // A repaired body is no anchor for the view fence. + assert!(partition.view_fence_unanchored()); + partition + .commit_journal(owner.plane.partitions().config()) + .await; + assert_eq!(partition.consensus().commit_min(), 0); + + for _ in 0..8 { + assert!(owner.tick_partitions(&mut Vec::new()).await.is_none()); + if !probes.borrow().is_empty() { + break; + } + } + assert!( + !probes.borrow().is_empty(), + "an unanchored fence probes the view's primary" + ); + let head = *prepares.last().unwrap().header(); + let answer = if canonical { head } else { forked(head) }; + owner + .on_start_view(start_view_suffix( + namespace, + BLANK_HEAD_VIEW, + HEAD, + HEAD, + &[answer], + )) + .await; + let partition = owner.plane.partitions().get_mut_by_ns(&namespace).unwrap(); + partition + .commit_journal(owner.plane.partitions().config()) + .await; + if canonical { + assert!(!partition.view_fenced()); + assert_eq!(partition.consensus().commit_min(), HEAD); + } else { + assert!( + partition.consensus().commit_min() < HEAD, + "held={held}: a head the view names differently must not commit" + ); + } + } + } + } + + #[compio::test] + async fn given_a_probe_answer_naming_a_held_entry_differently_when_proving_should_not_commit_it() + { + const HEAD: u64 = 4; + let (bus, _probes) = probe_capturing_bus(); + let (owner, _sender) = test_owner(&bus, None); + let namespace = IggyNamespace::new(0, 0, 0); + let prepares = Box::pin(blank_head_backup(&owner, &bus, namespace, HEAD, true)).await; + let headers: Vec<_> = prepares.iter().map(|prepare| *prepare.header()).collect(); + // The lowest sealed header anchors the fence below the diverging head. + let answer = [forked(headers[3]), headers[2], headers[1]]; + owner + .on_start_view(start_view_suffix( + namespace, + BLANK_HEAD_VIEW, + HEAD, + HEAD, + &answer, + )) + .await; + let partition = owner.plane.partitions().get_mut_by_ns(&namespace).unwrap(); + partition + .commit_journal(owner.plane.partitions().config()) + .await; + assert!(partition.consensus().commit_min() < HEAD); + assert!( + partition.view_truncation_pending() || !partition.log.journal().inner.holds_op(HEAD) + ); + } + + const BLANK_HEAD_VIEW: u32 = 3; + + /// A bus that records the `RequestStartView` probes sent to the view's primary. + fn probe_capturing_bus() -> (Rc<IggyMessageBus>, Rc<RefCell<Vec<RequestStartViewHeader>>>) { + let bus = Rc::new(IggyMessageBus::new(0)); + let probes = Rc::new(RefCell::new(Vec::new())); + let captured = Rc::clone(&probes); + bus.set_replica_forward_fn(Box::new(move |_, _, frame| { + let message = + Message::<GenericHeader>::try_from(Owned::copy_from_slice(frame.as_slice())) + .unwrap(); + if message.header().command == Command::RequestStartView { + let probe = Message::<RequestStartViewHeader>::try_from(Owned::copy_from_slice( + frame.as_slice(), + )) + .unwrap(); + captured.borrow_mut().push(*probe.header()); + } + Ok(()) + })); + let primary = u8::try_from(BLANK_HEAD_VIEW % 3).unwrap(); + assert!(bus.owner_table().try_claim(primary, 1)); + (bus, probes) + } + + /// A backup that adopted [`BLANK_HEAD_VIEW`] from a `StartView` with a blank head + /// and then received the history up to `head` through repair, so the view's log + /// proof settles on the repaired head body. With `journaled`, it held that + /// history before the adoption. + #[allow(clippy::future_not_send)] + async fn blank_head_backup( + owner: &IoTestShard, + bus: &Rc<IggyMessageBus>, + namespace: IggyNamespace, + head: u64, + journaled: bool, + ) -> Vec<Message<PrepareHeader>> { + let prepares = primary_prepares(bus, namespace, head).await; + let mut partition = + repairing_replica(bus, namespace, (REPAIR_BACKUP, 3), REPAIR_PRIMARY, 0, 0); + partition.repair = None; + if journaled { + for prepare in &prepares { partition.on_replicate(prepare.clone()).await; } - owner.plane.partitions().remove(&namespace).unwrap(); - owner.plane.partitions().insert(namespace, partition); - owner - .on_start_view(start_view_suffix( - namespace, - NEW_VIEW, - HEAD, - HEAD, - &[consensus::dvc_blank(HEAD)], - )) - .await; - let partition = owner.plane.partitions().get_by_ns(&namespace).unwrap(); - assert!(partition.consensus().view_log_unproven()); - assert!(!partition.view_truncation_pending()); - assert_eq!(partition.log.journal().inner.holds_op(HEAD), held); - let repaired = prepare.transmute_header( - |header, repaired: &mut iggy_binary_protocol::RepairPrepareHeader| { - repaired.0 = header; - repaired.0.command = Command::RepairPrepare; - }, - ); - owner.on_repair_prepare(repaired).await; - let partition = owner.plane.partitions().get_by_ns(&namespace).unwrap(); - assert!(!partition.consensus().view_log_unproven()); - // A repaired body is no anchor for the view fence: it waits for the - // primary's answer to its probe, which now names a sealed head. - assert!(partition.view_fence_unanchored()); - owner - .on_start_view(start_view_suffix( - namespace, - NEW_VIEW, - HEAD, - HEAD, - &[head_header], - )) - .await; - let partition = owner.plane.partitions().get_mut_by_ns(&namespace).unwrap(); - assert!(!partition.view_fenced()); - partition - .commit_journal(owner.plane.partitions().config()) - .await; - assert_eq!(partition.consensus().commit_min(), HEAD); } + owner.plane.partitions().remove(&namespace).unwrap(); + owner.plane.partitions().insert(namespace, partition); + owner + .on_start_view(start_view_suffix( + namespace, + BLANK_HEAD_VIEW, + head, + head, + &[consensus::dvc_blank(head)], + )) + .await; + let partition = owner.plane.partitions().get_by_ns(&namespace).unwrap(); + assert!(partition.consensus().view_log_unproven()); + assert!(!partition.view_truncation_pending()); + assert_eq!(partition.log.journal().inner.holds_op(head), journaled); + let repaired: Vec<_> = if journaled { + prepares.last().into_iter().cloned().collect() + } else { + prepares.clone() + }; + for prepare in repaired.into_iter().rev() { + owner + .on_repair_prepare(prepare.transmute_header( + |header, repaired: &mut iggy_binary_protocol::RepairPrepareHeader| { + repaired.0 = header; + repaired.0.command = Command::RepairPrepare; + }, + )) + .await; + } + prepares + } + + /// A sealed header at the same op and on the same parent as `header`, naming a + /// different entry. + fn forked(header: PrepareHeader) -> PrepareHeader { + let mut fork = header; + fork.checksum_body ^= 1; + fork.checksum = fork.identity_checksum(); + assert_ne!(fork.checksum, header.checksum); + fork } fn start_view_suffix(
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 51a1af2..fe1a9b1 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs
@@ -8968,7 +8968,7 @@ } /// `(status, view, commit_min, commit_max)` of one replica's partition group. - fn group_state( + pub fn group_state( sim: &Simulator, replica: u8, namespace: IggyNamespace,
diff --git a/core/simulator/src/stale_prepare_commit_tests.rs b/core/simulator/src/stale_prepare_commit_tests.rs index e5faff8..3e92fb5 100644 --- a/core/simulator/src/stale_prepare_commit_tests.rs +++ b/core/simulator/src/stale_prepare_commit_tests.rs
@@ -33,10 +33,10 @@ use crate::Simulator; use crate::client::SimClient; -use crate::packet::{PacketSimulatorOptions, ProcessId}; +use crate::packet::ProcessId; +use crate::partition_repair_driver_tests::{cluster, group_state}; 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. @@ -45,39 +45,6 @@ /// 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());