This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch state_transfer_view_change in repository https://gitbox.apache.org/repos/asf/iggy.git
commit e89d2950c710b06f49f3bde9864ff4814f27fcca Author: Grzegorz Koszyk <[email protected]> AuthorDate: Fri Oct 9 15:07:18 2026 +0200 fix(partitions): finish state transfer for a replica whose view ran ahead --- core/consensus/src/impls.rs | 144 ++++++ core/partitions/src/iggy_partition.rs | 785 +++++++++++++++++++++++++----- core/partitions/src/journal.rs | 34 ++ core/partitions/src/types.rs | 11 +- core/server/src/partition_reconciler.rs | 5 +- core/shard/src/lib.rs | 210 ++++++-- core/shard/src/partition_io.rs | 4 +- core/shard/src/poll/completion_tests.rs | 829 +++++++++++++++++++++++++++++++- 8 files changed, 1846 insertions(+), 176 deletions(-) diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 61d4d6dbc..1f821312c 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -1264,6 +1264,13 @@ where /// strict `>`. observed_newer_view: Cell<u32>, + /// `(view, replica)` of the highest-view sender heard acting as that view's + /// primary: a partition `Commit` heartbeat, which only a `Normal` primary + /// sends, or a state-transfer offer, which only a primary serves. A replica + /// that cannot rejoin keeps electing alone, so its own `view` names a + /// primary nobody follows. + observed_primary: Cell<Option<(u32, u8)>>, + /// Highest op number that has been locally executed (state machine applied, /// client table updated). Advances one-by-one in `commit_journal` (backup) /// and `on_ack` (primary). On a normal primary, `commit_min == commit_max`. @@ -1677,6 +1684,7 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { status: Cell::new(Status::Recovering), state_transfer_stage: Cell::new(StateTransferStage::Idle), observed_newer_view: Cell::new(0), + observed_primary: Cell::new(None), sequencer: LocalSequencer::new(0), commit_min: Cell::new(0), commit_max: Cell::new(0), @@ -3399,6 +3407,53 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { } } + /// Record a frame from `replica` stamped with `view`, if `replica` is that + /// view's primary. Monotone in `view`. + pub fn observe_primary(&self, view: u32, replica: u8) { + if replica != self.primary_index(view) { + return; + } + if self + .observed_primary + .get() + .is_none_or(|(observed_view, _)| view > observed_view) + { + self.observed_primary.set(Some((view, replica))); + } + } + + /// Drop the observed primary if it is `replica`, after a request to it + /// failed. A dead primary sends nothing newer, so a kept hint would steer + /// every later rotation back to it. + pub fn forget_primary(&self, replica: u8) { + if self + .observed_primary + .get() + .is_some_and(|(_, observed)| observed == replica) + { + self.observed_primary.set(None); + } + } + + /// The replica most likely to be the group's primary, or `None` when this + /// replica cannot tell. + /// + /// `primary_index(view)` only while `view` is the view the log was adopted + /// in: past it, `view` may have climbed through elections nobody joined. + /// An observed primary counts only from `log_view` on, because below it the + /// group has provably moved on. + #[must_use] + pub fn primary_hint(&self) -> Option<u8> { + let log_view = self.log_view.get(); + if let Some((view, replica)) = self.observed_primary.get() + && view >= log_view + { + return Some(replica); + } + let view = self.view.get(); + (view == log_view).then(|| self.primary_index(view)) + } + #[must_use] pub const fn state_transfer_stage(&self) -> StateTransferStage { self.state_transfer_stage.get() @@ -6553,3 +6608,92 @@ mod probe_answer_tests { ); } } + +#[cfg(test)] +mod primary_hint_tests { + //! Which replica a partition peer choice treats as the group's primary. + + use super::*; + use crate::LocalPipeline; + use crate::test_bus::NoopBus; + + /// Replica 0 of five whose log was adopted in `log_view` and whose own + /// view has since moved to `view`. + fn replica(view: u32, log_view: u32) -> VsrConsensus<NoopBus> { + let mut consensus = VsrConsensus::new(1, 0, 5, 7, NoopBus, LocalPipeline::new()); + consensus.init(); + consensus.set_view(view); + consensus.set_log_view(log_view); + consensus + } + + #[test] + fn given_a_view_matching_log_view_when_nothing_was_observed_should_name_its_primary() { + assert_eq!(replica(12, 12).primary_hint(), Some(2)); + } + + #[test] + fn given_a_view_ratcheted_past_log_view_when_nothing_was_observed_should_have_no_hint() { + assert_eq!(replica(58, 11).primary_hint(), None); + } + + #[test] + fn given_a_view_ratcheted_past_log_view_when_the_group_primary_was_heard_should_name_it() { + let consensus = replica(58, 11); + consensus.observe_primary(21, 1); + assert_eq!(consensus.primary_hint(), Some(1)); + } + + #[test] + fn given_a_view_ratcheted_past_log_view_when_the_primary_of_log_view_was_heard_should_name_it() + { + let consensus = replica(58, 11); + let primary = consensus.primary_index(11); + consensus.observe_primary(11, primary); + assert_eq!(consensus.primary_hint(), Some(primary)); + } + + #[test] + fn given_an_observed_primary_when_it_is_forgotten_should_fall_back_to_no_hint() { + let consensus = replica(58, 11); + consensus.observe_primary(21, 1); + consensus.forget_primary(1); + assert_eq!(consensus.primary_hint(), None); + consensus.observe_primary(21, 1); + assert_eq!( + consensus.primary_hint(), + Some(1), + "a forgotten primary that speaks again is heard again" + ); + } + + #[test] + fn given_an_observed_primary_when_another_replica_is_forgotten_should_keep_it() { + let consensus = replica(58, 11); + consensus.observe_primary(21, 1); + consensus.forget_primary(2); + assert_eq!(consensus.primary_hint(), Some(1)); + } + + #[test] + fn given_a_sender_that_is_not_its_view_primary_when_observed_should_ignore_it() { + let consensus = replica(58, 11); + consensus.observe_primary(21, 2); + assert_eq!(consensus.primary_hint(), None); + } + + #[test] + fn given_a_primary_below_log_view_when_observed_should_ignore_it() { + let consensus = replica(58, 11); + consensus.observe_primary(8, 3); + assert_eq!(consensus.primary_hint(), None); + } + + #[test] + fn given_an_older_view_primary_when_observed_after_a_newer_one_should_keep_the_newer() { + let consensus = replica(58, 11); + consensus.observe_primary(21, 1); + consensus.observe_primary(12, 2); + assert_eq!(consensus.primary_hint(), Some(1)); + } +} diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 86e752fb3..e04caf728 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -11323,22 +11323,7 @@ where return; } let applied = if header.operation == Operation::SendMessages { - match self.append_repaired_send_messages(message).await { - Ok(base_offset) => { - if header.op <= session.commit_to_op - && let (Some(base_offset), Some(session)) = - (base_offset, self.repair.as_mut()) - { - session.first_batch_offset = Some( - session - .first_batch_offset - .map_or(base_offset, |first| first.min(base_offset)), - ); - } - Ok(()) - } - Err(error) => Err(error), - } + self.append_repaired_send_messages(message).await } else { self.apply_replicated_operation(message).await.map(|_| ()) }; @@ -11425,57 +11410,44 @@ where .journal() .inner .repaired_window_shape(floor, session.commit_to_op); - let connected = match (session.first_batch_offset, stand_in) { - (Some(first), Some(bound)) => first <= bound, - (Some(first), None) => first == 0, - // No repaired batch arrived, so there is no offset anchor to - // verify the floor's continuum claim against. `None` is only - // safe when the served window itself proves it carried no - // messages: every op in `(floor, to_op]` journaled and none - // of them `SendMessages`. Anything less -- dropped frames, or - // a fully evicted window -- is indistinguishable from a - // message range below the floor that this replica does not - // durably own, and accepting it would serve a holed log. + let anchor = self.repair_window_anchor(floor, session.commit_to_op); + let connected = match (anchor, stand_in) { + (Some(anchor), Some(bound)) => anchor.base_offset <= bound, + (Some(anchor), None) => anchor.base_offset == 0, + // No batch above the floor to anchor the floor's continuum + // claim. `None` is only safe when the served window itself + // proves it carried no messages: every op in `(floor, to_op]` + // journaled and none of them `SendMessages`. Anything less -- + // dropped frames, or a fully evicted window -- is + // indistinguishable from a message range below the floor that + // this replica does not durably own, and accepting it would + // serve a holed log. Sends at or below the local purge fence + // still count in `holds_messages`, so a fence at or above + // `commit_to_op` refuses here for one extra transfer: the + // fence is local, and uncounted they would let a window whose + // messages the primary still holds pass as message-free. (None, _) => { floor < session.commit_to_op && committed_shape.complete && !committed_shape.holds_messages } }; - if !connected { - tracing::error!( - target: "iggy.partitions.diag", - plane = "partitions", - namespace_raw = self.namespace().inner(), - floor, - first_batch_offset = ?session.first_batch_offset, - recovered_durable_offset = ?durable_end, - "refusing commit floor: repaired window does not connect \ - to recovered durable state (needs state transfer)" - ); - self.commit_journal(config).await; - // A refusal is DEFINITIVE only once the window itself is - // fully present (or provably empty): until then more frames - // can still lower `first_batch_offset` into connection, so - // the session stays armed and the stall retry re-requests. - // A complete window that still cannot connect will never - // improve -- the peer retains nothing below the floor and - // this replica holds nothing either -- and an EMPTY window - // (everything evicted) re-raises identically every round. - // Both are the state-transfer trigger; the session is - // dropped here so the caller's arming funnel starts clean, - // and a transfer-unavailable fallback re-arms repair fresh. - if committed_shape.complete { - self.repair = None; - return RepairConclusion::FloorRefused { - floor, - to_op: session.commit_to_op, - }; + if connected { + if floor > self.consensus().commit_min() { + self.consensus().set_commit_floor(floor); } - return RepairConclusion::InProgress; - } - if floor > self.consensus().commit_min() { - self.consensus().set_commit_floor(floor); + } else if let Some(conclusion) = self + .refuse_disconnected_floor( + config, + session, + floor, + durable_end, + anchor, + committed_shape.complete, + ) + .await + { + return conclusion; } } if let Some(conclusion) = self.repair_persistence_pending(session) { @@ -11530,6 +11502,158 @@ where } } + /// Refuse a floor whose window does not connect to recovered durable state, + /// or `None` when the walk below makes the floor moot and completion + /// proceeds as without one. + async fn refuse_disconnected_floor( + &mut self, + config: &PartitionsConfig, + session: RepairSession, + floor: u64, + durable_end: Option<u64>, + anchor: Option<RepairAnchor>, + window_complete: bool, + ) -> Option<RepairConclusion> { + self.commit_journal(config).await; + // Ops this replica already held can carry the walk past the floor. + // Under lazy flush their headers stay resident, so the leading edge + // below would read as resident and refuse a floor that no longer + // moves `commit_min`, spending a state transfer on nothing. Judged by + // the peer's unclamped floor: one clamped down to `commit_max` still + // names ops the peer evicted. + if session + .floor + .is_some_and(|raw_floor| self.consensus().commit_min() >= raw_floor) + { + return None; + } + let leading_edge_resident = self.repair_leading_edge_resident(floor, anchor); + // Definitive once the window, or its leading edge `(floor, anchor.op)`, + // is resident: until then a frame still to come can land below the + // anchor and connect. The leading edge suffices because the primary + // assigns offsets in op order. A purge this replica's metadata has not + // committed yet can still skew the connect check; the committed-purge + // lag is pinned by `repair_completion_defers_until_committed_purge_applies`. + if window_complete || leading_edge_resident { + let reason = if window_complete { + "whole_window" + } else { + "leading_edge" + }; + tracing::error!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace().inner(), + floor, + commit_to_op = session.commit_to_op, + anchor_op = ?anchor.map(|anchor| anchor.op), + anchor_base_offset = ?anchor.map(|anchor| anchor.base_offset), + reason, + recovered_durable_offset = ?durable_end, + "refusing commit floor: repaired window does not connect \ + to recovered durable state (needs state transfer)" + ); + self.repair = None; + return Some(RepairConclusion::FloorRefused { + floor, + to_op: session.commit_to_op, + }); + } + tracing::warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace().inner(), + floor, + commit_to_op = session.commit_to_op, + anchor_op = ?anchor.map(|anchor| anchor.op), + anchor_base_offset = ?anchor.map(|anchor| anchor.base_offset), + recovered_durable_offset = ?durable_end, + "commit floor does not connect yet; repairing below the anchor" + ); + Some(RepairConclusion::InProgress) + } + + /// Where the next pull of the repair window starts: past the resident + /// prefix above a floor that still holds the walk, else `commit_min + 1`. + /// `None` without a session. + /// + /// A held walk never moves `commit_min`, so pulling from it re-serves the + /// same first chunk forever while the rest of the window stays absent. + #[must_use] + pub fn repair_pull_from(&self) -> Option<u64> { + let session = self.repair.as_ref()?; + let commit_min = self.consensus().commit_min(); + let Some(floor) = session.floor.filter(|&floor| floor > commit_min) else { + return Some(commit_min.saturating_add(1)); + }; + let journal = &self.log.journal().inner; + let mut op = floor.saturating_add(1); + while op <= session.fetch_to_op && journal.holds_op(op) { + op += 1; + } + Some(op) + } + + /// The next chunk to pull right after a `RepairDone` while a floor holds + /// the walk, or `None` when the resident prefix has not grown since the + /// last such pull or reaches past `fetch_to_op`. + pub fn next_floor_pull(&mut self) -> Option<u64> { + let commit_min = self.consensus().commit_min(); + let from_op = self.repair_pull_from()?; + let session = self.repair.as_mut()?; + let floor_pending = session.floor.is_some_and(|floor| floor > commit_min); + if !floor_pending || from_op > session.fetch_to_op || from_op <= session.floor_pulled_from { + return None; + } + session.floor_pulled_from = from_op; + Some(from_op) + } + + /// The lowest resident `SendMessages` batch in `(floor, commit_to_op]`: + /// where the repair window begins in offset space. + /// + /// Read from the journal rather than tracked per session. A batch at or + /// below the floor is not what the floor claims recovered state stands in + /// for, a batch resident from an earlier session or the boot WAL counts + /// as much as one repaired now, and purged bytes at or below the purge + /// fence cannot stand in for durable state. + fn repair_window_anchor(&self, floor: u64, commit_to_op: u64) -> Option<RepairAnchor> { + let from = floor.max(self.purge_floor_op).checked_add(1)?; + let (op, entry) = self + .log + .journal() + .inner + .lowest_resident_send_in(from..=commit_to_op)?; + let batch = decode_prepare_slice(entry.as_slice()) + .inspect_err(|error| { + tracing::warn!( + target: "iggy.partitions.diag", + plane = "partitions", + namespace_raw = self.namespace().inner(), + op, + %error, + "resident repair batch does not decode; the window has no anchor" + ); + }) + .ok()?; + Some(RepairAnchor { + op, + base_offset: batch.header.base_offset, + }) + } + + /// Whether every op in `(floor, anchor.op)` is resident, so no frame still + /// to come can land below the anchor. + fn repair_leading_edge_resident(&self, floor: u64, anchor: Option<RepairAnchor>) -> bool { + anchor.is_some_and(|anchor| { + self.log + .journal() + .inner + .repaired_window_shape(floor, anchor.op.saturating_sub(1)) + .complete + }) + } + fn repair_persistence_pending(&mut self, session: RepairSession) -> Option<RepairConclusion> { if let Some(persistence) = &self.persistence { self.persist_repaired_prefix(); @@ -11562,7 +11686,7 @@ where async fn append_repaired_send_messages( &mut self, message: Message<PrepareHeader>, - ) -> Result<Option<u64>, IggyError> { + ) -> Result<(), IggyError> { let write_lock = self.write_lock.clone(); let _guard = write_lock.lock().await; @@ -11584,12 +11708,10 @@ where // repaired pre-purge batch is still journaled -- the commit walk stops // at the first missing op, so dropping it would wedge `commit_min` -- // but it must not re-advance the reset counters or re-count purged - // bytes. `None` also keeps it out of the session's - // `first_batch_offset`: that anchors the floor-connect check, and - // purged bytes cannot stand in for durable state. + // bytes. if op <= self.purge_floor_op { self.journal_append(message.into_frozen()).await?; - return Ok(None); + return Ok(()); } let last_offset = base_offset @@ -11630,7 +11752,7 @@ where .store(dirty.max(last_offset), Ordering::Relaxed); self.log.segments_mut()[segment_index].current_position = next_position; self.log.journal_mut().info = journal_info; - Ok(Some(base_offset)) + Ok(()) } async fn send_prepare_ok(&self, header: &PrepareHeader) -> bool { @@ -11675,6 +11797,13 @@ where } } +/// The batch a repair window's floor-connect check is judged by. +#[derive(Debug, Clone, Copy)] +struct RepairAnchor { + op: u64, + base_offset: u64, +} + #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] enum DeferredViewPhase { StartViewChange, @@ -12630,6 +12759,82 @@ mod tests { (directory, receiver) } + /// A replica that needs state transfer while producers keep writing. The + /// serving primary builds its offer at its `commit_min`. The pull takes + /// seconds for a large active segment, and meanwhile the primary's commit + /// heartbeats raise the receiver's `commit_max` past the offer. The + /// receiver's journal stopped at the gap, so its sequencer is below the + /// offer and the install erases nothing it journaled. The committed tail + /// above the offer is what journal repair fetches after the install. + /// + /// Pins `install_state_transfer`, not the repair floor check. Refusing + /// such an offer is a livelock: under sustained load every offer + /// is below `commit_max` by the time it lands, every round is refused with + /// "transfer frontier X is below the local commit frontier Y", and the + /// replica keeps serving reads from its stale prefix. + #[compio::test] + async fn given_commit_max_advanced_during_the_pull_when_installing_should_land_and_leave_the_tail_to_repair() + { + let origin_directory = tempfile::tempdir().unwrap(); + let receiver_directory = tempfile::tempdir().unwrap(); + let (mut origin, _) = recording_partition_at(0, 3); + origin.set_partition_dir(origin_directory.path().to_string_lossy().into_owned()); + let committed = checksummed_segment_prepare(1, 0, 0, b"committed"); + origin + .log + .journal() + .inner + .append(committed.clone().into_frozen()) + .await + .unwrap(); + origin.consensus().sequencer().set_sequence(1); + origin + .consensus() + .set_last_prepare_checksum(committed.header().checksum); + origin.consensus().advance_commit_max(1); + origin.consensus().advance_commit_min(1); + origin + .dedup_mut() + .commit_capacity(usize::try_from(committed.header().retry_capacity).unwrap()) + .unwrap(); + origin.offset_space.committed_seeded = true; + origin.offset.store(0, Ordering::Relaxed); + origin.log.journal().inner.evict_prefix(1); + let offer = origin.state_transfer_offer(&repair_config()).await.unwrap(); + assert_eq!(offer.commit_op, 1); + + let (mut receiver, _) = recording_partition_at(1, 3); + receiver.set_partition_dir(receiver_directory.path().to_string_lossy().into_owned()); + set_offset_dirs_under(&mut receiver, receiver_directory.path()); + // Learned from the primary's commit heartbeats while the pull ran. The + // journal never got past the gap, so nothing above the offer is + // journaled here. + receiver.consensus().advance_commit_max(4); + assert_eq!(receiver.consensus().sequencer().current_sequence(), 0); + + let installed = receiver + .install_state_transfer( + &repair_config(), + offer.commit_op, + Vec::new(), + &offer.offsets.1, + 0, + ) + .await; + + assert!( + installed.is_ok(), + "an offer that erases no journaled op must install, got {installed:?}" + ); + assert_eq!(receiver.consensus().commit_min(), 1); + assert_eq!(receiver.consensus().sequencer().current_sequence(), 1); + assert_eq!( + receiver.consensus().commit_max(), + 4, + "the committed tail above the offer stays known for journal repair to fetch" + ); + } + #[compio::test] async fn transfer_establishes_wal_body_ownership_without_losing_the_active_index_writer() { for durability in [ @@ -15041,7 +15246,7 @@ mod tests { .consensus() .handle_start_view(PlaneKind::Partitions, start.header(), &[]); assert!(!partition.consensus().view_log_is_pending()); - partition.repair = Some(armed_fetch_session(0, 3, 0, None)); + partition.repair = Some(armed_fetch_session(0, 3, 0)); partition.apply_repaired_prepare(first.clone()).await; assert!( !partition.log.journal().inner.holds_op(1), @@ -23348,29 +23553,28 @@ mod tests { } } - pub(super) fn armed_session( - to_op: u64, - floor: u64, - first_batch_offset: Option<u64>, - ) -> RepairSession { - armed_fetch_session(to_op, to_op, floor, first_batch_offset) + pub(super) fn armed_session(to_op: u64, floor: u64) -> RepairSession { + armed_fetch_session(to_op, to_op, floor) } - fn armed_fetch_session( - to_op: u64, - fetch_to_op: u64, - floor: u64, - first_batch_offset: Option<u64>, - ) -> RepairSession { + fn armed_fetch_session(to_op: u64, fetch_to_op: u64, floor: u64) -> RepairSession { + RepairSession { + floor: Some(floor), + ..floorless_session(to_op, fetch_to_op) + } + } + + /// A session at view 0 against peer 0 that has not learned a floor yet. + fn floorless_session(commit_to_op: u64, fetch_to_op: u64) -> RepairSession { RepairSession { nonce: 1, view: 0, - commit_to_op: to_op, + commit_to_op, fetch_to_op, - floor: Some(floor), + floor: None, peer: 0, - first_batch_offset, idle_ticks: 0, + floor_pulled_from: 0, } } @@ -23523,9 +23727,19 @@ mod tests { op: u64, parent: u128, checksum: u128, + ) -> Message<PrepareHeader> { + repaired_send_prepare_at(op, op, parent, checksum) + } + + /// [`repaired_send_prepare`] with its batch stamped at `base_offset`. + fn repaired_send_prepare_at( + op: u64, + base_offset: u64, + parent: u128, + checksum: u128, ) -> Message<PrepareHeader> { let namespace = IggyNamespace::new(1, 1, 0); - let record = build_segment_record(namespace, op); + let record = build_segment_record(namespace, base_offset); let header_size = std::mem::size_of::<PrepareHeader>(); let total = header_size + record.len(); let mut message = Message::<PrepareHeader>::new(total); @@ -23550,7 +23764,7 @@ mod tests { { const IDLE_TICKS: u32 = 7; let mut partition = test_partition(); - partition.repair = Some(armed_session(3, 0, None)); + partition.repair = Some(armed_session(3, 0)); partition.repair.as_mut().unwrap().idle_ticks = IDLE_TICKS; partition.repair_attempts = 2; partition.transition = Some(PendingPartitionTransition::View { drain: None }); @@ -23595,7 +23809,7 @@ mod tests { const CHECKSUM_2: u128 = 0x22; const CHECKSUM_3: u128 = 0x33; let mut partition = test_partition(); - partition.repair = Some(armed_session(3, 0, None)); + partition.repair = Some(armed_session(3, 0)); partition .apply_repaired_prepare(repaired_send_prepare(1, 0, CHECKSUM_1)) @@ -23637,7 +23851,7 @@ mod tests { const CHECKSUM_2: u128 = 0x22; const CHECKSUM_3: u128 = 0x33; let mut partition = test_partition(); - partition.repair = Some(armed_session(3, 0, None)); + partition.repair = Some(armed_session(3, 0)); partition .apply_repaired_prepare(repaired_send_prepare(1, 0, CHECKSUM_1)) @@ -23663,7 +23877,7 @@ mod tests { #[compio::test] async fn given_prior_view_repair_when_a_new_view_started_should_discard_it() { let mut partition = test_partition(); - partition.repair = Some(armed_fetch_session(0, 1, 0, None)); + partition.repair = Some(armed_fetch_session(0, 1, 0)); partition.consensus.set_view(1); partition @@ -23724,7 +23938,7 @@ mod tests { async fn given_no_repaired_batch_when_window_never_arrived_should_refuse_commit_floor() { let mut partition = test_partition(); partition.consensus().advance_commit_max(8); - partition.repair = Some(armed_session(8, 5, None)); + partition.repair = Some(armed_session(8, 5)); let conclusion = partition.complete_repair(&repair_config()).await; @@ -23750,7 +23964,7 @@ mod tests { for op in 6..=8 { journal_prepare(&partition, op, Operation::CreateStream).await; } - partition.repair = Some(armed_session(8, 5, None)); + partition.repair = Some(armed_session(8, 5)); let conclusion = partition.complete_repair(&repair_config()).await; @@ -23766,7 +23980,7 @@ mod tests { for op in 7..=8 { journal_prepare(&partition, op, Operation::CreateStream).await; } - partition.repair = Some(armed_session(8, 5, None)); + partition.repair = Some(armed_session(8, 5)); let conclusion = partition.complete_repair(&repair_config()).await; @@ -23786,7 +24000,7 @@ mod tests { async fn given_no_repaired_batch_when_window_fully_evicted_should_refuse_commit_floor() { let mut partition = test_partition(); partition.consensus().advance_commit_max(8); - partition.repair = Some(armed_session(8, 8, None)); + partition.repair = Some(armed_session(8, 8)); let conclusion = partition.complete_repair(&repair_config()).await; @@ -23805,7 +24019,7 @@ mod tests { async fn given_empty_committed_window_with_a_suffix_fetch_should_escape_to_state_transfer() { let mut partition = test_partition(); partition.consensus().advance_commit_max(5); - partition.repair = Some(armed_fetch_session(5, 9, 5, None)); + partition.repair = Some(armed_fetch_session(5, 9, 5)); let conclusion = partition.complete_repair(&repair_config()).await; @@ -23898,7 +24112,7 @@ mod tests { journal_send_batch(&mut partition, op).await; } partition.consensus().advance_commit_max(3); - partition.repair = Some(armed_fetch_session(2, 3, 0, Some(0))); + partition.repair = Some(armed_fetch_session(2, 3, 0)); partition.commit_journal(&repair_config()).await; let _ = partition.log.journal().inner.evict_prefix(3); @@ -23915,7 +24129,7 @@ mod tests { #[compio::test] async fn given_suffix_fetch_when_its_view_is_discarded_should_clear_the_session() { let mut partition = test_partition(); - partition.repair = Some(armed_fetch_session(0, 3, 0, None)); + partition.repair = Some(armed_fetch_session(0, 3, 0)); partition.consensus.set_view(1); let conclusion = partition.complete_repair(&repair_config()).await; @@ -23933,22 +24147,379 @@ mod tests { let mut partition = test_partition(); partition.consensus().advance_commit_max(8); // No recovered segments (durable end None) and the served window's - // first batch starts at offset 3: ops below the floor are neither + // first batch starts at offset 8: ops below the floor are neither // locally durable nor repaired. - partition.repair = Some(armed_session(8, 5, Some(3))); + partition.repair = Some(armed_session(8, 5)); + partition + .apply_repaired_prepare(repaired_send_prepare(8, 0, 0x88)) + .await; let conclusion = partition.complete_repair(&repair_config()).await; assert_eq!( conclusion, RepairConclusion::InProgress, - "with the window incomplete, later frames can still lower the \ - first batch offset into connection" + "with the window incomplete, later frames can still land below \ + the anchor and connect" + ); + assert_eq!(partition.consensus().commit_min(), 0); + assert!(partition.repair.is_some()); + } + + /// The repair window `(floor, to_op]` is longer than one repair chunk and + /// only its first chunk arrived. The anchor (op 7, offset 7) starts above + /// the recovered durable end (offset 3) and every op between the floor and + /// it is resident, so the refusal is final and must hand recovery to state + /// transfer. + #[compio::test] + async fn given_a_resident_leading_edge_above_durable_end_when_the_window_is_longer_than_a_chunk_should_refuse_commit_floor() + { + let mut partition = test_partition(); + partition.consensus().advance_commit_max(300); + partition.recovered_durable_offset = Some(3); + partition.repair = Some(armed_session(300, 5)); + journal_prepare(&partition, 6, Operation::CreateStream).await; + partition + .apply_repaired_prepare(repaired_send_prepare(7, 0, 0x77)) + .await; + for op in 8..=133 { + journal_prepare(&partition, op, Operation::CreateStream).await; + } + + let conclusion = partition.complete_repair(&repair_config()).await; + + assert_eq!( + conclusion, + RepairConclusion::FloorRefused { + floor: 5, + to_op: 300 + }, + "a resident leading edge fixes where the window starts, so the \ + refusal is final" + ); + assert!( + partition.repair.is_none(), + "a definitive refusal hands recovery to state transfer" + ); + } + + /// A stall rotation drops the session that repaired ops 6 and 7, and the + /// next session learns the floor. The batches are resident whichever + /// session journaled them, so they anchor the check. + #[compio::test] + async fn given_a_connecting_batch_from_an_earlier_session_when_the_floor_arrives_should_accept_it() + { + let mut partition = test_partition(); + partition.consensus().advance_commit_max(300); + partition.recovered_durable_offset = Some(3); + partition.repair = Some(floorless_session(300, 300)); + partition + .apply_repaired_prepare(repaired_send_prepare_at(6, 4, 0, 0x66)) + .await; + partition + .apply_repaired_prepare(repaired_send_prepare_at(7, 5, 0x66, 0x77)) + .await; + partition.repair = None; + partition.repair = Some(armed_session(300, 5)); + + let conclusion = partition.complete_repair(&repair_config()).await; + + assert_eq!(conclusion, RepairConclusion::InProgress); + assert_eq!( + partition.consensus().commit_min(), + 7, + "the window connects at offset 4, so the floor stands and the walk \ + commits the resident batches above it" + ); + assert!(partition.repair.is_some()); + } + + /// The same shape as above with batches that start past the durable end: + /// resident from an earlier session or not, nothing below them is + /// missing, so the refusal is final. + #[compio::test] + async fn given_a_disconnected_batch_from_an_earlier_session_when_the_floor_arrives_should_refuse_commit_floor() + { + let mut partition = test_partition(); + partition.consensus().advance_commit_max(300); + partition.recovered_durable_offset = Some(3); + partition.repair = Some(floorless_session(300, 300)); + partition + .apply_repaired_prepare(repaired_send_prepare(6, 0, 0x66)) + .await; + partition + .apply_repaired_prepare(repaired_send_prepare(7, 0x66, 0x77)) + .await; + partition.repair = None; + partition.repair = Some(armed_session(300, 5)); + + let conclusion = partition.complete_repair(&repair_config()).await; + + assert_eq!( + conclusion, + RepairConclusion::FloorRefused { + floor: 5, + to_op: 300 + } ); assert_eq!(partition.consensus().commit_min(), 0); + assert!(partition.repair.is_none()); + } + + /// The anchor sits directly above the floor, so the leading edge is empty + /// and the refusal is final at once. + #[compio::test] + async fn given_the_anchor_directly_above_the_floor_when_it_does_not_connect_should_refuse_commit_floor() + { + let mut partition = test_partition(); + partition.consensus().advance_commit_max(300); + partition.recovered_durable_offset = Some(3); + partition.repair = Some(armed_session(300, 5)); + partition + .apply_repaired_prepare(repaired_send_prepare(6, 0, 0x66)) + .await; + + let conclusion = partition.complete_repair(&repair_config()).await; + + assert_eq!( + conclusion, + RepairConclusion::FloorRefused { + floor: 5, + to_op: 300 + } + ); + assert!(partition.repair.is_none()); + } + + /// An op below the anchor is still missing, so a frame for it may yet + /// connect the window. + #[compio::test] + async fn given_a_hole_below_the_anchor_when_the_window_is_incomplete_should_keep_repairing() { + let mut partition = test_partition(); + partition.consensus().advance_commit_max(300); + partition.recovered_durable_offset = Some(3); + partition.repair = Some(armed_session(300, 5)); + journal_prepare(&partition, 6, Operation::CreateStream).await; + partition + .apply_repaired_prepare(repaired_send_prepare(8, 0, 0x88)) + .await; + + let conclusion = partition.complete_repair(&repair_config()).await; + + assert_eq!(conclusion, RepairConclusion::InProgress); assert!(partition.repair.is_some()); } + /// Round one repaired op 4 and lost op 5. Before the stall retry the peer + /// evicted through op 5, so the floor lands at 5 and the retry serves only + /// ops 6 to 10. Op 4 connects to the durable end, but the floor claims + /// recovered state stands in for op 5, which this replica holds nowhere. + /// Accepting that floor commits and serves a log missing offset 5. + #[compio::test] + async fn given_a_connecting_batch_below_a_raised_floor_when_the_floor_arrives_should_refuse_it() + { + let mut partition = test_partition(); + partition.consensus().advance_commit_max(10); + partition.recovered_durable_offset = Some(3); + partition.repair = Some(armed_session(10, 5)); + partition + .apply_repaired_prepare(repaired_send_prepare(4, 0, 0x44)) + .await; + for op in 6..=10u64 { + partition + .apply_repaired_prepare(repaired_send_prepare(op, 0, 0x100 + u128::from(op))) + .await; + } + + let conclusion = partition.complete_repair(&repair_config()).await; + + assert_eq!( + conclusion, + RepairConclusion::FloorRefused { + floor: 5, + to_op: 10 + }, + "a batch at or below the floor must not anchor the connect check" + ); + assert!( + partition.consensus().commit_min() < 5, + "the floor must not skip op 5" + ); + } + + /// The first stream repaired op 4 and the walk committed through it. The + /// stall retry then learns a floor at 6, above the still-missing op 5. + /// The committed op 4 must not stand in for the window above the floor. + #[compio::test] + async fn given_a_committed_batch_below_the_floor_when_the_floor_arrives_should_keep_repairing() + { + let mut partition = test_partition(); + partition.consensus().advance_commit_max(300); + partition.recovered_durable_offset = Some(3); + partition.repair = Some(armed_session(300, 6)); + for op in 1..=3 { + journal_prepare(&partition, op, Operation::CreateStream).await; + } + partition + .apply_repaired_prepare(repaired_send_prepare(4, 0, 0x44)) + .await; + partition.commit_journal(&repair_config()).await; + assert_eq!(partition.consensus().commit_min(), 4); + + let conclusion = partition.complete_repair(&repair_config()).await; + + assert_eq!(conclusion, RepairConclusion::InProgress); + assert_eq!( + partition.consensus().commit_min(), + 4, + "the floor must not skip op 5" + ); + assert!(partition.repair.is_some()); + } + + /// Ops up to the floor were resident but uncommitted, so the walk inside + /// the refusal passes the floor. With lazy flush their headers stay + /// resident and the leading edge reads as complete, but a floor below + /// `commit_min` moves nothing and is no reason for a state transfer. + #[compio::test] + async fn given_lazy_flush_when_the_walk_passes_a_disconnected_floor_should_keep_repairing() { + let mut partition = test_partition(); + partition.consensus().advance_commit_max(300); + partition.recovered_durable_offset = Some(3); + partition.repair = Some(armed_session(300, 5)); + for op in 1..=6 { + journal_prepare(&partition, op, Operation::CreateStream).await; + } + partition + .apply_repaired_prepare(repaired_send_prepare(7, 0, 0x77)) + .await; + let lazy = PartitionsConfig { + messages_required_to_save: 10_000, + size_of_messages_required_to_save: IggyByteSize::from(512 * 1024 * 1024), + ..repair_config() + }; + + let conclusion = partition.complete_repair(&lazy).await; + + assert_eq!(conclusion, RepairConclusion::InProgress); + assert_eq!(partition.consensus().commit_min(), 7); + assert!( + partition.log.journal().inner.holds_op(6), + "lazy flush keeps the committed headers resident" + ); + assert!(partition.repair.is_some()); + } + + /// Repair frames arrive out of order. Op 7 connects at offset 4, op 9 does + /// not, and the window is judged by the lower one whichever lands first. + #[compio::test] + async fn given_batches_applied_out_of_order_when_the_floor_arrives_should_anchor_on_the_lowest() + { + for ops in [[9, 7], [7, 9]] { + let mut partition = test_partition(); + partition.consensus().advance_commit_max(300); + partition.recovered_durable_offset = Some(3); + partition.repair = Some(armed_session(300, 5)); + for op in ops { + let base_offset = if op == 7 { 4 } else { 6 }; + partition + .apply_repaired_prepare(repaired_send_prepare_at( + op, + base_offset, + 0, + u128::from(op), + )) + .await; + } + + let conclusion = partition.complete_repair(&repair_config()).await; + + assert_eq!(conclusion, RepairConclusion::InProgress, "order {ops:?}"); + assert_eq!( + partition.consensus().commit_min(), + 5, + "order {ops:?}: the lowest batch connects, so the floor stands" + ); + } + } + + /// A floor holds the walk and the first served chunk carries no message, + /// so nothing anchors the window yet. Pulling again from `commit_min + 1` + /// re-serves that same chunk forever; the pull must start past it. + #[compio::test] + async fn given_a_floor_holding_the_walk_when_the_first_chunk_has_no_batch_should_pull_past_it() + { + const FLOOR: u64 = 5; + // One full shard repair chunk above the floor. + const FIRST_CHUNK_LAST: u64 = FLOOR + 128; + const TO_OP: u64 = FIRST_CHUNK_LAST + 10; + const SEND_OP: u64 = FIRST_CHUNK_LAST + 7; + let mut partition = test_partition(); + partition.consensus().advance_commit_max(TO_OP); + partition.recovered_durable_offset = Some(3); + partition.repair = Some(armed_session(TO_OP, FLOOR)); + for op in FLOOR + 1..=FIRST_CHUNK_LAST { + journal_prepare(&partition, op, Operation::CreateStream).await; + } + + for _ in 0..2 { + assert_eq!( + partition.complete_repair(&repair_config()).await, + RepairConclusion::InProgress + ); + assert_eq!(partition.consensus().commit_min(), 0); + } + assert_eq!(partition.repair_pull_from(), Some(FIRST_CHUNK_LAST + 1)); + assert_eq!(partition.next_floor_pull(), Some(FIRST_CHUNK_LAST + 1)); + assert_eq!( + partition.next_floor_pull(), + None, + "nothing new is resident, so the next pull waits for the stall retry" + ); + + for op in FIRST_CHUNK_LAST + 1..=TO_OP { + if op == SEND_OP { + partition + .apply_repaired_prepare(repaired_send_prepare_at(op, 200, 0, u128::from(op))) + .await; + } else { + journal_prepare(&partition, op, Operation::CreateStream).await; + } + } + + assert_eq!( + partition.complete_repair(&repair_config()).await, + RepairConclusion::FloorRefused { + floor: FLOOR, + to_op: TO_OP + } + ); + } + + /// A batch at or below the local purge fence carries purged bytes, which + /// cannot stand in for durable state even when its offset would connect. + /// The anchor search skips it, and the `(None, _)` arm still counts it as + /// a message, which is what refuses here. + #[compio::test] + async fn given_a_connecting_batch_at_the_purge_fence_when_the_floor_arrives_should_refuse_it() { + let mut partition = test_partition(); + partition.consensus().advance_commit_max(6); + partition.recovered_durable_offset = Some(3); + partition.purge_floor_op = 6; + partition.repair = Some(armed_session(6, 5)); + partition + .apply_repaired_prepare(repaired_send_prepare_at(6, 4, 0, 0x66)) + .await; + + let conclusion = partition.complete_repair(&repair_config()).await; + + assert_eq!( + conclusion, + RepairConclusion::FloorRefused { floor: 5, to_op: 6 } + ); + assert_eq!(partition.consensus().commit_min(), 0); + } + /// A `RangeEvicted` floor arrives above `commit_min`, and the walk passes /// it before `RepairDone` lands: ops resident just above the commit point /// committed and their headers were evicted. Verifying that moot floor @@ -23963,7 +24534,7 @@ mod tests { // the boot-recovered segments end at 10. Meaningless below commit_min. partition.recovered_durable_offset = Some(10); journal_prepare(&partition, 8, Operation::CreateStream).await; - partition.repair = Some(armed_session(8, 5, Some(20))); + partition.repair = Some(armed_session(8, 5)); let conclusion = partition.complete_repair(&repair_config()).await; @@ -23982,7 +24553,7 @@ mod tests { partition.consensus().restore_commit_state(5, 5); // The peer retains nothing below op 10, so it cannot serve the suffix // either. Only the raw floor tells this apart from a moot one. - partition.repair = Some(armed_fetch_session(5, 9, 9, None)); + partition.repair = Some(armed_fetch_session(5, 9, 9)); let conclusion = partition.complete_repair(&repair_config()).await; @@ -24002,7 +24573,7 @@ mod tests { // The peer retains from op 6, everything this replica still needs, so // the suffix fetch must go on. Escaping to state transfer here copied // segments for a window the peer can serve. - partition.repair = Some(armed_fetch_session(5, 9, 5, None)); + partition.repair = Some(armed_fetch_session(5, 9, 5)); let conclusion = partition.complete_repair(&repair_config()).await; @@ -24960,7 +25531,7 @@ mod tests { .await; let partition = &mut fixture.partition; partition.log.journal().inner.set_repair_retention(true); - partition.repair = Some(armed_session(4, 0, None)); + partition.repair = Some(armed_session(4, 0)); let prepares: Vec<_> = (1..=4) .map(|op| repaired_send_prepare(op, 0, u128::from(op)).into_frozen()) .collect(); @@ -26331,15 +26902,11 @@ mod purge_floor_tests { header.size = u32::try_from(total).expect("prepare size fits u32"); }); - let base_offset = partition + partition .append_repaired_send_messages(message) .await .expect("a fenced repaired prepare is journaled, not refused"); - assert_eq!( - base_offset, None, - "a purged batch must not anchor the repair floor's connect check" - ); assert_eq!( partition.dirty_offset.load(Ordering::Relaxed), 0, @@ -26429,7 +26996,7 @@ mod purge_floor_tests { const OPS: u64 = COMMIT_WALK_OPS_MAX as u64 + 1; let (mut partition, dir) = purge_test_partition("persisted-twice"); let record_len = build_segment_record(IggyNamespace::new(1, 1, 0), 1).len() as u64; - partition.repair = Some(armed_session(OPS, 0, None)); + partition.repair = Some(armed_session(OPS, 0)); for op in 1..=OPS { partition @@ -26466,7 +27033,7 @@ mod purge_floor_tests { is what lets the repair ingest re-deliver it" ); - partition.repair = Some(armed_session(OPS, 0, None)); + partition.repair = Some(armed_session(OPS, 0)); partition .apply_repaired_prepare(repaired_send_prepare(OPS, 0, CHECKSUM)) .await; diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index 3d803a360..4deca674c 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -402,6 +402,40 @@ impl PartitionJournal<PartitionJournalMemStorage> { ring.get(&op).cloned() } + /// The lowest resident `SendMessages` op in `ops` with its entry bytes, in + /// one pass over the headers. Resident only: the evicted ring holds what + /// already committed, which is not what a repair window is judged by. + #[must_use] + pub fn lowest_resident_send_in( + &self, + ops: RangeInclusive<u64>, + ) -> Option<(u64, JournalBuffer)> { + let op = { + let headers = unsafe { &*self.headers.get() }; + headers + .iter() + .filter(|header| { + header.operation == Operation::SendMessages && ops.contains(&header.op) + }) + .map(|header| header.op) + .min()? + }; + let entry = { + let op_to_storage_offset = unsafe { &*self.op_to_storage_offset.get() }; + let inner = unsafe { &*self.inner.get() }; + op_to_storage_offset + .get(&op) + .and_then(|&storage_offset| inner.storage.read_at_sync(storage_offset)) + }; + if entry.is_none() { + warn!( + op, + "resident SendMessages header has no stored entry; the repair window has no anchor" + ); + } + entry.map(|entry| (op, entry)) + } + /// The header at `op`, over exactly the range [`Self::repair_entry`] serves. /// /// NOT [`Self::header_by_op`], which reads the resident headers alone. The diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs index c1f578676..6b351a7a7 100644 --- a/core/partitions/src/types.rs +++ b/core/partitions/src/types.rs @@ -303,16 +303,13 @@ pub struct RepairSession { pub floor: Option<u64>, /// The peer serving this stream (re-request target on stall). pub peer: u8, - /// Lowest `base_offset` among the repaired `SendMessages` batches: - /// where the served window begins in offset space. Compared against the - /// boot-recovered durable end when a commit floor arrives -- a window - /// starting above `recovered_durable_offset + 1` means ops below the - /// floor are neither locally durable nor repaired (state-transfer - /// territory), and the floor must be refused. - pub first_batch_offset: Option<u64>, /// Ticks since the stream last made progress; at /// [`REPAIR_RETRY_TICKS`] the remaining window is re-requested. pub idle_ticks: u32, + /// Start of the last chunk pulled above a pending [`Self::floor`]. A + /// `RepairDone` pulls again only past it, so a peer that serves nothing + /// is left to the stall retry instead of answering an unthrottled loop. + pub floor_pulled_from: u64, } /// How a repair-window commit walk concluded, decided by diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index c0870f1f8..b67b4381d 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -3083,8 +3083,8 @@ mod tests { fetch_to_op: 5, floor: None, peer: 1, - first_batch_offset: None, idle_ticks: 0, + floor_pulled_from: 0, }); // Committed purge: generation 1 > applied 0. @@ -3206,8 +3206,8 @@ mod tests { fetch_to_op: 8, floor: None, peer: 1, - first_batch_offset: None, idle_ticks: 0, + floor_pulled_from: 0, }); for op in 1..=8 { let parent = partition.consensus().last_prepare_checksum(); @@ -3235,7 +3235,6 @@ mod tests { partition.recovered_durable_offset = Some(10); let repair = partition.repair.as_mut().expect("repair remains armed"); repair.floor = Some(5); - repair.first_batch_offset = Some(20); } Box::pin(compio::time::timeout( diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index de75f91c1..c88f83a63 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -2790,23 +2790,49 @@ where ) } +/// Whether a transfer descriptor comes from a replica that knows less than this +/// one. The serving view is compared against `log_view`, not `view`: a replica +/// that needs state transfer cannot join a view change, so its `view` runs ahead. +const fn transfer_offer_is_behind<B, P>( + serving_view: u32, + serving_commit_max: u64, + local: &VsrConsensus<B, P>, +) -> bool +where + B: MessageBus, + P: Pipeline<Entry = consensus::PipelineEntry>, +{ + serving_view < local.log_view() || serving_commit_max < local.commit_max() +} + /// The next replica to try after a transfer against `failed_peer` failed. /// -/// Prefers the view's primary: it is the only replica that can pass the serving +/// Prefers the group's primary: it is the only replica that can pass the serving /// side's primary gate, so rotating by ring index alone can spend a /// full backoff round on a backup that must refuse -- and, worse, can land on a /// phantom view-0 primary of an empty group. Falls back to walking the ring past /// the failed peer, skipping this replica; a cluster of two has no alternative /// and retries the same peer. /// +/// `primary` is `None` when the group's primary is unknown, which also walks +/// the ring. +/// /// `failed_peer` and `primary` are both bounded by `replica_count` at their /// ingress, which is what keeps the `+ 1` here from wrapping a peer id of 255 /// onto replica 0. -const fn next_transfer_peer(self_id: u8, failed_peer: u8, replica_count: u8, primary: u8) -> u8 { +const fn next_transfer_peer( + self_id: u8, + failed_peer: u8, + replica_count: u8, + primary: Option<u8>, +) -> u8 { if replica_count <= 1 { return failed_peer; } - if primary != self_id && primary != failed_peer { + if let Some(primary) = primary + && primary != self_id + && primary != failed_peer + { return primary; } let mut candidate = (failed_peer + 1) % replica_count; @@ -4980,6 +5006,14 @@ where } let config = planes.1.0.config(); + // Before the view filter: a replica whose view climbed through + // elections nobody joined drops the real primary's lower-view + // heartbeats, and they are how it learns whom to repair from. + if let Some(partition) = planes.1.0.get_by_ns(&IggyNamespace::from_raw(header.group)) { + partition + .consensus() + .observe_primary(header.view, header.replica); + } let Some(partition) = self.resolve_partition_target( &planes.1.0, header.group, @@ -5795,9 +5829,10 @@ where // `complete_repair` walks the window and clears the session // only when the LOCAL commit frontier reached the requested // op (the peer's served-through claim proves nothing about - // delivery on a lossy bus). While the walk makes progress - // the next chunk is pulled immediately; a stalled window is - // left to the retry timer. + // delivery on a lossy bus). While the walk makes progress, or + // a floor holds it and the resident prefix grew, the next chunk + // is pulled immediately; a stalled window is left to the retry + // timer. let before = partition.consensus().commit_min(); if let partitions::RepairConclusion::FloorRefused { floor, to_op } = partition.complete_repair(&config).await @@ -5859,9 +5894,18 @@ where ); } else { let commit_min = partition.consensus().commit_min(); - let next = partition.repair.as_ref().and_then(|live| { - partition_repair_next_chunk(before, commit_min, live.fetch_to_op) - .map(|from_op| (live.peer, live.nonce, from_op, live.fetch_to_op)) + let from_op = partition + .repair + .as_ref() + .and_then(|live| { + partition_repair_next_chunk(before, commit_min, live.fetch_to_op) + }) + .or_else(|| partition.next_floor_pull()); + let next = from_op.and_then(|from_op| { + partition + .repair + .as_ref() + .map(|live| (live.peer, live.nonce, from_op, live.fetch_to_op)) }); let cluster = partition.consensus().cluster(); let self_id = partition.consensus().replica(); @@ -6275,7 +6319,7 @@ where consensus.replica(), peer, consensus.replica_count(), - primary, + Some(primary), ); if next_peer == peer { // The ring had nobody else to offer (a solo group, or a two-replica @@ -7964,6 +8008,7 @@ where ); continue; } + let pull_from = partition.repair_pull_from(); let due = partition.repair.as_mut().and_then(|session| { if !session_live { return None; @@ -7973,10 +8018,14 @@ where return None; } session.idle_ticks = 0; + // A fully resident window puts the pull past `fetch_to_op`, + // yet only a `RepairDone` completes the session. Asking again + // for the last op draws one, and keeps the stall budget able + // to rotate off a dead peer. Some(( session.peer, session.nonce, - commit_min.saturating_add(1), + pull_from?.min(session.fetch_to_op), session.fetch_to_op, cluster, self_id, @@ -7991,17 +8040,19 @@ where // that. due.map(|stalled| (stalled, partition.burn_repair_attempt())) }; - if let Some(((peer, nonce, from_op, to_op, cluster, self_id), rotate)) = stalled - && from_op <= to_op - { + if let Some(((peer, nonce, from_op, to_op, cluster, self_id), rotate)) = stalled { if rotate { let Some(partition) = partitions.get_mut_by_ns(&namespace) else { continue; }; let consensus = partition.consensus(); - let primary = consensus.primary_index(consensus.view()); - let next_peer = - next_transfer_peer(self_id, peer, consensus.replica_count(), primary); + consensus.forget_primary(peer); + let next_peer = next_transfer_peer( + self_id, + peer, + consensus.replica_count(), + consensus.primary_hint(), + ); tracing::warn!( shard = self.id, namespace_raw = namespace.inner(), @@ -8105,7 +8156,7 @@ where let peer = gap_repair_peer( consensus.replica(), consensus.replica_count(), - consensus.primary_index(consensus.view()), + consensus.primary_hint(), ); if peer.is_none() { // Restart the debounce so this repeats at its @@ -9108,8 +9159,8 @@ where fetch_to_op, floor: None, peer, - first_batch_offset: None, idle_ticks: 0, + floor_pulled_from: 0, }); tracing::info!( shard = self.id, @@ -9194,8 +9245,8 @@ where fetch_to_op: to_op, floor: None, peer, - first_batch_offset: None, idle_ticks: 0, + floor_pulled_from: 0, }); tracing::info!( shard = self.id, @@ -9244,6 +9295,13 @@ where if !session_matches { return; } + // Offers only: a refusal is sent by whoever was asked, primary or not, + // and stamps the sender's own view, which climbs alone while it lags. + if header.available == 1 { + partition + .consensus() + .observe_primary(header.view, header.replica); + } if header.available == 0 { // A refusal the peer marked transient (its offer build is still // flushing or hashing, which is routine under produce load) must @@ -9262,20 +9320,17 @@ where "partition transfer peer cannot serve; backing off before re-arming" ); if transient { - // The peer that refused is the node that would otherwise serve, - // and on the partition arm only the primary can. Keep - // asking it unless it is not the primary this replica knows: a - // rotation spends the next round on a backup that can only - // refuse, and the serving side's partial offer-build progress - // is memoized per node, so that round advances no hashing. - let primary = { - let consensus = partition.consensus(); - consensus.primary_index(consensus.view()) - }; + // Keep asking the refusing peer only while it is the primary + // this replica last heard from: only the primary serves, and its + // partial offer-build progress is memoized per node. The + // refusal's own view proves nothing, since a lagging replica + // stamps a view that climbed alone. + let is_hinted_primary = + partition.consensus().primary_hint() == Some(header.replica); self.rearm_partition_transfer_after_refusal( partition, header.replica, - header.replica != primary, + !is_hinted_primary, ) .await; } else { @@ -9291,6 +9346,7 @@ where // caught up). Installing it would unlink a chain this replica already // holds; nonce match alone cannot tell the two apart. let local_view = partition.consensus().view(); + let local_log_view = partition.consensus().log_view(); let local_commit_max = partition.consensus().commit_max(); // `commit_op` past the sender's OWN `commit_max` is self-contradictory: // the offer cannot be built past the frontier its builder had. Nothing @@ -9313,7 +9369,7 @@ where .await; return; } - if header.view < local_view || header.commit_max < local_commit_max { + if transfer_offer_is_behind(header.view, header.commit_max, partition.consensus()) { tracing::warn!( shard = self.id, namespace_raw = header.group, @@ -9321,6 +9377,7 @@ where serving_view = header.view, serving_commit_max = header.commit_max, local_view, + local_log_view, local_commit_max, "refusing a partition transfer offer from a replica behind this one" ); @@ -9926,11 +9983,12 @@ where consensus.set_state_transfer_stage(consensus::StateTransferStage::Idle); } let next_peer = if rotate { + consensus.forget_primary(peer); next_transfer_peer( consensus.replica(), peer, consensus.replica_count(), - consensus.primary_index(consensus.view()), + consensus.primary_hint(), ) } else { peer @@ -10310,7 +10368,7 @@ where match gap_repair_peer( consensus.replica(), consensus.replica_count(), - consensus.primary_index(consensus.view()), + Some(consensus.primary_index(consensus.view())), ) { None => { // Restart the debounce so this repeats at its interval, @@ -10422,7 +10480,7 @@ where if let Some(next_peer) = gap_repair_peer( consensus.replica(), consensus.replica_count(), - consensus.primary_index(consensus.view()), + Some(consensus.primary_index(consensus.view())), ) { self.maybe_request_metadata_repair(consensus, next_peer) .await; @@ -11060,11 +11118,10 @@ const fn drive_group_gap_debounce( /// /// Shared by both planes so the rule cannot drift: the partition sweep and /// `tick_metadata` arm off the same predicate and owe the same answer. -const fn gap_repair_peer(self_id: u8, replica_count: u8, primary: u8) -> Option<u8> { - let peer = if primary == self_id { - next_transfer_peer(self_id, self_id, replica_count, primary) - } else { - primary +const fn gap_repair_peer(self_id: u8, replica_count: u8, primary: Option<u8>) -> Option<u8> { + let peer = match primary { + Some(primary) if primary != self_id => primary, + _ => next_transfer_peer(self_id, self_id, replica_count, primary), }; // A solo group (or a ring with nobody else live to name) rotates back to // self, which no session can be opened against. @@ -13354,8 +13411,8 @@ mod metadata_repair_session_tests { #[test] fn given_a_gap_stopped_backup_when_picking_a_peer_should_ask_the_primary() { - assert_eq!(gap_repair_peer(2, 3, 0), Some(0)); - assert_eq!(gap_repair_peer(1, 5, 3), Some(3)); + assert_eq!(gap_repair_peer(2, 3, Some(0)), Some(0)); + assert_eq!(gap_repair_peer(1, 5, Some(3)), Some(3)); } #[test] @@ -13365,16 +13422,23 @@ mod metadata_repair_session_tests { // self-addressed request fails to send AFTER the session is recorded. for replica_count in 2..=7u8 { for primary in 0..replica_count { - let peer = gap_repair_peer(primary, replica_count, primary); + let peer = gap_repair_peer(primary, replica_count, Some(primary)); assert_ne!(peer, Some(primary), "count {replica_count}"); assert!(peer.is_some(), "count {replica_count}"); } } } + #[test] + fn given_an_unknown_primary_when_picking_a_peer_should_walk_the_ring() { + assert_eq!(gap_repair_peer(0, 5, None), Some(1)); + assert_eq!(next_transfer_peer(0, 1, 5, None), 2); + assert_eq!(next_transfer_peer(0, 4, 5, None), 1); + } + #[test] fn given_a_solo_group_when_picking_a_peer_should_answer_nobody() { - assert_eq!(gap_repair_peer(0, 1, 0), None); + assert_eq!(gap_repair_peer(0, 1, Some(0)), None); } #[test] @@ -13384,8 +13448,9 @@ mod metadata_repair_session_tests { for replica_count in 3..=7u8 { for primary in 0..replica_count { let self_id = (primary + 1) % replica_count; - let failed = gap_repair_peer(self_id, replica_count, primary).expect("a peer"); - let next = next_transfer_peer(self_id, failed, replica_count, primary); + let failed = + gap_repair_peer(self_id, replica_count, Some(primary)).expect("a peer"); + let next = next_transfer_peer(self_id, failed, replica_count, Some(primary)); assert_ne!(next, failed, "count {replica_count}, primary {primary}"); assert_ne!(next, self_id, "count {replica_count}, primary {primary}"); } @@ -13396,7 +13461,7 @@ mod metadata_repair_session_tests { fn given_two_replicas_when_the_stall_budget_is_spent_should_name_the_same_peer_back() { // Which is how the caller reads "nobody else to ask" and drops the // session instead of re-arming it. - assert_eq!(next_transfer_peer(1, 0, 2, 0), 0); + assert_eq!(next_transfer_peer(1, 0, 2, Some(0)), 0); } #[test] @@ -13481,6 +13546,57 @@ mod metadata_repair_session_tests { } } +#[cfg(test)] +mod transfer_descriptor_gate_tests { + //! Which state-transfer descriptors a replica refuses as coming from a + //! replica behind it. + + use super::transfer_offer_is_behind; + use consensus::{LocalPipeline, VsrConsensus}; + use message_bus::IggyMessageBus; + + /// A replica whose log was last adopted in `log_view`, whose own view has + /// since moved to `view`, and which knows `commit_max`. + fn replica(view: u32, log_view: u32, commit_max: u64) -> VsrConsensus<IggyMessageBus> { + let mut consensus = + VsrConsensus::new(1, 0, 3, 42, IggyMessageBus::new(0), LocalPipeline::new()); + consensus.init(); + consensus.set_view(view); + consensus.set_log_view(log_view); + consensus.advance_commit_max(commit_max); + consensus + } + + #[test] + fn given_the_primary_offer_when_this_replica_ratcheted_its_view_alone_should_accept() { + // Numbers from a reproduction: the lagging replica last adopted view + // 11, solo elections that nobody heard pushed its view to 58, and the + // group's primary at view 20 offers more committed state than it holds. + let lagging = replica(58, 11, 53_052); + assert!(!transfer_offer_is_behind(20, 55_013, &lagging)); + } + + #[test] + fn given_an_offer_from_before_the_view_this_log_adopted_when_checked_should_refuse() { + let replica = replica(11, 11, 53_052); + assert!(transfer_offer_is_behind(10, 55_013, &replica)); + } + + #[test] + fn given_an_offer_below_the_known_commit_point_when_checked_should_refuse() { + let replica = replica(58, 11, 53_052); + assert!(transfer_offer_is_behind(20, 53_000, &replica)); + } + + #[test] + fn given_a_phantom_view_zero_primary_when_checked_should_refuse() { + // An empty group at view 0 must never unlink a chain this replica + // holds, whichever comparison catches it. + assert!(transfer_offer_is_behind(0, 0, &replica(3, 1, 0))); + assert!(transfer_offer_is_behind(0, 0, &replica(0, 0, 42))); + } +} + #[cfg(test)] mod partition_ack_durability_tests { use super::*; diff --git a/core/shard/src/partition_io.rs b/core/shard/src/partition_io.rs index 11aafd587..4e6df621f 100644 --- a/core/shard/src/partition_io.rs +++ b/core/shard/src/partition_io.rs @@ -2095,8 +2095,8 @@ mod tests { fetch_to_op: 0, floor: None, peer: 1, - first_batch_offset: None, idle_ticks: 0, + floor_pulled_from: 0, }); let reply = Message::<RepairRangeReplyHeader>::new(size_of::<RepairRangeReplyHeader>()) @@ -2167,8 +2167,8 @@ mod tests { fetch_to_op: 0, floor: None, peer: 0, - first_batch_offset: None, idle_ticks: 0, + floor_pulled_from: 0, }); assert_eq!( partition.complete_repair(partitions.config()).await, diff --git a/core/shard/src/poll/completion_tests.rs b/core/shard/src/poll/completion_tests.rs index 5e8f8d4f0..a13e5f042 100644 --- a/core/shard/src/poll/completion_tests.rs +++ b/core/shard/src/poll/completion_tests.rs @@ -32,13 +32,15 @@ use consensus::{ use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; use iggy_binary_protocol::requests::topics::{DeleteTopicRequest, PurgeTopicRequest}; use iggy_binary_protocol::{ - AckLevel, ReplyHeader, RequestPreparesHeader, RequestStateChunkHeader, - RequestStateTransferHeader, RoutedRequestHeader, StateTransferTargetHeader, WireConsumer, + AckLevel, CommitHeader, RepairRangeReplyHeader, ReplyHeader, RequestPreparesHeader, + RequestStateChunkHeader, RequestStateTransferHeader, RoutedRequestHeader, + StateTransferTargetHeader, WireConsumer, }; use iggy_binary_protocol::{ Command, ConsensusHeader, Operation, PrepareHeader, WireEncode, WireIdentifier, }; -use iggy_common::{IggyError, IggyTimestamp, PollingStrategy}; +use iggy_common::{IggyError, IggyTimestamp, PartitionStats, PollingStrategy}; +use journal::Journal; use journal::prepare_journal::PrepareJournal; use message_bus::IggyMessageBus; use metadata::IggyMetadata; @@ -47,9 +49,13 @@ use metadata::stm::StateMachine; use metadata::stm::consumer_group::{ConsumerGroup, ConsumerGroupMember, JoinConsumerGroupRequest}; use metadata::stm::stream::{Partition, Stream, StreamsInner, Topic}; use metadata::stm::user::Users; -use partitions::{IggyPartitions, PartitionsConfig, PollingArgs, PollingConsumer}; +use partitions::state_transfer::PartitionTransferSession; +use partitions::{ + IggyPartition, IggyPartitions, PartitionsConfig, PollingArgs, PollingConsumer, RepairSession, +}; +use server_common::MESSAGE_ALIGN; use server_common::Message; -use server_common::iobuf::Owned; +use server_common::iobuf::{Frozen, Owned}; use server_common::send_messages::decode_batch_slice; use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId}; @@ -58,8 +64,8 @@ use crate::metrics::ShardMetrics; use crate::shards_table::{PapayaShardsTable, ShardsTable}; use crate::{ ConsumerAttachment, IggyShard, LifecycleFrame, NoopHost, PartitionConsensusConfig, - PartitionRead, PartitionReadReply, Receiver, ReplicaTopology, ShardFrame, ShardIdentity, - TaggedSender, channel, shard_channel, + PartitionRead, PartitionReadReply, REPAIR_CHUNK_MAX, Receiver, ReplicaTopology, ShardFrame, + ShardIdentity, TaggedSender, channel, shard_channel, }; #[compio::test] @@ -1215,6 +1221,798 @@ async fn given_accepted_descriptor_when_install_fails_should_exhaust_its_generat ); } +const LAGGING_NONCE: u128 = 0x5eed; +const LAGGING_LOG_VIEW: u32 = 11; +const LAGGING_VIEW: u32 = 58; +const LAGGING_COMMIT_MAX: u64 = 53_052; +const GROUP_VIEW: u32 = 20; +const GROUP_COMMIT: u64 = 55_013; +const GROUP_PRIMARY: u8 = 2; + +/// What a partition transfer descriptor answers. +#[derive(Clone, Copy)] +enum DescriptorReply { + Offer, + TransientRefusal, + HardRefusal, +} + +/// Replica 0 of three, awaiting a transfer descriptor from `GROUP_PRIMARY`. Its +/// log was last adopted in `LAGGING_LOG_VIEW` while elections nobody heard +/// raised its own view to `LAGGING_VIEW`, so `primary_index(view())` names a +/// different replica than the group's real primary. +fn lagging_replica_awaiting_target(namespace: IggyNamespace) -> CompletionTestShard { + let owner = lagging_replica_in_cluster(namespace, 3); + { + let partitions = owner.plane.partitions(); + let consensus = partitions + .get_by_ns(&namespace) + .expect("the fixture registers the partition") + .consensus(); + assert_ne!(consensus.primary_index(LAGGING_VIEW), GROUP_PRIMARY); + assert_eq!(consensus.primary_index(GROUP_VIEW), GROUP_PRIMARY); + } + owner +} + +/// Replica 0 of `replica_count` in the lagging state above, with a transfer +/// session awaiting a descriptor from `GROUP_PRIMARY`. +fn lagging_replica_in_cluster(namespace: IggyNamespace, replica_count: u8) -> CompletionTestShard { + let bus = Rc::new(IggyMessageBus::new(0)); + let config = partitions_config(); + let metadata = IggyMetadata::new(None, None, None, None, PollTestMetadata::default(), None); + let (owner, _sender) = + owner_in_cluster(&bus, config.clone(), namespace, metadata, replica_count); + let mut consensus = VsrConsensus::new( + 1, + 0, + replica_count, + namespace.inner(), + bus, + LocalPipeline::new(), + ); + consensus.init(); + consensus.set_view(LAGGING_VIEW); + consensus.set_log_view(LAGGING_LOG_VIEW); + consensus.advance_commit_max(LAGGING_COMMIT_MAX); + consensus.set_state_transfer_stage(StateTransferStage::AwaitingTarget); + let mut partition = IggyPartition::with_in_memory_storage( + Arc::new(PartitionStats::default()), + consensus, + config.segment_size, + ); + arm_lagging_transfer(&mut partition, GROUP_PRIMARY); + owner.plane.partitions().insert(namespace, partition); + owner +} + +fn arm_lagging_transfer<B, S>(partition: &mut IggyPartition<B, S>, peer: u8) +where + B: message_bus::MessageBus, +{ + partition.transfer = Some(PartitionTransferSession { + nonce: LAGGING_NONCE, + peer, + commit_op: 0, + artifacts: Vec::new(), + target_accepted: false, + idle_ticks: 0, + }); +} + +/// A descriptor from `GROUP_PRIMARY` at `GROUP_VIEW`: an offer at +/// `GROUP_COMMIT` when `offer` is set, a transient refusal otherwise. +fn group_primary_descriptor( + namespace: IggyNamespace, + offer: bool, +) -> Message<StateTransferTargetHeader> { + let reply = if offer { + DescriptorReply::Offer + } else { + DescriptorReply::TransientRefusal + }; + lagging_descriptor(namespace, GROUP_PRIMARY, GROUP_VIEW, reply) +} + +/// A descriptor answering the lagging replica's transfer session, sent by +/// `replica` at `view`. +fn lagging_descriptor( + namespace: IggyNamespace, + replica: u8, + view: u32, + reply: DescriptorReply, +) -> Message<StateTransferTargetHeader> { + let manifest = if matches!(reply, DescriptorReply::Offer) { + encode_state_manifest(&[StateArtifact::for_bytes( + artifact_kind::SEGMENT_LOG, + GROUP_COMMIT, + b"segment", + )]) + } else { + Vec::new() + }; + let size = size_of::<StateTransferTargetHeader>() + manifest.len(); + let mut message = Message::<StateTransferTargetHeader>::new(size); + message.as_mut_slice()[size_of::<StateTransferTargetHeader>()..].copy_from_slice(&manifest); + message.transmute_header(|_, header: &mut StateTransferTargetHeader| { + header.command = Command::StateTransferTarget; + header.cluster = 1; + header.replica = replica; + header.view = view; + header.group = namespace.inner(); + header.nonce = LAGGING_NONCE; + header.size = u32::try_from(size).unwrap(); + header.commit_max = GROUP_COMMIT; + match reply { + DescriptorReply::Offer => { + header.available = 1; + header.commit_op = GROUP_COMMIT; + } + DescriptorReply::TransientRefusal => header.unavailable_transient = 1, + DescriptorReply::HardRefusal => {} + } + header.seal(); + }) +} + +/// A partition `Commit` heartbeat from `replica` as the primary of `view`. +fn primary_commit( + namespace: IggyNamespace, + replica: u8, + view: u32, + commit: u64, +) -> Message<CommitHeader> { + Message::<CommitHeader>::new(size_of::<CommitHeader>()).transmute_header( + |_, header: &mut CommitHeader| { + header.command = Command::Commit; + header.cluster = 1; + header.replica = replica; + header.view = view; + header.commit = commit; + header.group = namespace.inner(); + header.size = u32::try_from(size_of::<CommitHeader>()).unwrap(); + header.seal(); + }, + ) +} + +/// Restart the lagging replica's transfer against `peer` after a re-arm. +fn rearm_lagging_transfer_to(owner: &CompletionTestShard, namespace: IggyNamespace, peer: u8) { + let partitions = owner.plane.partitions(); + let partition = partitions + .get_mut_by_ns(&namespace) + .expect("the partition stays registered"); + partition.transfer_rearm = None; + arm_lagging_transfer(partition, peer); +} + +/// The peer the lagging replica's next transfer attempt is scheduled against. +fn scheduled_rearm_peer(owner: &CompletionTestShard, namespace: IggyNamespace) -> u8 { + owner + .plane + .partitions() + .get_by_ns(&namespace) + .expect("the partition stays registered") + .transfer_rearm + .expect("a refusal schedules a re-arm") + .peer +} + +/// A replica that needs partition state transfer cannot finish a view change, +/// and with `persisted` durability it cannot vote in one either, so its own +/// `view` climbs past the group's. Refusing the group primary's offer as "from +/// a replica behind this one" keeps it behind for good. +#[compio::test] +async fn given_the_primary_offer_when_this_replica_ratcheted_its_view_alone_should_accept_it() { + let namespace = IggyNamespace::new(0, 0, 0); + let owner = lagging_replica_awaiting_target(namespace); + + owner + .on_partition_state_transfer_target(&group_primary_descriptor(namespace, true)) + .await; + + let partitions = owner.plane.partitions(); + let session = partitions + .get_by_ns(&namespace) + .expect("the partition stays registered") + .transfer + .as_ref(); + assert!( + session.is_some_and(|session| session.target_accepted), + "the group primary's offer holds more committed state than this replica \ + knows, so it must be accepted" + ); +} + +/// Only the primary serves a partition transfer, so a transient refusal from +/// it must be retried against it, not against the replica this node's +/// inflated view would name. Its heartbeats carry a view below this replica's, +/// which the partition view filter drops, yet they still name the primary. +#[compio::test] +async fn given_a_transient_refusal_from_the_primary_when_this_replica_ratcheted_its_view_alone_should_retry_the_primary() + { + let namespace = IggyNamespace::new(0, 0, 0); + let owner = lagging_replica_awaiting_target(namespace); + owner + .on_commit(&primary_commit( + namespace, + GROUP_PRIMARY, + GROUP_VIEW, + GROUP_COMMIT, + )) + .await; + + owner + .on_partition_state_transfer_target(&group_primary_descriptor(namespace, false)) + .await; + + assert_eq!( + scheduled_rearm_peer(&owner, namespace), + GROUP_PRIMARY, + "the next request must go to the primary this replica heard from" + ); +} + +/// A refusal names no primary: without a heartbeat from the refusing peer, +/// staying on it may spend every round on a backup. +#[compio::test] +async fn given_a_transient_refusal_from_an_unheard_peer_when_lagging_should_rotate_away() { + let namespace = IggyNamespace::new(0, 0, 0); + let owner = lagging_replica_awaiting_target(namespace); + + owner + .on_partition_state_transfer_target(&group_primary_descriptor(namespace, false)) + .await; + + assert_ne!(scheduled_rearm_peer(&owner, namespace), GROUP_PRIMARY); +} + +/// A backup can only refuse, so staying on it spends every round on nothing. +#[compio::test] +async fn given_a_transient_refusal_from_a_backup_of_the_group_view_when_lagging_should_rotate_away() +{ + const BACKUP: u8 = 1; + let namespace = IggyNamespace::new(0, 0, 0); + let owner = lagging_replica_awaiting_target(namespace); + assert_ne!(BACKUP, GROUP_PRIMARY); + + owner + .on_partition_state_transfer_target(&lagging_descriptor( + namespace, + BACKUP, + GROUP_VIEW, + DescriptorReply::TransientRefusal, + )) + .await; + + assert_ne!(scheduled_rearm_peer(&owner, namespace), BACKUP); +} + +/// The sender was primary of a view older than the one this replica's log was +/// adopted in, so the group has provably moved past it. +#[compio::test] +async fn given_a_transient_refusal_from_the_primary_of_a_view_below_log_view_when_lagging_should_rotate_away() + { + const STALE_VIEW: u32 = 8; + let namespace = IggyNamespace::new(0, 0, 0); + let owner = lagging_replica_awaiting_target(namespace); + let stale_primary = { + let partitions = owner.plane.partitions(); + let consensus = partitions + .get_by_ns(&namespace) + .expect("the fixture registers the partition") + .consensus(); + assert!(STALE_VIEW < consensus.log_view()); + consensus.primary_index(STALE_VIEW) + }; + + owner + .on_partition_state_transfer_target(&lagging_descriptor( + namespace, + stale_primary, + STALE_VIEW, + DescriptorReply::TransientRefusal, + )) + .await; + + assert_ne!(scheduled_rearm_peer(&owner, namespace), stale_primary); +} + +/// Five replicas, so the primary this replica's inflated view names (3), the +/// group's primary (1) and the failed backup (2) are all distinct. A rotation +/// steered by the inflated view asks replica 3, which can only refuse. +#[compio::test] +async fn given_a_hard_refusal_from_a_backup_when_this_replica_ratcheted_its_view_alone_should_rotate_to_the_group_primary() + { + const GROUP_VIEW_OF_FIVE: u32 = 21; + const FAILED_BACKUP: u8 = 2; + let namespace = IggyNamespace::new(0, 0, 0); + let owner = lagging_replica_in_cluster(namespace, 5); + let group_primary = { + let partitions = owner.plane.partitions(); + let consensus = partitions + .get_by_ns(&namespace) + .expect("the fixture registers the partition") + .consensus(); + let group_primary = consensus.primary_index(GROUP_VIEW_OF_FIVE); + let inflated_primary = consensus.primary_index(LAGGING_VIEW); + assert_eq!( + HashSet::from([0, group_primary, inflated_primary, FAILED_BACKUP]).len(), + 4 + ); + group_primary + }; + owner + .on_commit(&primary_commit( + namespace, + group_primary, + GROUP_VIEW_OF_FIVE, + GROUP_COMMIT, + )) + .await; + owner + .on_partition_state_transfer_target(&lagging_descriptor( + namespace, + group_primary, + GROUP_VIEW_OF_FIVE, + DescriptorReply::TransientRefusal, + )) + .await; + assert_eq!(scheduled_rearm_peer(&owner, namespace), group_primary); + { + let partitions = owner.plane.partitions(); + let partition = partitions + .get_mut_by_ns(&namespace) + .expect("the partition stays registered"); + partition.transfer_rearm = None; + arm_lagging_transfer(partition, FAILED_BACKUP); + } + + owner + .on_partition_state_transfer_target(&lagging_descriptor( + namespace, + FAILED_BACKUP, + GROUP_VIEW_OF_FIVE, + DescriptorReply::HardRefusal, + )) + .await; + + assert_eq!( + scheduled_rearm_peer(&owner, namespace), + group_primary, + "the rotation must follow the primary the group was last heard from" + ); +} + +/// A second lagging replica whose view climbed alone to a view it is primary +/// of refuses transiently, stamping that view. Taken as the primary, it would +/// pin every later re-arm to itself. +#[compio::test] +async fn given_a_lagging_peer_refusing_at_its_inflated_view_when_the_group_primary_was_heard_should_rearm_to_the_group_primary() + { + const GROUP_VIEW_OF_FIVE: u32 = 21; + const INFLATED_PEER: u8 = 4; + const INFLATED_VIEW: u32 = 59; + const FAILED_BACKUP: u8 = 2; + let namespace = IggyNamespace::new(0, 0, 0); + let owner = lagging_replica_in_cluster(namespace, 5); + let group_primary = { + let partitions = owner.plane.partitions(); + let consensus = partitions + .get_by_ns(&namespace) + .expect("the fixture registers the partition") + .consensus(); + assert_eq!(consensus.primary_index(INFLATED_VIEW), INFLATED_PEER); + consensus.primary_index(GROUP_VIEW_OF_FIVE) + }; + owner + .on_commit(&primary_commit( + namespace, + group_primary, + GROUP_VIEW_OF_FIVE, + GROUP_COMMIT, + )) + .await; + rearm_lagging_transfer_to(&owner, namespace, INFLATED_PEER); + + owner + .on_partition_state_transfer_target(&lagging_descriptor( + namespace, + INFLATED_PEER, + INFLATED_VIEW, + DescriptorReply::TransientRefusal, + )) + .await; + assert_eq!(scheduled_rearm_peer(&owner, namespace), group_primary); + + rearm_lagging_transfer_to(&owner, namespace, group_primary); + owner + .on_partition_state_transfer_target(&lagging_descriptor( + namespace, + group_primary, + GROUP_VIEW_OF_FIVE, + DescriptorReply::TransientRefusal, + )) + .await; + assert_eq!(scheduled_rearm_peer(&owner, namespace), group_primary); + + rearm_lagging_transfer_to(&owner, namespace, FAILED_BACKUP); + owner + .on_partition_state_transfer_target(&lagging_descriptor( + namespace, + FAILED_BACKUP, + GROUP_VIEW_OF_FIVE, + DescriptorReply::HardRefusal, + )) + .await; + assert_eq!(scheduled_rearm_peer(&owner, namespace), group_primary); + assert_eq!( + owner + .plane + .partitions() + .get_by_ns(&namespace) + .expect("the partition stays registered") + .consensus() + .primary_hint(), + Some(group_primary) + ); +} + +/// The primary this replica heard from goes silent. Kept as the hint, it +/// would draw every rotation back to itself, so the ring walk would only ever +/// reach the next replica and never the new primary beyond it. +#[compio::test] +async fn given_the_heard_primary_failing_when_rotating_should_walk_the_ring_to_the_new_primary() { + const DEAD_PRIMARY_VIEW: u32 = 21; + const NEW_PRIMARY_VIEW: u32 = 23; + let namespace = IggyNamespace::new(0, 0, 0); + let owner = lagging_replica_in_cluster(namespace, 5); + let (dead_primary, new_primary) = { + let partitions = owner.plane.partitions(); + let consensus = partitions + .get_by_ns(&namespace) + .expect("the fixture registers the partition") + .consensus(); + ( + consensus.primary_index(DEAD_PRIMARY_VIEW), + consensus.primary_index(NEW_PRIMARY_VIEW), + ) + }; + assert_eq!((dead_primary, new_primary), (1, 3)); + owner + .on_commit(&primary_commit( + namespace, + dead_primary, + DEAD_PRIMARY_VIEW, + GROUP_COMMIT, + )) + .await; + + let mut tried = Vec::new(); + let mut peer = dead_primary; + for _ in 0..2 { + rearm_lagging_transfer_to(&owner, namespace, peer); + { + let partitions = owner.plane.partitions(); + let partition = partitions + .get_mut_by_ns(&namespace) + .expect("the partition stays registered"); + owner + .abandon_or_rearm_partition_transfer(partition, peer) + .await; + } + peer = scheduled_rearm_peer(&owner, namespace); + tried.push(peer); + } + + assert_eq!(tried, vec![2, new_primary]); +} + +/// Frames the bus forwarded, by destination replica. +type SentFrames = Rc<RefCell<Vec<(u8, Frozen<MESSAGE_ALIGN>)>>>; + +/// A backup at view 1 of five, behind its own `commit_max`. +fn gap_stopped_backup( + namespace: IggyNamespace, + commit_max: u64, +) -> (CompletionTestShard, SentFrames) { + let bus = Rc::new(IggyMessageBus::new(0)); + let sent = Rc::new(RefCell::new(Vec::new())); + let captured = Rc::clone(&sent); + bus.set_replica_forward_fn(Box::new(move |peer, _, frame| { + captured.borrow_mut().push((peer, frame)); + Ok(()) + })); + for peer in 1..5 { + assert!(bus.owner_table().try_claim(peer, 1)); + } + let config = partitions_config(); + let metadata = IggyMetadata::new(None, None, None, None, PollTestMetadata::default(), None); + let (owner, _sender) = owner_in_cluster(&bus, config.clone(), namespace, metadata, 5); + let mut consensus = VsrConsensus::new(1, 0, 5, namespace.inner(), bus, LocalPipeline::new()); + consensus.init(); + consensus.set_view(1); + consensus.set_log_view(1); + consensus.advance_commit_max(commit_max); + let partition = IggyPartition::with_in_memory_storage( + Arc::new(PartitionStats::default()), + consensus, + config.segment_size, + ); + owner.plane.partitions().insert(namespace, partition); + owner.set_repair_gap_debounce_ticks(1); + owner.set_repair_retry_ticks(1); + (owner, sent) +} + +/// Tick until the partition's repair session targets a peer other than +/// `previous`, and return it. +#[allow(clippy::future_not_send)] +async fn tick_until_repair_peer_changes( + owner: &CompletionTestShard, + namespace: IggyNamespace, + previous: Option<u8>, +) -> u8 { + let mut scratch = Vec::new(); + for _ in 0..64 { + assert!(owner.tick_partitions(&mut scratch).await.is_none()); + scratch.clear(); + let peer = owner + .plane + .partitions() + .get_by_ns(&namespace) + .expect("the partition stays registered") + .repair + .map(|session| session.peer); + if let Some(peer) = peer + && Some(peer) != previous + { + return peer; + } + } + panic!("the repair session never moved off {previous:?}"); +} + +/// The gap-repair arm and the stall rotation both aim at the primary this +/// replica last heard from, which is not the primary of its own view. +#[compio::test] +async fn given_a_primary_heard_at_a_newer_view_when_the_partition_gap_stops_should_repair_from_it() +{ + const HEARD_VIEW: u32 = 3; + let namespace = IggyNamespace::new(0, 0, 0); + let (owner, _sent) = gap_stopped_backup(namespace, 5); + let (own_view_primary, heard_primary) = { + let partitions = owner.plane.partitions(); + let consensus = partitions + .get_by_ns(&namespace) + .expect("the fixture registers the partition") + .consensus(); + ( + consensus.primary_index(consensus.view()), + consensus.primary_index(HEARD_VIEW), + ) + }; + assert_eq!((own_view_primary, heard_primary), (1, 3)); + let heartbeat = primary_commit(namespace, heard_primary, HEARD_VIEW, 5); + owner.on_commit(&heartbeat).await; + + let armed = tick_until_repair_peer_changes(&owner, namespace, None).await; + assert_eq!(armed, heard_primary, "the gap arm asks the heard primary"); + + let rotated = tick_until_repair_peer_changes(&owner, namespace, Some(armed)).await; + assert_eq!( + rotated, own_view_primary, + "a stalled heard primary is forgotten, leaving the primary of this view" + ); + + owner.on_commit(&heartbeat).await; + let back = tick_until_repair_peer_changes(&owner, namespace, Some(rotated)).await; + assert_eq!( + back, heard_primary, + "the stall rotation goes to the primary heard again" + ); +} + +/// A floor holds the walk and the served chunk above it carries no message, so +/// nothing anchors the window. Pulling again from `commit_min + 1` would +/// re-serve that same chunk forever. +#[compio::test] +async fn given_a_floor_holding_the_walk_when_a_chunk_without_messages_lands_should_pull_the_next_chunk() + { + const PEER: u8 = 1; + const NONCE: u128 = 0xf100; + const FLOOR: u64 = 5; + const FIRST_CHUNK_LAST: u64 = FLOOR + REPAIR_CHUNK_MAX; + const TO_OP: u64 = FIRST_CHUNK_LAST + 10; + let namespace = IggyNamespace::new(0, 0, 0); + let (owner, sent) = gap_stopped_backup(namespace, TO_OP); + { + let partitions = owner.plane.partitions(); + let partition = partitions + .get_mut_by_ns(&namespace) + .expect("the fixture registers the partition"); + partition.repair = Some(RepairSession { + nonce: NONCE, + view: partition.consensus().view(), + commit_to_op: TO_OP, + fetch_to_op: TO_OP, + floor: Some(FLOOR), + peer: PEER, + idle_ticks: 0, + floor_pulled_from: 0, + }); + for op in FLOOR + 1..=FIRST_CHUNK_LAST { + partition + .log + .journal() + .inner + .append(control_prepare(namespace, op).into_frozen()) + .await + .expect("journal append"); + } + } + let repair_done = |served_through: u64| { + Message::<RepairRangeReplyHeader>::new(size_of::<RepairRangeReplyHeader>()) + .transmute_header(|_, header: &mut RepairRangeReplyHeader| { + header.command = Command::RepairDone; + header.size = u32::try_from(size_of::<RepairRangeReplyHeader>()).unwrap(); + header.group = namespace.inner(); + header.cluster = 1; + header.replica = PEER; + header.nonce = NONCE; + header.op = served_through; + header.seal(); + }) + }; + + owner + .on_repair_range_reply(&repair_done(FIRST_CHUNK_LAST)) + .await; + + assert_eq!( + drain_prepare_requests(&sent), + vec![(PEER, FIRST_CHUNK_LAST + 1, TO_OP)] + ); + + owner + .on_repair_range_reply(&repair_done(FIRST_CHUNK_LAST)) + .await; + assert!( + sent.borrow().is_empty(), + "a reply that delivered nothing new leaves the rest to the stall retry" + ); + + let mut scratch = Vec::new(); + assert!(owner.tick_partitions(&mut scratch).await.is_none()); + assert_eq!( + drain_prepare_requests(&sent), + vec![(PEER, FIRST_CHUNK_LAST + 1, TO_OP)], + "the stall retry starts past the resident chunk too" + ); +} + +/// A floor holds the walk and every op up to `fetch_to_op` is resident, so the +/// pull start lies past the window. Only a `RepairDone` completes the session. +#[allow(clippy::future_not_send)] +async fn resident_window_backup( + namespace: IggyNamespace, + peer: u8, + nonce: u128, + floor: u64, + to_op: u64, +) -> (CompletionTestShard, SentFrames) { + let (owner, sent) = gap_stopped_backup(namespace, to_op); + { + let partitions = owner.plane.partitions(); + let partition = partitions + .get_mut_by_ns(&namespace) + .expect("the fixture registers the partition"); + partition.repair = Some(RepairSession { + nonce, + view: partition.consensus().view(), + commit_to_op: to_op, + fetch_to_op: to_op, + floor: Some(floor), + peer, + idle_ticks: 0, + floor_pulled_from: 0, + }); + for op in floor + 1..=to_op { + partition + .log + .journal() + .inner + .append(control_prepare(namespace, op).into_frozen()) + .await + .expect("journal append"); + } + } + (owner, sent) +} + +#[compio::test] +async fn given_a_resident_window_when_the_last_repair_done_is_lost_should_re_request_and_complete() +{ + const PEER: u8 = 1; + const NONCE: u128 = 0xf100; + const FLOOR: u64 = 5; + const TO_OP: u64 = FLOOR + 10; + let namespace = IggyNamespace::new(0, 0, 0); + let (owner, sent) = resident_window_backup(namespace, PEER, NONCE, FLOOR, TO_OP).await; + + let mut scratch = Vec::new(); + assert!(owner.tick_partitions(&mut scratch).await.is_none()); + assert_eq!( + drain_prepare_requests(&sent), + vec![(PEER, TO_OP, TO_OP)], + "the stall retry asks again for the last op to draw a fresh RepairDone" + ); + + let repair_done = Message::<RepairRangeReplyHeader>::new(size_of::<RepairRangeReplyHeader>()) + .transmute_header(|_, header: &mut RepairRangeReplyHeader| { + header.command = Command::RepairDone; + header.size = u32::try_from(size_of::<RepairRangeReplyHeader>()).unwrap(); + header.group = namespace.inner(); + header.cluster = 1; + header.replica = PEER; + header.nonce = NONCE; + header.op = TO_OP; + header.seal(); + }); + owner.on_repair_range_reply(&repair_done).await; + + let partitions = owner.plane.partitions(); + let partition = partitions + .get_by_ns(&namespace) + .expect("the partition stays registered"); + assert!( + partition.repair.is_none(), + "the re-served RepairDone completes the session" + ); + assert_eq!(partition.consensus().commit_min(), TO_OP); +} + +#[compio::test] +async fn given_a_resident_window_when_the_serving_peer_is_dead_should_rotate_to_another_peer() { + const PEER: u8 = 1; + let namespace = IggyNamespace::new(0, 0, 0); + let (owner, _sent) = resident_window_backup(namespace, PEER, 0xf100, 5, 15).await; + + let rotated = tick_until_repair_peer_changes(&owner, namespace, Some(PEER)).await; + assert_ne!(rotated, PEER); +} + +/// The `RequestPrepares` frames the bus forwarded, as `(peer, from_op, to_op)`. +fn drain_prepare_requests(sent: &SentFrames) -> Vec<(u8, u64, u64)> { + sent.borrow_mut() + .drain(..) + .filter_map(|(peer, frame)| { + let request = Message::<RequestPreparesHeader>::try_from(Owned::copy_from_slice( + frame.as_slice(), + )) + .ok()?; + (request.header().command == Command::RequestPrepares) + .then(|| (peer, request.header().from_op, request.header().to_op)) + }) + .collect() +} + +/// A bare non-`SendMessages` prepare at `op`, as the commit walk skips it. +fn control_prepare(namespace: IggyNamespace, op: u64) -> Message<PrepareHeader> { + let size = size_of::<PrepareHeader>(); + Message::<PrepareHeader>::new(size).transmute_header(|_, header: &mut PrepareHeader| { + header.command = Command::Prepare; + header.retry_capacity = u32::try_from(consensus::PARTITION_DEDUP_CLIENTS_MAX).unwrap(); + header.client = message_bus::AUTO_COMMIT_CLIENT_ID; + header.session = 1; + header.request = op; + header.op = op; + header.operation = Operation::CreateStream; + header.group = namespace.inner(); + header.size = u32::try_from(size).unwrap(); + }) +} + fn invalid_metadata_descriptor( nonce: u128, generation: u64, @@ -1288,6 +2086,21 @@ fn owner_with_metadata_plane( IggySnapshot, PollTestMetadata, >, +) -> (CompletionTestShard, TaggedSender) { + owner_in_cluster(bus, config, namespace, metadata, 3) +} + +fn owner_in_cluster( + bus: &Rc<IggyMessageBus>, + config: PartitionsConfig, + namespace: IggyNamespace, + metadata: IggyMetadata< + consensus::VsrConsensus<Rc<IggyMessageBus>>, + PrepareJournal, + IggySnapshot, + PollTestMetadata, + >, + replica_count: u8, ) -> (CompletionTestShard, TaggedSender) { let shard_id = ShardId::new(0); let partitions = IggyPartitions::new(shard_id, config); @@ -1306,7 +2119,7 @@ fn owner_with_metadata_plane( 2, None, routes, - PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 3), bus.clone()), + PartitionConsensusConfig::new(1, ReplicaTopology::new(0, replica_count), bus.clone()), None, ShardMetrics::for_shard(), )
