This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/partition-routing in repository https://gitbox.apache.org/repos/asf/iggy.git
commit ee0ebb576847d11622a3c00e8db4f862bdc94f24 Author: Hubert Gruszecki <[email protected]> AuthorDate: Wed Oct 14 14:04:26 2026 +0200 fix(cluster): bound history repair requests Backward proof repair repeatedly fetched the held suffix and could stall after proving a chain whose canonical head still lacked a body. Fetch chunks ending at the proof hole and continue when it changes. Reconcile only frames that can close that hole. Keep the metadata repair ceiling above the adopted head after a retransmit rewinds its sequencer. Cover metadata suffix refresh and same-view proof renewal, with a large-gap repair regression. --- core/consensus/src/impls.rs | 61 +++++++++- core/partitions/src/types.rs | 4 +- core/shard/src/lib.rs | 124 +++++++++++--------- core/shard/src/partition_io.rs | 259 ++++++++++++++++++++++++++++++++++++++--- 4 files changed, 377 insertions(+), 71 deletions(-) diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 8b2df5992..c01b95267 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -5169,7 +5169,8 @@ mod timestamp_clamp_tests { use super::*; use crate::LocalPipeline; - use crate::test_bus::{NoopBus, make_start_view}; + use crate::test_bus::{NoopBus, make_commit, make_start_view}; + use iggy_binary_protocol::CommitHeader; #[test] fn observed_log_timestamp_floors_new_primary_stamps() { @@ -5492,6 +5493,64 @@ mod timestamp_clamp_tests { } } + #[test] + fn given_unproven_same_view_when_commit_and_start_view_arrive_should_refresh_proof() { + const VIEW: u32 = 4; + const HEAD: u64 = 8; + let consensus = VsrConsensus::with_clock( + 1, + 0, + 3, + METADATA_GROUP, + NoopBus, + LocalPipeline::new(), + ConsensusClock::system(), + ); + let primary = u8::try_from(VIEW % 3).unwrap(); + let start = make_start_view(VIEW, HEAD, 0, primary, 0); + consensus.handle_start_view(PlaneKind::Metadata, start.header(), &[]); + consensus.fence_view_log(HEAD - 1, HEAD - 1); + let commit = make_commit(VIEW, primary, 1).transmute_header( + |original, header: &mut CommitHeader| { + *header = original; + header.commit = HEAD; + header.seal(); + }, + ); + assert!(matches!( + consensus.handle_commit(commit.header()), + CommitOutcome::Advanced + )); + assert_eq!(consensus.commit_max(), HEAD); + assert_eq!(consensus.commit_min(), 0); + assert!(consensus.view_log_unproven()); + let mut canonical = PrepareHeader { + command: Command::Prepare, + operation: Operation::CreateStream, + size: u32::try_from(size_of::<PrepareHeader>()).unwrap(), + op: HEAD, + ..PrepareHeader::default() + }; + canonical.checksum = canonical.identity_checksum(); + let mut suffix = vec![0; size_of::<PrepareHeader>()]; + crate::encode_prepare_headers(&[canonical], &mut suffix); + + let actions = consensus.handle_start_view( + PlaneKind::Metadata, + make_start_view(VIEW, HEAD, HEAD, primary, 0).header(), + &suffix, + ); + + assert!(!actions.is_empty()); + assert_eq!( + consensus.pending_view_log().unwrap().headers, + vec![canonical] + ); + assert!(consensus.view_log_unproven()); + assert!(!consensus.view_log_entry_proven(1)); + assert_eq!(consensus.commit_min(), 0); + } + #[test] fn given_unreconciled_start_view_when_head_grows_and_times_out_should_keep_commit_fenced() { const VIEW: u32 = 4; diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs index 554070a8a..b985d7054 100644 --- a/core/partitions/src/types.rs +++ b/core/partitions/src/types.rs @@ -292,8 +292,8 @@ pub struct RepairSession { /// Consensus view in which this session was armed. A later view discards /// the session before any delayed repair body can enter its journal. pub view: u32, - /// Start of the latest request. A proof gap moving below it is progress - /// even while the unproven prefix fences the commit walk. + /// Latest repair cursor: the proof hole for backward chunks, otherwise + /// the first requested op. A changed proof hole is progress despite fenced commits. pub from_op: u64, /// Committed frontier this repair must make locally walkable. Floor /// completeness and session completion are bounded here. diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 3e0d326c9..ab11c0b20 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -1042,7 +1042,7 @@ pub const REPAIR_CHUNK_MAX: u64 = 128; #[derive(Debug, Clone, Copy)] struct MetadataRepairSession { nonce: u128, - /// Lowest op this session must fetch, and the floor its stall retry reopens at. + /// Repair cursor: the backward proof hole, otherwise the lowest requested op. /// /// Not re-derivable from `commit_min + 1`: the merged-log scan opens at /// [`merged_log_scan_floor`], above the snapshot floor, while its @@ -5491,6 +5491,9 @@ where } let is_primary = consensus.is_primary_for_view(consensus.view()); let commit_min = consensus.commit_min(); + let closes_proof_gap = consensus + .view_log_repair_op() + .is_none_or(|op| op == header.op); // In place: runs once per repaired prepare, and the clone is two Vecs. let in_scope = consensus .with_pending_view_log(|pending| { @@ -5565,7 +5568,9 @@ where .is_some_and(|held| held.checksum == header.checksum) { consensus.note_repaired_view_header(header); - self.reconcile_metadata_view_divergence().await; + if closes_proof_gap { + self.reconcile_metadata_view_divergence().await; + } } return; } @@ -5600,7 +5605,9 @@ where consensus.sequencer().set_sequence(frontier); consensus.set_last_prepare_checksum(frontier_checksum); } - self.reconcile_metadata_view_divergence().await; + if closes_proof_gap { + self.reconcile_metadata_view_divergence().await; + } return; } // A metadata-plane op that did not match above (no metadata consensus on @@ -5638,9 +5645,15 @@ where ); return; } + let closes_proof_gap = partition + .consensus() + .view_log_repair_op() + .is_none_or(|op| op == header.op); partition.apply_repaired_prepare(msg).await; - let pending = partition.consensus().pending_view_log(); - reconcile_partition_view_divergence(self.id, partition, pending.as_ref()).await; + if closes_proof_gap { + let pending = partition.consensus().pending_view_log(); + reconcile_partition_view_divergence(self.id, partition, pending.as_ref()).await; + } } /// Repair stream terminator: `RangeEvicted` settles the partition commit @@ -5683,7 +5696,7 @@ where let done = commit_min >= session.to_op; let proof_advanced = consensus .view_log_repair_op() - .is_some_and(|op| op < session.from_op); + .is_some_and(|op| op != session.from_op); tracing::info!( shard = self.id, through_op = header.op, @@ -5707,9 +5720,7 @@ where live.from_op = from_op; } self.send_request_prepares( - consensus.cluster(), - consensus.replica(), - consensus.view(), + consensus, session.peer, session.nonce, from_op, @@ -5897,8 +5908,8 @@ 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). Proof can move backward through - // a gap while fencing commits, so either kind of progress + // delivery on a lossy bus). Proof moves backward through gaps, + // then may need a higher canonical header's body. Either change // pulls the next chunk without waiting for the retry timer. let before = partition.consensus().commit_min(); if let partitions::RepairConclusion::FloorRefused { floor, to_op } = @@ -5963,19 +5974,15 @@ where let commit_min = partition.consensus().commit_min(); let proof_op = partition.consensus().view_log_repair_op(); let next = partition.repair.as_mut().and_then(|live| { - let from_op = proof_op.filter(|&op| op < live.from_op).or_else(|| { + let from_op = proof_op.filter(|&op| op != live.from_op).or_else(|| { partition_repair_next_chunk(before, commit_min, live.fetch_to_op) })?; live.from_op = from_op; Some((live.peer, live.nonce, from_op, live.fetch_to_op)) }); - let cluster = partition.consensus().cluster(); - let self_id = partition.consensus().replica(); if let Some((peer, nonce, from_op, to_op)) = next { self.send_request_prepares( - cluster, - self_id, - partition.consensus().view(), + partition.consensus(), peer, nonce, from_op, @@ -5997,11 +6004,9 @@ where /// is no longer retained). #[allow(clippy::too_many_arguments)] #[allow(clippy::future_not_send, clippy::cast_possible_truncation)] - async fn send_request_prepares( + async fn send_request_prepares<P>( &self, - cluster: u128, - self_id: u8, - view: u32, + consensus: &VsrConsensus<B, P>, target: u8, nonce: u128, from_op: u64, @@ -6010,13 +6015,32 @@ where partition_incarnation: u64, ) where B: MessageBus, + P: Pipeline<Entry = consensus::PipelineEntry>, { + // Proof walks backward. End the chunk at its hole instead of re-serving + // the already held suffix for every missing predecessor. + let (from_op, to_op) = consensus + .view_log_repair_op() + .map_or((from_op, to_op), |missing| { + if consensus.view_log_entry_proven(missing) { + return (missing, to_op.max(missing)); + } + ( + missing + .saturating_sub(self.repair_chunk_max.get() - 1) + .max(consensus.commit_min() + 1), + missing, + ) + }); + if from_op == 0 || from_op > to_op { + return; + } let msg = Message::<RequestPreparesHeader>::new(size_of::<RequestPreparesHeader>()) .transmute_header(|_, h: &mut RequestPreparesHeader| { h.command = Command::RequestPrepares; - h.cluster = cluster; - h.replica = self_id; - h.view = view; + h.cluster = consensus.cluster(); + h.replica = consensus.replica(); + h.view = consensus.view(); h.nonce = nonce; h.from_op = from_op; h.to_op = to_op; @@ -6323,9 +6347,7 @@ where "metadata repair stalled; re-requesting remaining window" ); self.send_request_prepares( - consensus.cluster(), - consensus.replica(), - consensus.view(), + consensus, peer, nonce, from_op, @@ -6662,9 +6684,7 @@ where idle_ticks: 0, }); self.send_request_prepares( - consensus.cluster(), - consensus.replica(), - consensus.view(), + consensus, peer, nonce, from_op, @@ -6721,14 +6741,22 @@ where && (consensus.commit_min() < consensus.commit_max() || consensus.view_log_unproven()) && self.metadata_repair.borrow().is_none() { + let from_op = consensus + .view_log_repair_op() + .unwrap_or_else(|| consensus.commit_min() + 1); let to_op = if consensus.view_log_unproven() { - consensus.sequencer().current_sequence() + consensus + .sequencer() + .current_sequence() + .max( + consensus + .with_pending_view_log(|pending| pending.op_head) + .unwrap_or(0), + ) + .max(from_op) } else { consensus.commit_max() }; - let from_op = consensus - .view_log_repair_op() - .unwrap_or_else(|| consensus.commit_min() + 1); // Spent here rather than at the detector, so the edge-triggered // sites spend it too: an edge-armed repair that completes before // the next tick would otherwise leave the count saturated and hand @@ -8025,8 +8053,6 @@ where } let consensus_view = partition.consensus().view(); let commit_min = partition.consensus().commit_min(); - let cluster = partition.consensus().cluster(); - let self_id = partition.consensus().replica(); // A primary-elect's merged-log session legitimately runs outside // `Normal` (`request_partition_view_repair`). Same predicate as the // two ingest sites, so the arming side and the ingest side cannot @@ -8084,8 +8110,6 @@ where session.nonce, repair_from, session.fetch_to_op, - cluster, - self_id, )) }); // A session pins its peer and fences every arming site while it @@ -8111,7 +8135,7 @@ where ) }) }; - if let Some(((peer, nonce, from_op, to_op, cluster, self_id), rotate)) = stalled + if let Some(((peer, nonce, from_op, to_op), rotate)) = stalled && from_op <= to_op { if rotate @@ -8150,8 +8174,12 @@ where }; let consensus = partition.consensus(); let primary = consensus.primary_index(consensus.view()); - let next_peer = - next_transfer_peer(self_id, peer, consensus.replica_count(), primary); + let next_peer = next_transfer_peer( + consensus.replica(), + peer, + consensus.replica_count(), + primary, + ); tracing::warn!( shard = self.id, namespace_raw = namespace.inner(), @@ -8211,9 +8239,7 @@ where "partition repair stalled; re-requesting remaining window" ); self.send_request_prepares( - cluster, - self_id, - partition.consensus().view(), + partition.consensus(), peer, nonce, from_op, @@ -9314,9 +9340,7 @@ where "partition behind the group frontier; requesting repair" ); self.send_request_prepares( - partition.consensus().cluster(), - partition.consensus().replica(), - partition.consensus().view(), + partition.consensus(), peer, nonce, from_op, @@ -9373,8 +9397,6 @@ where return; }; let nonce = iggy_common::random_id::get_uuid(); - let cluster = consensus.cluster(); - let self_id = consensus.replica(); let namespace = consensus.group(); let view = consensus.view(); self.partition_repairs_inflight @@ -9401,9 +9423,7 @@ where "repairing toward the merged partition log before starting the view" ); self.send_request_prepares( - cluster, - self_id, - partition.consensus().view(), + partition.consensus(), peer, nonce, from_op, diff --git a/core/shard/src/partition_io.rs b/core/shard/src/partition_io.rs index 6c3d08ca6..592091d6a 100644 --- a/core/shard/src/partition_io.rs +++ b/core/shard/src/partition_io.rs @@ -790,7 +790,7 @@ mod tests { use std::time::Duration; use consensus::{ - ArtifactProgress, Consensus, DvcSuffix, LocalPipeline, PartitionsHandle, Pipeline, + ArtifactProgress, Consensus, DvcSuffix, LocalPipeline, PartitionsHandle, Pipeline, Plane, Sequencer, StateArtifact, StateTransferStage, VsrAction, VsrConsensus, artifact_kind, }; use futures::FutureExt; @@ -798,12 +798,14 @@ mod tests { use iggy_binary_protocol::namespace::METADATA_GROUP; use iggy_binary_protocol::primitives::consumer::WireConsumer; use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; + use iggy_binary_protocol::requests::streams::CreateStreamRequest; use iggy_binary_protocol::{ AckLevel, Command, CommitHeader, ConsensusHeader, DoViewChangeHeader, GenericHeader, Operation, PrepareHeader, PrepareOkHeader, RepairRangeReplyHeader, ReplyHeader, RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader, RoutedRequestHeader, StartViewChangeHeader, StartViewHeader, - StateChunkHeader, StateTransferTargetHeader, WireEncode, WireIdentifier, + StateChunkHeader, StateTransferTargetHeader, WireEncode, WireIdentifier, WireName, + WireOptions, }; use iggy_common::{ ConsumerGroupOffsets, ConsumerKind, ConsumerOffsets, Durability, IggyByteSize, @@ -996,9 +998,7 @@ mod tests { owner.send_request_state_transfer(&consensus, PEER, 1).await; owner .send_request_prepares( - consensus.cluster(), - consensus.replica(), - consensus.view(), + &consensus, PEER, 1, 1, @@ -4171,6 +4171,208 @@ mod tests { type IoTestShard<B = Rc<IggyMessageBus>> = IggyShard<B, PrepareJournal, (), IoTestMetadata, PapayaShardsTable, HeldSuperblock>; + #[compio::test] + async fn given_metadata_suffix_mutation_when_stale_commit_arrives_should_publish_fresh_headers() + { + const VIEW: u32 = 3; + const HEAD: u64 = 2; + let directory = tempfile::tempdir().unwrap(); + 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 |_, _, frame| { + captured.borrow_mut().push(frame); + Ok(()) + })); + for replica in [1, 2] { + assert!(bus.owner_table().try_claim(replica, 1)); + } + let (mut owner, _sender) = test_owner(&bus, None); + let mut consensus = VsrConsensus::new( + 1, + 0, + 3, + METADATA_GROUP, + Rc::clone(&bus), + LocalPipeline::new(), + ); + consensus.set_view(VIEW); + consensus.set_log_view(VIEW); + consensus.init(); + consensus.mark_superblock_durable(VIEW, VIEW); + let journal = PrepareJournal::open(&directory.path().join("metadata.wal"), 0) + .await + .unwrap(); + owner.plane.inner_mut().0.consensus = Some(consensus); + owner.plane.inner_mut().0.journal = Some(journal); + let metadata = &owner.plane.inner().0; + let consensus = metadata.consensus.as_ref().unwrap(); + crate::refresh_metadata_dvc_suffix(consensus, metadata.journal.as_ref()); + let prepares = metadata_prepares(HEAD, VIEW); + for prepare in &prepares { + metadata + .journal + .as_ref() + .unwrap() + .append(prepare.clone()) + .await + .unwrap(); + consensus.note_journal_mutation(); + } + consensus.sequencer().set_sequence(HEAD); + assert!(consensus.local_dvc_suffix_stale()); + let heartbeat = Message::<CommitHeader>::new(size_of::<CommitHeader>()).transmute_header( + |_, header: &mut CommitHeader| { + header.command = Command::Commit; + header.cluster = 1; + header.group = METADATA_GROUP; + header.replica = 2; + header.view = VIEW - 1; + header.size = u32::try_from(size_of::<CommitHeader>()).unwrap(); + header.seal(); + }, + ); + + owner.on_commit(&heartbeat).await; + + assert_eq!(sent.borrow().len(), 2); + let expected: Vec<_> = prepares + .iter() + .rev() + .map(|prepare| *prepare.header()) + .collect(); + for frame in sent.borrow().iter() { + let reply = + Message::<StartViewHeader>::try_from(Owned::copy_from_slice(frame.as_slice())) + .unwrap(); + let suffix = consensus::dvc_suffix_decode( + &reply.as_slice()[size_of::<StartViewHeader>()..], + HEAD, + 0, + 0, + ) + .unwrap(); + assert_eq!(suffix.headers(), expected); + } + } + + #[compio::test] + async fn given_metadata_retransmit_below_adopted_head_when_repairing_should_keep_a_valid_range() + { + const LOCAL_HEAD: u64 = 4; + const HEAD: u64 = 8; + const VIEW: u32 = 3; + let directory = tempfile::tempdir().unwrap(); + let bus = Rc::new(IggyMessageBus::new(0)); + let requests = Rc::new(RefCell::new(Vec::new())); + let captured = Rc::clone(&requests); + 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::RequestPrepares { + captured.borrow_mut().push( + Message::<RequestPreparesHeader>::try_from(Owned::copy_from_slice( + frame.as_slice(), + )) + .unwrap(), + ); + } + Ok(()) + })); + assert!(bus.owner_table().try_claim(0, 1)); + let (mut owner, _sender) = test_owner(&bus, None); + let mut consensus = VsrConsensus::new( + 1, + 2, + 3, + METADATA_GROUP, + Rc::clone(&bus), + LocalPipeline::new(), + ); + consensus.set_view(1); + consensus.set_log_view(1); + consensus.init(); + let journal = PrepareJournal::open(&directory.path().join("metadata.wal"), 0) + .await + .unwrap(); + let prepares = metadata_prepares(HEAD, 1); + for prepare in prepares.iter().take(usize::try_from(LOCAL_HEAD).unwrap()) { + journal.append(prepare.clone()).await.unwrap(); + } + consensus.sequencer().set_sequence(LOCAL_HEAD); + owner.plane.inner_mut().0.consensus = Some(consensus); + owner.plane.inner_mut().0.journal = Some(journal); + owner + .on_start_view(start_view_suffix( + IggyNamespace::from_raw(METADATA_GROUP), + VIEW, + HEAD, + 0, + &[*prepares.last().unwrap().header()], + )) + .await; + let metadata = &owner.plane.inner().0; + let consensus = metadata.consensus.as_ref().unwrap(); + assert!(consensus.view_log_unproven()); + assert_eq!(consensus.view_log_repair_op(), Some(HEAD - 1)); + let retransmit = prepares[usize::try_from(LOCAL_HEAD).unwrap()] + .clone() + .transmute_header(|original, header: &mut PrepareHeader| { + *header = original; + header.view = VIEW; + header.replica = 0; + header.checksum = header.identity_checksum(); + }); + metadata.on_replicate(retransmit).await; + assert_eq!(consensus.sequencer().current_sequence(), LOCAL_HEAD + 1); + assert_eq!(consensus.commit_min(), 0); + *owner.metadata_repair.borrow_mut() = None; + requests.borrow_mut().clear(); + + owner.maybe_request_metadata_repair(consensus, 0).await; + + let requests = requests.borrow(); + assert_eq!(requests.len(), 1); + assert_eq!( + (requests[0].header().from_op, requests[0].header().to_op), + (1, HEAD - 1) + ); + assert_eq!(owner.metadata_repair.borrow().unwrap().to_op, HEAD); + } + + fn metadata_prepares(head: u64, view: u32) -> Vec<Message<PrepareHeader>> { + let mut parent = 0; + (1..=head) + .map(|op| { + let body = CreateStreamRequest { + name: WireName::new(format!("repaired-stream-{op}")).unwrap(), + options: WireOptions::empty(), + } + .to_bytes(); + let size = size_of::<PrepareHeader>() + body.len(); + let mut prepare = Message::<PrepareHeader>::new(size); + prepare.as_mut_slice()[size_of::<PrepareHeader>()..].copy_from_slice(&body); + let prepare = prepare.transmute_header(|_, header: &mut PrepareHeader| { + header.command = Command::Prepare; + header.operation = Operation::CreateStream; + header.cluster = 1; + header.group = METADATA_GROUP; + header.view = view; + header.replica = (view % 3) as u8; + header.op = op; + header.parent = parent; + header.timestamp = op; + header.size = u32::try_from(size).unwrap(); + header.checksum_body = u128::from(iggy_common::calculate_checksum(&body)); + header.checksum = header.identity_checksum(); + }); + parent = prepare.header().checksum; + prepare + }) + .collect() + } + #[compio::test] async fn given_a_stale_primary_heartbeat_when_the_suffix_cache_is_stale_should_reply_with_fresh_headers() { @@ -4249,7 +4451,8 @@ mod tests { async fn given_unproven_suffix_when_adopting_start_view_should_repair_before_applying() { const APPLIED: u64 = 2; const LOCAL_HEAD: u64 = 4; - const NEW_HEAD: u64 = 8; + const NEW_HEAD: u64 = 2050; + const CHUNK: u64 = crate::REPAIR_CHUNK_MAX; const NEW_VIEW: u32 = 3; const NEW_PRIMARY: u8 = 0; let bus = Rc::new(IggyMessageBus::new(0)); @@ -4301,14 +4504,25 @@ mod tests { assert!(partition.log.journal().inner.holds_op(APPLIED + 1)); assert!(partition.consensus().view_log_unproven()); assert!(!partition.consensus().view_log_entry_proven(APPLIED + 1)); - for expected_from in (LOCAL_HEAD + 1..NEW_HEAD).rev() { - let request = requests - .borrow_mut() - .pop() - .expect("proof progress must request the next gap without a retry tick"); + let mut rounds = 0; + let mut bodies = 0; + while partition.consensus().view_log_unproven() { + let request = requests.borrow_mut().pop().unwrap_or_else(|| { + panic!("missing repair request after {rounds} rounds; hole={:?}, session={:?}, transfer={:?}", + partition.consensus().view_log_repair_op(), partition.repair, + partition.consensus().state_transfer_stage()); + }); let request = request.header(); + let missing = partition.consensus().view_log_repair_op().unwrap(); + let expected_from = if partition.consensus().view_log_entry_proven(missing) { + missing + } else { + missing.saturating_sub(CHUNK - 1).max(APPLIED + 1) + }; assert_eq!(request.from_op, expected_from); - assert_eq!(request.to_op, NEW_HEAD); + assert_eq!(request.to_op, missing); + rounds += 1; + assert!(rounds <= (NEW_HEAD - LOCAL_HEAD).div_ceil(CHUNK) + 1); let reply = Message::<RepairRangeReplyHeader>::new(size_of::<RepairRangeReplyHeader>()) .transmute_header(|_, header: &mut RepairRangeReplyHeader| { header.command = Command::RepairDone; @@ -4326,6 +4540,7 @@ mod tests { "a terminator without bodies must wait for the retry timer" ); for op in request.from_op..=request.to_op { + bodies += 1; let repaired = prepares[usize::try_from(op - 1).unwrap()] .clone() .transmute_header( @@ -4354,14 +4569,23 @@ mod tests { "a different peer cannot complete or advance this repair session" ); } + if !partition.consensus().view_log_unproven() { + let partition = owner.plane.partitions().get_mut_by_ns(&namespace).unwrap(); + while partition.consensus().commit_min() < NEW_HEAD { + let before = partition.consensus().commit_min(); + partition + .commit_journal(owner.plane.partitions().config()) + .await; + assert!(partition.consensus().commit_min() > before); + } + } owner.on_repair_range_reply(&reply).await; } + assert_eq!(rounds, (NEW_HEAD - LOCAL_HEAD).div_ceil(CHUNK) + 1); + assert_eq!(bodies, NEW_HEAD - APPLIED); assert!(requests.borrow().is_empty()); - let partition = owner.plane.partitions().get_mut_by_ns(&namespace).unwrap(); + let partition = owner.plane.partitions().get_by_ns(&namespace).unwrap(); assert!(!partition.consensus().view_log_unproven()); - partition - .commit_journal(owner.plane.partitions().config()) - .await; assert_eq!(partition.consensus().commit_min(), NEW_HEAD); } @@ -5038,6 +5262,9 @@ mod tests { let mut prepare = Message::<PrepareHeader>::new(frozen.len()); prepare.as_mut_slice().copy_from_slice(frozen.as_slice()); prepares.push(prepare); + // Capture historical bodies beyond one live acknowledgment window. + origin.consensus().clear_pipeline(); + origin.consensus().drain_loopback_into(&mut Vec::new()); } prepares }
