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 d29f96524e5621ae617085e880f9e5861f5215f5 Author: Hubert Gruszecki <[email protected]> AuthorDate: Wed Oct 14 12:45:27 2026 +0200 fix(partitions): resume repair on proof progress History proof fences commits while repairing missing predecessors. Partition repair waited for commit progress, so each predecessor stalled until the retry timer and restarted replicas timed out. Track the last requested start and continue when proof advances. Keep timer retries for unchanged proof and verify both paths. Refs #4506 --- core/partitions/src/iggy_partition.rs | 1 + core/partitions/src/types.rs | 3 ++ core/server/src/partition_reconciler.rs | 1 + core/shard/src/lib.rs | 24 +++++++----- core/shard/src/partition_io.rs | 67 +++++++++++++++++++++++++++++---- 5 files changed, 79 insertions(+), 17 deletions(-) diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index a1d0cf132..0655c760b 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -28285,6 +28285,7 @@ mod tests { RepairSession { nonce: 1, view: 0, + from_op: 1, commit_to_op: to_op, fetch_to_op, floor: Some(floor), diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs index 87e19fa80..554070a8a 100644 --- a/core/partitions/src/types.rs +++ b/core/partitions/src/types.rs @@ -292,6 +292,9 @@ 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. + pub from_op: u64, /// Committed frontier this repair must make locally walkable. Floor /// completeness and session completion are bounded here. pub commit_to_op: u64, diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index c7c631d54..0ab61c05c 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -3573,6 +3573,7 @@ mod tests { partition.repair = Some(RepairSession { nonce: NONCE, view: 0, + from_op: 1, commit_to_op: 8, fetch_to_op: 8, floor: None, diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index e7e76a925..b7c340c43 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -5902,9 +5902,9 @@ 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). Proof can move backward through + // a gap while fencing commits, so either kind of progress + // pulls the next chunk without waiting for the retry timer. let before = partition.consensus().commit_min(); if let partitions::RepairConclusion::FloorRefused { floor, to_op } = partition.complete_repair(&config).await @@ -5966,9 +5966,13 @@ 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 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(|| { + 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(); @@ -8074,6 +8078,7 @@ where return None; } session.idle_ticks = 0; + session.from_op = repair_from; Some(( session.peer, session.nonce, @@ -9282,7 +9287,6 @@ where let from_op = consensus .view_log_repair_op() .unwrap_or_else(|| consensus.commit_min() + 1); - let namespace = consensus.group(); // Spent here for the same reason the ceiling is: the four edge-triggered // sites never touch it, so a short edge-armed repair would leave the // count saturated and hand the next real gap an arm on its first tick. @@ -9292,6 +9296,7 @@ where partition.repair = Some(partitions::RepairSession { nonce, view: consensus.view(), + from_op, commit_to_op, fetch_to_op, floor: None, @@ -9300,7 +9305,7 @@ where }); tracing::info!( shard = self.id, - namespace_raw = namespace, + namespace_raw = partition.consensus().group(), from_op, commit_to_op, fetch_to_op, @@ -9314,7 +9319,7 @@ where nonce, from_op, fetch_to_op, - namespace, + partition.consensus().group(), partition.created_revision(), ) .await; @@ -9375,6 +9380,7 @@ where partition.repair = Some(partitions::RepairSession { nonce, view, + from_op, // Merged-log numbers, not the live frontier: `commit_to_op` is what // the walk must reach to finish the session, `fetch_to_op` the head // the view will announce. diff --git a/core/shard/src/partition_io.rs b/core/shard/src/partition_io.rs index 21fcf5931..0f0895cee 100644 --- a/core/shard/src/partition_io.rs +++ b/core/shard/src/partition_io.rs @@ -2469,6 +2469,7 @@ mod tests { partition.repair = Some(RepairSession { nonce: 1, view: 1, + from_op: 1, commit_to_op: 0, fetch_to_op: 0, floor: None, @@ -2540,6 +2541,7 @@ mod tests { partition.repair = Some(RepairSession { nonce: 1, view: partition.consensus().view(), + from_op: 1, commit_to_op: 0, fetch_to_op: 0, floor: None, @@ -3184,6 +3186,7 @@ mod tests { partition.repair = repair_finished.then_some(partitions::RepairSession { nonce: 1, view: 0, + from_op: 1, commit_to_op: commit_max, fetch_to_op: commit_max, floor: None, @@ -4245,9 +4248,27 @@ 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 = 6; + const NEW_HEAD: u64 = 8; const NEW_VIEW: u32 = 3; + const NEW_PRIMARY: u8 = 0; 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(NEW_PRIMARY, 1)); let (owner, _sender) = test_owner(&bus, None); let namespace = IggyNamespace::new(0, 0, 0); let prepares = primary_prepares(&bus, namespace, NEW_HEAD).await; @@ -4279,15 +4300,44 @@ 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 prepare in prepares.iter().rev() { - let repaired = prepare.clone().transmute_header( - |header, repaired: &mut iggy_binary_protocol::RepairPrepareHeader| { - repaired.0 = header; - repaired.0.command = Command::RepairPrepare; - }, + 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 request = request.header(); + assert_eq!(request.from_op, expected_from); + assert_eq!(request.to_op, NEW_HEAD); + let reply = 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.cluster = request.cluster; + header.group = request.group; + header.replica = NEW_PRIMARY; + header.nonce = request.nonce; + header.op = request.to_op; + header.seal(); + }); + owner.on_repair_range_reply(&reply).await; + assert!( + requests.borrow().is_empty(), + "a terminator without bodies must wait for the retry timer" ); - owner.on_repair_prepare(repaired).await; + for op in request.from_op..=request.to_op { + let repaired = prepares[usize::try_from(op - 1).unwrap()] + .clone() + .transmute_header( + |header, repaired: &mut iggy_binary_protocol::RepairPrepareHeader| { + repaired.0 = header; + repaired.0.command = Command::RepairPrepare; + }, + ); + owner.on_repair_prepare(repaired).await; + } + owner.on_repair_range_reply(&reply).await; } + assert!(requests.borrow().is_empty()); let partition = owner.plane.partitions().get_mut_by_ns(&namespace).unwrap(); assert!(!partition.consensus().view_log_unproven()); partition @@ -4800,6 +4850,7 @@ mod tests { partition.repair = Some(RepairSession { nonce: 1, view: REPAIR_VIEW, + from_op: commit_min + 1, commit_to_op: commit_max, fetch_to_op: commit_max, floor: None,
