This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new fa8070494 fix(cluster): stop partition repair asking for an inverted
op range (#4246)
fa8070494 is described below
commit fa8070494a94a1e78f0b598fd1d394ad34ae7679
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Tue Sep 22 10:02:20 2026 +0200
fix(cluster): stop partition repair asking for an inverted op range (#4246)
---
core/partitions/src/iggy_partition.rs | 80 +++++++++++++++++++++++++++++++--
core/server/src/partition_reconciler.rs | 67 +++++++++++++++++++++++++++
core/shard/src/lib.rs | 30 +++++++++++--
3 files changed, 171 insertions(+), 6 deletions(-)
diff --git a/core/partitions/src/iggy_partition.rs
b/core/partitions/src/iggy_partition.rs
index 8f7f4f100..d3ae0e6c7 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -8202,7 +8202,16 @@ where
self.repair = None;
return RepairConclusion::Done;
}
- if let Some(floor) = session.floor {
+ // A floor at or below the live commit point is moot: it cannot move
+ // `commit_min`, so nothing below it is skipped and the connection
+ // check has no jump to guard. Verifying it could only refuse: below
+ // `commit_to_op` the window never proved complete, because the ops
+ // between the floor and `commit_min` are committed with their headers
+ // evicted, and at `commit_to_op` the empty window refused outright.
+ if let Some(floor) = session
+ .floor
+ .filter(|&floor| floor > self.consensus().commit_min())
+ {
// A peer may have evicted past this replica's commit frontier;
// an unclamped floor would drive commit_min above commit_max and
// panic the next advance.
@@ -8276,8 +8285,7 @@ where
}
return RepairConclusion::InProgress;
}
- let commit_min = self.consensus().commit_min();
- if floor > commit_min {
+ if floor > self.consensus().commit_min() {
self.consensus().set_commit_floor(floor);
}
}
@@ -15663,6 +15671,72 @@ mod tests {
assert_eq!(partition.consensus().commit_min(), 0);
assert!(partition.repair.is_some());
}
+
+ /// 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
+ /// refused every round (evicted headers never complete the window) while
+ /// the walk inside the refusal ran `commit_min` to the fetch ceiling, and
+ /// the reply handler asked the peer for `fetch_to_op + 1 ..= fetch_to_op`.
+ #[compio::test]
+ async fn
given_floor_below_commit_min_when_completing_repair_should_ignore_the_floor() {
+ let mut partition = test_partition();
+ partition.consensus().restore_commit_state(7, 8);
+ // Disconnected on its face: the served window starts at offset 20 and
+ // 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)));
+
+ let conclusion = partition.complete_repair(&repair_config()).await;
+
+ assert_eq!(conclusion, RepairConclusion::Done);
+ assert_eq!(partition.consensus().commit_min(), 8);
+ assert!(
+ partition.repair.is_none(),
+ "nothing is left to fetch, so the session must close instead of \
+ re-requesting past its ceiling"
+ );
+ }
+
+ #[compio::test]
+ async fn
given_floor_clamped_to_commit_min_when_completing_repair_should_still_refuse() {
+ let mut partition = test_partition();
+ 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));
+
+ let conclusion = partition.complete_repair(&repair_config()).await;
+
+ assert_eq!(
+ conclusion,
+ RepairConclusion::FloorRefused { floor: 5, to_op: 5 },
+ "a floor the clamp pulls down to commit_min still names an evicted
\
+ range and must escape to state transfer"
+ );
+ assert!(partition.repair.is_none());
+ }
+
+ #[compio::test]
+ async fn
given_moot_floor_with_unfetched_suffix_when_completing_repair_should_keep_repairing()
{
+ let mut partition = test_partition();
+ partition.consensus().restore_commit_state(5, 5);
+ // 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));
+
+ let conclusion = partition.complete_repair(&repair_config()).await;
+
+ assert_eq!(conclusion, RepairConclusion::InProgress);
+ let session = partition.repair.expect("the suffix fetch stays armed");
+ assert!(
+ partition.consensus().commit_min() < session.fetch_to_op,
+ "the stall retry must still have a range to ask for"
+ );
+ }
+
/// Temp partition directory for the state-transfer fence specs below.
async fn transfer_fence_dir(label: &str) -> String {
let dir = std::env::temp_dir().join(format!(
diff --git a/core/server/src/partition_reconciler.rs
b/core/server/src/partition_reconciler.rs
index 2e1ce4c3e..85894ccb9 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -1488,6 +1488,7 @@ mod tests {
Command, Operation, PrepareHeader, RepairRangeReplyHeader, ReplyHeader,
RequestPreparesHeader, RoutedRequestHeader, WireIdentifier,
WireOptions,
};
+ use journal::Journal;
use message_bus::IggyMessageBus;
use metadata::IggyMetadata;
use metadata::MuxStateMachine;
@@ -3084,6 +3085,72 @@ mod tests {
);
}
+ /// Receive half of the inverted-range fix: a `RepairDone` landing on a
+ /// session whose floor the commit walk already passed must close the
+ /// session. Verifying the moot floor kept it armed while the walk ran to
+ /// the fetch ceiling, and the next chunk request was `9..=8`.
+ #[compio::test]
+ async fn repair_done_over_a_passed_floor_closes_the_session() {
+ const NONCE: u128 = 13;
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-repair-ceiling");
+ seed_topic(&mux, 2, 0, "topic-repair-ceiling", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ reconcile_pass(&ctx).await;
+
+ let ns = IggyNamespace::new(0, 0, 0);
+ let served = CreateStreamRequest {
+ name: WireName::new("served-op").expect("test stream name fits
WireName"),
+ options: WireOptions::empty(),
+ };
+ {
+ let partitions = shard.plane.partitions();
+ let partition = partitions
+ .get_mut_by_ns(&ns)
+ .expect("partition is materialised");
+ // Ops 5..=7 committed and evicted after the request went out, op 8
+ // is the served remainder, and the window's first batch sits above
+ // the boot-recovered durable end.
+ partition.consensus().restore_commit_state(7, 8);
+ partition.recovered_durable_offset = Some(10);
+ partition
+ .log
+ .journal()
+ .inner
+ .append(build_prepare(8, Operation::CreateStream,
&served).into_frozen())
+ .await
+ .expect("journal the served op");
+ partition.repair = Some(RepairSession {
+ nonce: NONCE,
+ view: 0,
+ commit_to_op: 8,
+ fetch_to_op: 8,
+ floor: Some(5),
+ peer: 1,
+ first_batch_offset: Some(20),
+ idle_ticks: 0,
+ });
+ }
+
+ shard
+ .on_message(build_repair_range_reply(ns, Command::RepairDone,
NONCE, 8))
+ .await;
+
+ let partitions = shard.plane.partitions();
+ let partition = partitions
+ .get_mut_by_ns(&ns)
+ .expect("partition survives the reply");
+ assert_eq!(partition.consensus().commit_min(), 8);
+ assert!(
+ partition.repair.is_none(),
+ "a walk that reached the fetch ceiling leaves nothing to request"
+ );
+ }
+
/// Serve half of the purge gate in `on_request_prepares`: while a
/// committed purge has not applied locally, the journal still holds
/// pre-purge entries with no floor to fence them, so serving a rejoiner
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 287b747b8..dd188fc0d 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -5701,17 +5701,18 @@ where
} else {
let commit_min = partition.consensus().commit_min();
let next = partition.repair.as_ref().and_then(|live| {
- (commit_min > before).then_some((live.peer,
live.nonce, live.fetch_to_op))
+ 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 cluster = partition.consensus().cluster();
let self_id = partition.consensus().replica();
- if let Some((peer, nonce, to_op)) = next {
+ if let Some((peer, nonce, from_op, to_op)) = next {
self.send_request_prepares(
cluster,
self_id,
peer,
nonce,
- commit_min + 1,
+ from_op,
to_op,
header.group,
)
@@ -11061,6 +11062,18 @@ fn partition_repair_fetch_to_op(
.then(|| missing_suffix.unwrap_or(commit_max))
}
+/// Start of the next chunk to pull after a `RepairDone`, or `None` when the
+/// walk made no progress (the stall retry owns the remainder) or already
+/// stands at the session's fetch ceiling. The sweep closes such a session,
+/// because `fetch_to_op` never sits below `commit_to_op`.
+///
+/// A session can outlive its last fetchable op until that close, and
+/// `from_op > to_op` fails `RequestPreparesHeader::validate` on the serving
+/// peer, which drops the frame as unparsable.
+fn partition_repair_next_chunk(before: u64, commit_min: u64, fetch_to_op: u64)
-> Option<u64> {
+ (commit_min > before && commit_min < fetch_to_op).then_some(commit_min + 1)
+}
+
/// Highest adopted suffix op whose bodies are not all present above
`commit_max`.
///
/// The shape `maybe_request_partition_repair` widens its window for, read here
@@ -12067,6 +12080,17 @@ mod repair_scope_tests {
assert_eq!(adopted_suffix_head(&missing, 98, 101), None);
}
+ #[test]
+ fn
given_a_walk_at_the_fetch_ceiling_when_repair_done_lands_should_not_request_a_chunk()
{
+ assert_eq!(super::partition_repair_next_chunk(4, 7, 8), Some(8));
+ // No progress leaves the remainder to the stall retry.
+ assert_eq!(super::partition_repair_next_chunk(7, 7, 8), None);
+ // Progress that reached the ceiling has nothing left to ask for: the
+ // sweep closes the session, and `9..=8` is not a range.
+ assert_eq!(super::partition_repair_next_chunk(7, 8, 8), None);
+ assert_eq!(super::partition_repair_next_chunk(7, 9, 8), None);
+ }
+
#[test]
fn
given_a_parked_view_when_fetching_above_commit_should_require_dense_canonical_suffix()
{
let pending = parked();
