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 b048d026345418e76365c9d98acacee9591295de Author: Hubert Gruszecki <[email protected]> AuthorDate: Wed Oct 14 13:20:13 2026 +0200 fix(consensus): fence repair history by view ViewChange donors could serve abandoned prepares to backups, while blank canonical slots triggered repeated removal of valid history. Restrict suffix serving by view and role, bind terminators to their peer, and keep blank-slot proof pending until repair supplies evidence. --- core/consensus/src/impls.rs | 21 +++-- core/consensus/src/plane_helpers.rs | 91 +++++++++++++++++++- core/partitions/src/iggy_partition.rs | 15 ++-- core/server/src/partition_reconciler.rs | 8 +- core/shard/src/lib.rs | 96 +++++++++++++++------ core/shard/src/partition_io.rs | 147 ++++++++++++++++++++++++++++++++ 6 files changed, 332 insertions(+), 46 deletions(-) diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 5a51303ae..8b2df5992 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -18,11 +18,11 @@ use crate::oneshot::{self, Receiver, Sender}; use crate::vsr_timeout::{TimeoutKind, TimeoutManager}; use crate::{ - AckLogEvent, Consensus, ControlActionLogEvent, DvcQuorumArray, DvcSuffix, IgnoreReason, - MergeOutcome, MergeQuorums, MergedLog, Pipeline, PlaneKind, PrepareLogEvent, Project, - ReplicaLogContext, SimEventKind, StoredDvc, ViewChangeLogEvent, ViewChangeReason, VsrState, - dvc_count, dvc_iter, dvc_quorum_array_empty, dvc_record, dvc_reset, dvc_suffix_decode, - emit_replica_event, emit_sim_event, merge_dvc_quorum, seal_prepare_checksum, + AckLogEvent, Consensus, ControlActionLogEvent, DvcHeaderKind, DvcQuorumArray, DvcSuffix, + IgnoreReason, MergeOutcome, MergeQuorums, MergedLog, Pipeline, PlaneKind, PrepareLogEvent, + Project, ReplicaLogContext, SimEventKind, StoredDvc, ViewChangeLogEvent, ViewChangeReason, + VsrState, dvc_count, dvc_header_kind, dvc_iter, dvc_quorum_array_empty, dvc_record, dvc_reset, + dvc_suffix_decode, emit_replica_event, emit_sim_event, merge_dvc_quorum, seal_prepare_checksum, }; use bit_set::BitSet; use clock::{Clock, IggySystemClock}; @@ -4046,10 +4046,17 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> { pub fn note_repaired_view_header(&self, header: PrepareHeader) { if let Some(pending) = self.pending_view_log.borrow_mut().as_mut() - && pending.headers.is_empty() && pending.op_head == header.op + && pending + .headers + .first() + .is_none_or(|head| dvc_header_kind(head) == DvcHeaderKind::Blank) { - pending.headers.push(header); + if let Some(head) = pending.headers.first_mut() { + *head = header; + } else { + pending.headers.push(header); + } self.invalidate_local_dvc_suffix(); } } diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs index d541aca5e..006da97a6 100644 --- a/core/consensus/src/plane_helpers.rs +++ b/core/consensus/src/plane_helpers.rs @@ -16,8 +16,8 @@ // under the License. use crate::{ - Consensus, IgnoreReason, Pipeline, PipelineEntry, PlaneKind, PrepareOkOutcome, Sequencer, - Status, VsrConsensus, + Consensus, DvcHeaderKind, IgnoreReason, Pipeline, PipelineEntry, PlaneKind, PrepareOkOutcome, + Sequencer, Status, VsrConsensus, dvc_header_kind, }; use iggy_binary_protocol::consensus::{REJECTION_SECTION_LEN, write_rejection_section}; use iggy_binary_protocol::{ @@ -927,7 +927,10 @@ pub fn view_log_proof( if op_head <= proven { return ViewLogProof::Proven; } - let Some(head) = canonical.first().filter(|head| head.op == op_head) else { + let Some(head) = canonical + .first() + .filter(|head| head.op == op_head && dvc_header_kind(head) != DvcHeaderKind::Blank) + else { return ViewLogProof::Missing { op: op_head, unproven_through: op_head, @@ -938,7 +941,10 @@ pub fn view_log_proof( for op in (proven + 1..=op_head).rev() { let header = if let Some(local) = header_at(op) { local - } else if let Some(header) = canonical.iter().find(|header| header.op == op) { + } else if let Some(header) = canonical + .iter() + .find(|header| header.op == op && dvc_header_kind(header) != DvcHeaderKind::Blank) + { missing.get_or_insert(op); *header } else { @@ -2655,4 +2661,81 @@ mod tests { ); assert_eq!(proof(6, &local, &[]), ViewLogProof::Proven); } + + #[test] + fn given_blank_view_slots_when_proving_history_should_wait_for_evidence() { + const HEAD: u64 = 4; + let blank = crate::dvc_blank(HEAD); + let local = PrepareHeader { + op: HEAD, + parent: 3, + checksum: 4, + ..Default::default() + }; + for held in [None, Some(local)] { + assert_eq!( + view_log_proof(HEAD - 1, &[blank], HEAD, |_| held), + ViewLogProof::Missing { + op: HEAD, + unproven_through: HEAD, + }, + "a blank head cannot prove either a missing or held prepare divergent" + ); + } + assert_eq!( + view_log_proof(HEAD - 2, &[local, crate::dvc_blank(HEAD - 1)], HEAD, |op| { + (op == HEAD).then_some(local) + }), + ViewLogProof::Missing { + op: HEAD - 1, + unproven_through: HEAD - 1, + }, + "a blank predecessor cannot bridge a proof gap" + ); + } + + #[test] + fn given_commit_stamps_when_proving_history_should_require_a_sealed_contiguous_chain() { + const FLOOR: u64 = 2; + let headers: Vec<_> = (FLOOR + 1..=6) + .map(|op| PrepareHeader { + op, + parent: u128::from(op - 1), + checksum: u128::from(op), + commit: op - 1, + ..Default::default() + }) + .collect(); + let proof = |headers: &[PrepareHeader]| { + view_log_proof(FLOOR, &[], 5, |op| { + headers.iter().find(|header| header.op == op).copied() + }) + }; + assert_eq!(proof(&headers), ViewLogProof::Proven); + for damaged in [0, 1] { + let mut broken = headers.clone(); + if damaged == 0 { + broken[1].checksum = CHECKSUM_UNSEALED; + } else { + broken[1].parent += 1; + } + assert_eq!( + proof(&broken), + ViewLogProof::Missing { + op: 5, + unproven_through: 5 + }, + "a later commit stamp must not bridge an unsealed or disconnected prefix" + ); + } + let mut gap = headers; + gap.remove(1); + assert_eq!( + proof(&gap), + ViewLogProof::Missing { + op: 5, + unproven_through: 5 + } + ); + } } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 0655c760b..9ad41d2ef 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -43,11 +43,11 @@ use consensus::Pipeline; use consensus::client_table::{CachedReply, ClientTableWireError}; use consensus::{ AutoCommitRequestContext, ClientTable, ClientTableMode, CommitLogEvent, Consensus, - LocalRequestOrder, PartitionDiagEvent, PipelineEntry, PlaneKind, Project, ReplicaLogContext, - RequestLogEvent, Sequencer, SimEventKind, VsrConsensus, ack_preflight, ack_quorum_reached, - build_deny_reply_from_request, build_reply_from_request, build_reply_message, - emit_namespace_progress_event, emit_partition_diag, emit_sim_event, - fence_old_prepare_by_commit, repair_session_live, repaired_frontier_update, + DvcHeaderKind, LocalRequestOrder, PartitionDiagEvent, PipelineEntry, PlaneKind, Project, + ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind, VsrConsensus, ack_preflight, + ack_quorum_reached, build_deny_reply_from_request, build_reply_from_request, + build_reply_message, dvc_header_kind, emit_namespace_progress_event, emit_partition_diag, + emit_sim_event, fence_old_prepare_by_commit, repair_session_live, repaired_frontier_update, replicate_frozen_to_next_in_chain, replicate_preflight, restamp_prepare_view, send_prepare_ok as send_prepare_ok_common, verify_prepare_integrity, }; @@ -11643,7 +11643,10 @@ where pending .headers .iter() - .find(|expected| expected.op == header.op) + .find(|expected| { + expected.op == header.op + && dvc_header_kind(expected) != DvcHeaderKind::Blank + }) .map(|expected| expected.checksum) }) .flatten(); diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index 0ab61c05c..29793c658 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -2016,10 +2016,11 @@ mod tests { /// Build a partition-plane `RepairRangeReply` as the serving peer would /// send it. Only the fields the receive path reads are stamped: routing - /// (`group`, `partition_incarnation`), session (`nonce`), and verdict. + /// (`group`, `partition_incarnation`), session (`peer`, `nonce`), and verdict. fn build_repair_range_reply( namespace: IggyNamespace, partition_incarnation: u64, + peer: u8, command: Command, nonce: u128, op: u64, @@ -2032,6 +2033,7 @@ mod tests { .expect("zeroed bytes form a valid RepairRangeReplyHeader"); header.command = command; header.size = u32::try_from(header_size).expect("header size fits u32"); + header.replica = peer; header.nonce = nonce; header.op = op; header.group = namespace.inner(); @@ -3546,6 +3548,7 @@ mod tests { #[compio::test] async fn repair_done_over_a_passed_floor_closes_the_session() { const NONCE: u128 = 13; + const PEER: u8 = 1; let tmp = TempDir::new().expect("tempdir for system path"); let config = test_config(&tmp); let mux = TestMux::default(); @@ -3577,7 +3580,7 @@ mod tests { commit_to_op: 8, fetch_to_op: 8, floor: None, - peer: 1, + peer: PEER, idle_ticks: 0, }); for op in 1..=8 { @@ -3637,6 +3640,7 @@ mod tests { .on_message(build_repair_range_reply( ns, incarnation, + PEER, Command::RepairDone, NONCE, 8, diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index b7c340c43..3e0d326c9 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -5227,9 +5227,8 @@ where if let Some(ref consensus) = planes.0.consensus && consensus.group() == header.group { - // Served in `ViewChange` too: the replicas holding a missing body are - // exactly those in `ViewChange`, so refusing would deadlock the repair - // the new primary waits on. Read-only; the requester decides. + // A primary-elect can need bodies held only by ViewChange peers. + // Other requesters may receive only those peers' committed prefix. if consensus.is_transferring() { // Transfer can replace these entries. return; @@ -5243,11 +5242,7 @@ where let journal = journal.handle(); let cluster = consensus.cluster(); let self_id = consensus.replica(); - let to_op = repair_serve_ceiling( - header.to_op, - consensus.commit_max(), - consensus.sequencer().current_sequence(), - ); + let to_op = repair_serve_ceiling(&header, consensus); // Skip the compacted prefix (below the snapshot floor) in one // RangeEvicted notice, then serve contiguously until the range // ends or the WAL runs out. @@ -5367,11 +5362,7 @@ where // until the body is journaled, and the primary's retransmit is dropped // by the backup gap check (adoption already advanced its sequencer to // the head), so repair is the only channel that can deliver them. - let to_op = repair_serve_ceiling( - header.to_op, - partition.consensus().commit_max(), - partition.consensus().sequencer().current_sequence(), - ); + let to_op = repair_serve_ceiling(&header, partition.consensus()); // `None` means the journal holds NOTHING, not "nothing was evicted": // the partition journal is memory-only and `clear_all` wipes the // evicted ring with it, so a freshly installed or freshly restarted @@ -5524,7 +5515,10 @@ where .headers .iter() .chain(pending.committed_elsewhere.iter()) - .find(|expected| expected.op == header.op) + .find(|expected| { + expected.op == header.op + && dvc_header_kind(expected) != DvcHeaderKind::Blank + }) .is_some_and(|expected| expected.checksum != header.checksum) }) .unwrap_or(false); @@ -5670,7 +5664,7 @@ where let Some(session) = session else { return; }; - if header.nonce != session.nonce { + if (header.replica, header.nonce) != (session.peer, session.nonce) { return; } match header.command { @@ -5697,7 +5691,7 @@ where done, "metadata journal repair walked" ); - // The one attributable credit: fenced on `session.nonce` + // The one attributable credit: fenced on nonce and peer // above, so it came from the targeted peer, and the walk moved. // Gating on the walk and not the frame keeps a peer that // terminates every round while serving nothing useful from @@ -5715,6 +5709,7 @@ where self.send_request_prepares( consensus.cluster(), consensus.replica(), + consensus.view(), session.peer, session.nonce, from_op, @@ -5884,7 +5879,7 @@ where let Some(session) = partition.repair else { return; }; - if header.nonce != session.nonce { + if (header.replica, header.nonce) != (session.peer, session.nonce) { return; } // Twin of the `apply_repaired_prepare` gate: a primary-elect's merged-log @@ -5980,6 +5975,7 @@ where self.send_request_prepares( cluster, self_id, + partition.consensus().view(), peer, nonce, from_op, @@ -6005,6 +6001,7 @@ where &self, cluster: u128, self_id: u8, + view: u32, target: u8, nonce: u128, from_op: u64, @@ -6019,6 +6016,7 @@ where h.command = Command::RequestPrepares; h.cluster = cluster; h.replica = self_id; + h.view = view; h.nonce = nonce; h.from_op = from_op; h.to_op = to_op; @@ -6327,6 +6325,7 @@ where self.send_request_prepares( consensus.cluster(), consensus.replica(), + consensus.view(), peer, nonce, from_op, @@ -6665,6 +6664,7 @@ where self.send_request_prepares( consensus.cluster(), consensus.replica(), + consensus.view(), peer, nonce, from_op, @@ -8213,6 +8213,7 @@ where self.send_request_prepares( cluster, self_id, + partition.consensus().view(), peer, nonce, from_op, @@ -9183,7 +9184,7 @@ where /// and nothing else is recovering it. The idempotence guards mirror /// `maybe_request_metadata_repair`: no-op unless Normal, not /// transferring, behind the frontier, and no session live. - #[allow(clippy::future_not_send)] + #[allow(clippy::future_not_send, clippy::too_many_lines)] /// Open a journal-repair session against `peer`, if this partition needs /// one and the shard has room for it. `true` when a session was recorded. /// @@ -9315,6 +9316,7 @@ where self.send_request_prepares( partition.consensus().cluster(), partition.consensus().replica(), + partition.consensus().view(), peer, nonce, from_op, @@ -9401,6 +9403,7 @@ where self.send_request_prepares( cluster, self_id, + partition.consensus().view(), peer, nonce, from_op, @@ -11006,16 +11009,33 @@ fn repair_op_in_scope( /// Ceiling on the op range a repair request may ask this replica to walk. /// -/// Not `commit_max` alone: a new primary repairing toward a merged log needs the -/// uncommitted suffix the view change kept, which sits above every commit point. +/// Only peers in this view may receive an uncommitted suffix. During view +/// change only its primary may receive it to repair its parked merged log. +/// An ordinary backup can accept an unnamed op as committed history. /// /// Bounded by the local frontier all the same. `RequestPreparesHeader::validate` /// accepts any `from_op <= to_op`, so `u64::MAX` is legal, and the metadata serve /// path then walks op by op with no `.await` -- on a single-threaded shard pump /// that ends the shard rather than merely serving slowly. Nothing above the /// frontier is servable, so the clamp costs nothing. -fn repair_serve_ceiling(requested_to_op: u64, commit_max: u64, head: u64) -> u64 { - requested_to_op.min(commit_max.max(head)) +fn repair_serve_ceiling<B, P>( + request: &RequestPreparesHeader, + consensus: &VsrConsensus<B, P>, +) -> u64 +where + B: MessageBus, + P: Pipeline<Entry = consensus::PipelineEntry>, +{ + let suffix_allowed = request.view == consensus.view() + && (consensus.is_normal() || request.replica == consensus.primary_index(consensus.view())); + let ceiling = if suffix_allowed { + consensus + .commit_max() + .max(consensus.sequencer().current_sequence()) + } else { + consensus.commit_min() + }; + request.to_op.min(ceiling) } /// Repair sessions the partition tick sweep will OPEN per pass. @@ -12244,7 +12264,10 @@ mod persist_gate_tests { mod repair_scope_tests { //! Who parked the log decides what it means. - use iggy_binary_protocol::{Command, PrepareHeader}; + use consensus::{LocalPipeline, Sequencer, VsrConsensus}; + use iggy_binary_protocol::{Command, PrepareHeader, RequestPreparesHeader}; + use message_bus::IggyMessageBus; + use server_common::Message; use super::{ MergedLog, adopted_suffix_head, pending_covers_suffix, repair_op_in_scope, @@ -12310,15 +12333,34 @@ mod repair_scope_tests { #[test] fn given_a_repair_request_when_serving_should_clamp_to_the_frontier_but_not_below_it() { + let consensus = VsrConsensus::new(1, 0, 3, 0, IggyMessageBus::new(0), LocalPipeline::new()); + consensus.init(); + consensus.advance_commit_max(40); + consensus.sequencer().set_sequence(90); + let ceiling = |to_op| { + let request = Message::<RequestPreparesHeader>::new(std::mem::size_of::< + RequestPreparesHeader, + >()) + .transmute_header(|_, header: &mut RequestPreparesHeader| { + header.command = Command::RequestPrepares; + header.cluster = consensus.cluster(); + header.size = u32::try_from(std::mem::size_of::<RequestPreparesHeader>()).unwrap(); + header.view = consensus.view(); + header.from_op = 1; + header.to_op = to_op; + }); + repair_serve_ceiling(request.header(), &consensus) + }; // `validate` accepts any `to_op >= from_op` and the serve path walks op by // op with no `.await`, so an unclamped ceiling hangs the whole shard. - assert_eq!(repair_serve_ceiling(u64::MAX, 40, 90), 90); - assert_eq!(repair_serve_ceiling(50, 40, 90), 50); + assert_eq!(ceiling(u64::MAX), 90); + assert_eq!(ceiling(50), 50); // The suffix a new primary repairs toward sits above every commit point, // so clamping to `commit_max` alone deadlocks the view change. - assert_eq!(repair_serve_ceiling(90, 40, 90), 90); + assert_eq!(ceiling(90), 90); // `commit_max` above the local head still counts: heartbeats outrun prepares. - assert_eq!(repair_serve_ceiling(u64::MAX, 120, 90), 120); + consensus.advance_commit_max(120); + assert_eq!(ceiling(u64::MAX), 120); } #[test] diff --git a/core/shard/src/partition_io.rs b/core/shard/src/partition_io.rs index 0f0895cee..6c3d08ca6 100644 --- a/core/shard/src/partition_io.rs +++ b/core/shard/src/partition_io.rs @@ -998,6 +998,7 @@ mod tests { .send_request_prepares( consensus.cluster(), consensus.replica(), + consensus.view(), PEER, 1, 1, @@ -4335,6 +4336,24 @@ mod tests { ); owner.on_repair_prepare(repaired).await; } + let before = partition.consensus().commit_min(); + for command in [Command::RangeEvicted, Command::RepairDone] { + let foreign_reply = reply.clone().transmute_header( + |header, foreign: &mut RepairRangeReplyHeader| { + *foreign = header; + foreign.command = command; + foreign.replica = REPAIR_PRIMARY; + foreign.seal(); + }, + ); + owner.on_repair_range_reply(&foreign_reply).await; + assert_eq!(partition.consensus().commit_min(), before); + assert_eq!(partition.repair.unwrap().floor, None); + assert!( + requests.borrow().is_empty(), + "a different peer cannot complete or advance this repair session" + ); + } owner.on_repair_range_reply(&reply).await; } assert!(requests.borrow().is_empty()); @@ -4699,6 +4718,134 @@ mod tests { ); } + #[compio::test] + async fn given_abandoned_prepares_when_a_backup_requests_repair_should_serve_only_committed_ops() + { + const REPLICAS: u8 = 5; + const DONOR: u8 = 3; + const BACKUP: u8 = 4; + const NEXT_VIEW: u32 = 2; + const NEXT_PRIMARY: u8 = 2; + const COMMITTED: u64 = 2; + const HEAD: u64 = 3; + 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| { + let message = + Message::<GenericHeader>::try_from(Owned::copy_from_slice(frame.as_slice())) + .unwrap(); + if message.header().command == Command::RepairPrepare { + let prepare = Message::<iggy_binary_protocol::RepairPrepareHeader>::try_from( + Owned::copy_from_slice(frame.as_slice()), + ) + .unwrap(); + captured.borrow_mut().push(prepare.header().0); + } + Ok(()) + })); + for requester in [BACKUP, NEXT_PRIMARY] { + assert!(bus.owner_table().try_claim(requester, 1)); + } + let (owner, _sender) = test_owner(&bus, None); + let namespace = IggyNamespace::new(0, 0, 0); + let prepares = primary_prepares(&bus, namespace, HEAD).await; + let mut donor = repairing_replica(&bus, namespace, (DONOR, REPLICAS), REPAIR_PRIMARY, 0, 0); + donor.repair = None; + for prepare in prepares { + donor.on_replicate(prepare).await; + } + donor.consensus().advance_commit_max(COMMITTED); + donor + .commit_journal(owner.plane.partitions().config()) + .await; + let change = + view_control_message(namespace, Command::StartViewChange, NEXT_VIEW, NEXT_PRIMARY); + let MessageBag::StartViewChange(change) = change else { + unreachable!() + }; + donor + .consensus() + .handle_start_view_change(consensus::PlaneKind::Partitions, change.header()); + assert_eq!(donor.consensus().status(), consensus::Status::ViewChange); + owner.plane.partitions().remove(&namespace).unwrap(); + owner.plane.partitions().insert(namespace, donor); + + for (requester, view, through) in [ + (BACKUP, NEXT_VIEW, COMMITTED), + (NEXT_PRIMARY, NEXT_VIEW, HEAD), + (NEXT_PRIMARY, REPAIR_VIEW, COMMITTED), + ] { + let request = Message::<RequestPreparesHeader>::new(size_of::<RequestPreparesHeader>()) + .transmute_header(|_, header: &mut RequestPreparesHeader| { + header.command = Command::RequestPrepares; + header.cluster = 1; + header.group = namespace.inner(); + header.replica = requester; + header.view = view; + header.from_op = 1; + header.to_op = HEAD; + header.size = u32::try_from(size_of::<RequestPreparesHeader>()).unwrap(); + header.seal(); + }); + sent.borrow_mut().clear(); + owner.on_request_prepares(&request).await; + let served: Vec<_> = sent.borrow().iter().map(|header| header.op).collect(); + assert_eq!( + served, + (1..=through).collect::<Vec<_>>(), + "requester {requester} in view {view}" + ); + } + } + + #[compio::test] + async fn given_a_blank_start_view_head_when_repair_arrives_should_preserve_and_prove_history() { + const HEAD: u64 = 1; + const NEW_VIEW: u32 = 3; + for held in [false, true] { + let bus = Rc::new(IggyMessageBus::new(0)); + let (owner, _sender) = test_owner(&bus, None); + let namespace = IggyNamespace::new(0, 0, 0); + let prepares = primary_prepares(&bus, namespace, HEAD).await; + let prepare = prepares.into_iter().next().unwrap(); + let mut partition = + repairing_replica(&bus, namespace, (REPAIR_BACKUP, 3), REPAIR_PRIMARY, 0, 0); + partition.repair = None; + if held { + partition.on_replicate(prepare.clone()).await; + } + owner.plane.partitions().remove(&namespace).unwrap(); + owner.plane.partitions().insert(namespace, partition); + owner + .on_start_view(start_view_suffix( + namespace, + NEW_VIEW, + HEAD, + HEAD, + &[consensus::dvc_blank(HEAD)], + )) + .await; + let partition = owner.plane.partitions().get_by_ns(&namespace).unwrap(); + assert!(partition.consensus().view_log_unproven()); + assert!(!partition.view_truncation_pending()); + assert_eq!(partition.log.journal().inner.holds_op(HEAD), held); + let repaired = prepare.transmute_header( + |header, repaired: &mut iggy_binary_protocol::RepairPrepareHeader| { + repaired.0 = header; + repaired.0.command = Command::RepairPrepare; + }, + ); + owner.on_repair_prepare(repaired).await; + let partition = owner.plane.partitions().get_mut_by_ns(&namespace).unwrap(); + assert!(!partition.consensus().view_log_unproven()); + partition + .commit_journal(owner.plane.partitions().config()) + .await; + assert_eq!(partition.consensus().commit_min(), HEAD); + } + } + fn start_view_suffix( namespace: IggyNamespace, view: u32,
