You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/10/09 13:17:16 UTC
(iggy) 01/01: merge master
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 7c0279935f3ee93522396fa1ed3b74877d74bc1b
Merge: e89d2950c 5753d4958
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Fri Oct 9 15:17:09 2026 +0200
merge master
Cargo.lock | 1 -
core/ai/mcp/README.md | 2 +-
core/ai/mcp/src/service/mod.rs | 25 -
core/ai/mcp/src/service/requests.rs | 15 -
core/binary_protocol/src/codes.rs | 8 +-
core/binary_protocol/src/consensus/error.rs | 8 +-
core/binary_protocol/src/consensus/operation.rs | 31 +-
core/binary_protocol/src/dispatch.rs | 68 +-
core/binary_protocol/src/requests/streams/mod.rs | 2 -
.../src/requests/streams/purge_stream.rs | 60 -
core/binary_protocol/src/requests/topics/mod.rs | 2 -
.../src/requests/topics/purge_topic.rs | 86 -
core/binary_protocol/src/responses/streams/mod.rs | 2 -
.../src/responses/streams/purge_stream.rs | 19 -
core/binary_protocol/src/responses/topics/mod.rs | 2 -
.../src/responses/topics/purge_topic.rs | 19 -
core/cli/src/args/stream.rs | 18 -
core/cli/src/args/topic.rs | 27 -
core/cli/src/commands/binary_streams/mod.rs | 1 -
.../src/commands/binary_streams/purge_stream.rs | 59 -
core/cli/src/commands/binary_topics/mod.rs | 1 -
core/cli/src/commands/binary_topics/purge_topic.rs | 67 -
core/cli/src/main.rs | 9 +-
core/common/src/http/streams/mod.rs | 1 -
core/common/src/http/streams/purge_stream.rs | 37 -
core/common/src/http/topics/mod.rs | 1 -
core/common/src/http/topics/purge_topic.rs | 41 -
core/common/src/traits/binary_impls/streams.rs | 16 +-
core/common/src/traits/binary_impls/topics.rs | 26 +-
core/common/src/traits/consumer_offset_client.rs | 4 -
core/common/src/traits/stream_client.rs | 4 -
core/common/src/traits/topic_client.rs | 8 -
core/consensus/src/impls.rs | 29 +-
core/consensus/src/observability.rs | 2 -
core/consensus/src/state_manifest.rs | 5 +-
core/integration/tests/cli/stream/mod.rs | 1 -
.../tests/cli/stream/test_stream_help_command.rs | 1 -
.../tests/cli/stream/test_stream_purge_command.rs | 223 --
core/integration/tests/cli/topic/mod.rs | 1 -
.../tests/cli/topic/test_topic_help_command.rs | 1 -
.../tests/cli/topic/test_topic_purge_command.rs | 295 --
.../tests/cluster/consumer_offset_quota.rs | 74 +-
core/integration/tests/cluster/partition_dedup.rs | 42 +-
.../tests/data_integrity/storage_compat.rs | 264 +-
core/integration/tests/mcp/mod.rs | 24 +-
.../server/{purge_delete.rs => delete_segments.rs} | 22 +-
.../tests/server/descriptor_exhaustion_vsr.rs | 4 +-
core/integration/tests/server/http_vsr.rs | 40 +
core/integration/tests/server/mod.rs | 9 +-
core/integration/tests/server/purge_vsr.rs | 142 -
core/integration/tests/server/raw_tcp.rs | 18 +-
.../server/scenarios/authentication_scenario.rs | 2 -
...ete_scenario.rs => delete_segments_scenario.rs} | 464 ---
.../scenarios/delete_stats_rollback_scenario.rs | 8 +-
core/integration/tests/server/scenarios/mod.rs | 4 +-
.../stale_client_consumer_group_scenario.rs | 3 +-
.../scenarios/stream_size_validation_scenario.rs | 52 +-
.../tests/server/scenarios/system_scenario.rs | 78 -
core/integration/tests/server/unknown_code_vsr.rs | 136 +-
core/journal/src/partition_journal.rs | 260 +-
core/journal/src/partition_journal/segments.rs | 17 +-
.../src/permissioner/permissioner_rules/streams.rs | 4 -
.../src/permissioner/permissioner_rules/topics.rs | 9 -
core/metadata/src/stm/authz.rs | 26 +-
core/metadata/src/stm/result.rs | 26 +-
core/metadata/src/stm/snapshot.rs | 13 +-
core/metadata/src/stm/stream.rs | 602 +---
core/partitions/Cargo.toml | 1 -
core/partitions/src/iggy_partition.rs | 3731 ++++----------------
core/partitions/src/iggy_partitions.rs | 9 +-
core/partitions/src/io.rs | 88 +-
core/partitions/src/journal.rs | 85 +-
core/partitions/src/lib.rs | 2 +-
core/partitions/src/log.rs | 28 +-
core/partitions/src/offset_storage.rs | 242 +-
core/partitions/src/persistence.rs | 28 -
core/partitions/src/poll_plan.rs | 14 +-
core/partitions/src/state_transfer.rs | 231 +-
core/partitions/tests/consumer_offset_history.rs | 719 ----
.../src/client_wrappers/binary_stream_client.rs | 10 -
.../sdk/src/client_wrappers/binary_topic_client.rs | 14 -
core/sdk/src/clients/binary_streams.rs | 4 -
core/sdk/src/clients/binary_topics.rs | 12 -
core/sdk/src/clients/client.rs | 4 +-
core/sdk/src/http/streams.rs | 9 -
core/sdk/src/http/topics.rs | 13 -
core/sdk/src/poll_routing.rs | 4 +-
core/server/README.md | 6 +-
core/server/server.http | 8 -
core/server/src/boot/mod.rs | 44 +-
core/server/src/consumer_group.rs | 2 +-
core/server/src/dispatch/authz.rs | 2 -
core/server/src/dispatch/failure.rs | 137 +-
core/server/src/dispatch/mod.rs | 158 +-
core/server/src/dispatch/partition.rs | 171 +-
core/server/src/dispatch/test_support.rs | 4 +
core/server/src/http.rs | 11 +-
core/server/src/http/handlers.rs | 50 +-
core/server/src/http/metrics.rs | 2 +-
core/server/src/http/submit.rs | 2 +-
core/server/src/partition_helpers.rs | 14 +-
core/server/src/partition_reconciler.rs | 329 +-
core/server/src/server_error.rs | 2 +-
core/server_common/src/consensus_message.rs | 12 +-
core/shard/src/lib.rs | 221 +-
core/shard/src/metrics.rs | 32 +-
core/shard/src/partition_io.rs | 264 +-
core/shard/src/poll.rs | 26 +-
core/shard/src/poll/completion_tests.rs | 76 +-
core/shard/src/router.rs | 122 +-
core/simulator/src/client.rs | 24 +-
.../simulator/src/consumer_offset_history_tests.rs | 219 --
core/simulator/src/lib.rs | 3 -
core/simulator/src/storage.rs | 2 +-
core/simulator/src/storage/bookmark_recovery.rs | 360 ++
core/simulator/src/storage/purge.rs | 758 ----
core/simulator/src/storage/tests.rs | 107 +-
core/simulator/src/workload/actions.rs | 4 +-
core/simulator/src/workload/ops/mod.rs | 4 -
core/simulator/src/workload/ops/purge_stream.rs | 74 -
core/simulator/src/workload/ops/purge_topic.rs | 83 -
core/simulator/src/workload/options.rs | 6 +-
foreign/cpp/include/iggy.hpp | 28 -
foreign/cpp/src/client.cpp | 8 -
foreign/cpp/src/client.rs | 37 -
foreign/cpp/src/lib.rs | 2 -
foreign/cpp/tests/e2e/stream.cpp | 372 --
foreign/cpp/tests/e2e/topic.cpp | 331 --
foreign/cpp/tests/unit/unit_tests.cpp | 2 -
.../Helpers/Eventually.cs | 43 -
.../Iggy_SDK.Tests.Integration/StreamsTests.cs | 32 -
.../Iggy_SDK.Tests.Integration/TopicsTests.cs | 32 -
.../csharp/Iggy_SDK/Contracts/Tcp/TcpContracts.cs | 7 -
foreign/csharp/Iggy_SDK/IggyClient/IIggyStream.cs | 12 -
foreign/csharp/Iggy_SDK/IggyClient/IIggyTopic.cs | 13 -
.../Implementations/HttpMessageStream.cs | 20 -
.../IggyClient/Implementations/TcpMessageStream.cs | 15 -
foreign/csharp/Iggy_SDK/Utils/CommandCodes.cs | 2 -
foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs | 10 +-
.../ClientTests/HttpTopicOptionsTests.cs | 17 -
.../Iggy_SDK_Tests/VsrTests/VsrOperationTests.cs | 6 +-
foreign/go/client/tcp/tcp_poll_routing_test.go | 4 +-
foreign/go/internal/command/code.go | 2 -
foreign/go/internal/vsr/header_test.go | 1 -
foreign/go/internal/vsr/operation.go | 6 -
foreign/go/internal/vsr/operation_test.go | 6 +-
foreign/go/internal/vsr/protocol_parity_test.go | 2 -
.../iggy/client/async/tcp/vsr/VsrOperation.java | 8 +-
.../java/org/apache/iggy/serde/CommandCode.java | 3 +-
foreign/node/src/e2e/tcp.stream.e2e.ts | 4 -
foreign/node/src/e2e/tcp.topic.e2e.ts | 4 -
foreign/node/src/wire/command-set.ts | 4 -
foreign/node/src/wire/command.code.ts | 2 -
foreign/node/src/wire/stream/index.ts | 1 -
.../node/src/wire/stream/purge-stream.command.ts | 49 -
foreign/node/src/wire/topic/index.ts | 1 -
foreign/node/src/wire/topic/purge-topic.command.ts | 54 -
foreign/node/src/wire/vsr/operation.test.ts | 4 +-
foreign/node/src/wire/vsr/operation.ts | 4 -
foreign/python/apache_iggy.pyi | 40 -
foreign/python/src/client.rs | 65 -
foreign/python/tests/test_stream.py | 97 +-
foreign/python/tests/test_topic.py | 119 -
foreign/swift/Tests/IggyTests/Fixtures/golden.json | 2 -
foreign/swift/Tools/golden-vectors/src/main.rs | 18 +-
gateways/kafka/README.md | 2 +-
gateways/kafka/src/protocol/handlers/fetch.rs | 7 +-
gateways/kafka/tests/fetch_real_bridge_tests.rs | 33 +-
168 files changed, 2274 insertions(+), 11415 deletions(-)
diff --cc core/partitions/src/iggy_partition.rs
index e04caf728,6e85b2833..7b3eb6d21
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@@ -11410,22 -10539,17 +10525,18 @@@ wher
.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.
++ // serve a holed log.
(None, _) => {
floor < session.commit_to_op
&& committed_shape.complete
@@@ -11502,158 -10644,6 +10613,155 @@@
}
}
+ /// 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`.
++ // assigns offsets in op order.
+ 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.
++ /// for, and a batch resident from an earlier session or the boot WAL
++ /// counts as much as one repaired now.
+ 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 from = floor.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();
@@@ -12759,82 -11719,6 +11844,81 @@@ 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 [
@@@ -23553,19 -22368,16 +22568,19 @@@
}
}
- pub(super) fn armed_session(to_op: u64, floor: u64) -> RepairSession {
- 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)
++ 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,
@@@ -23723,23 -22535,9 +22738,19 @@@
/// A repaired `SendMessages` prepare with an explicit chain identity, as a
/// serving peer ships it.
- pub(super) fn repaired_send_prepare(
- op: u64,
- parent: u128,
- checksum: u128,
- ) -> Message<PrepareHeader> {
+ fn repaired_send_prepare(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);
@@@ -24410,116 -22946,6 +23406,92 @@@
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
@@@ -25721,10 -23931,81 +24477,81 @@@
"both cursors must advance once the retry succeeded"
);
}
- }
- #[cfg(test)]
- mod retention_tests {
+ /// Eviction is what makes a second copy reachable at all:
+ /// `apply_repaired_prepare` refuses an op the journal still holds, so a
+ /// re-delivered repair frame is only re-journaled once the flush that
+ /// persisted it has evicted it. `commit_min` is what lets the frame past
+ /// the other guard: the flush persists the whole committed prefix while
+ /// the walk advances `commit_min` by at most [`COMMIT_WALK_OPS_MAX`], so
+ /// every op above the walk's reach is persisted, evicted, and still
+ /// re-deliverable. The flush is the last gate, and a durable line frozen at
+ /// boot cannot see its own writes -- a rejoining backup then diverged from
+ /// the group by exactly the ops it repaired.
+ ///
+ /// Journaled through the repair path, the one that keeps a batch's original
+ /// offsets; `apply_replicated_operation` re-stamps from the local counter
+ /// and so cannot collide with itself.
+ #[compio::test]
+ async fn given_offsets_already_persisted_when_flushed_again_should_not_append_a_second_copy() {
+ const CHECKSUM: u128 = 0x5a;
+ // One op past the walk budget, so the last one stays above `commit_min`
+ // after the flush that persists it.
+ const OPS: u64 = COMMIT_WALK_OPS_MAX as u64 + 1;
+ let mut partition = test_partition();
+ 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
+ .apply_repaired_prepare(repaired_send_prepare(op, 0, CHECKSUM))
+ .await;
+ }
+ partition.consensus().advance_commit_max(OPS);
+ partition.commit_journal(&repair_config()).await;
+ assert_eq!(partition.consensus().commit_min(), OPS - 1);
+ assert_eq!(
+ partition.log.active_segment().size.as_bytes_u64(),
+ record_len * (OPS - 1)
+ );
+ assert_eq!(
+ partition
+ .collect_committable_from_journal(COMMIT_WALK_OPS_MAX, &repair_config())
+ .iter()
+ .map(|entry| entry.op)
+ .collect::<Vec<_>>(),
+ vec![OPS]
+ );
+ partition
+ .flush_committed_messages(&repair_config())
+ .await
+ .unwrap();
+ assert_eq!(
+ partition.log.active_segment().size.as_bytes_u64(),
+ record_len * OPS,
+ "the first flush persists the whole committed prefix"
+ );
+ assert!(
+ partition.consensus().commit_min() < OPS,
+ "the premise: the walk leaves the last op above `commit_min`, which \
+ 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;
+ partition.commit_journal(&repair_config()).await;
+ assert_eq!(
+ partition.log.active_segment().size.as_bytes_u64(),
+ record_len * OPS,
+ "offsets the segment already holds must not be appended twice"
+ );
+ }
+ }
+
+ #[cfg(test)]
+ mod retention_tests {
use super::*;
use iggy_common::IggyDuration;
use std::time::Duration;
diff --cc core/shard/src/poll/completion_tests.rs
index a13e5f042,e940eab44..dee713662
--- a/core/shard/src/poll/completion_tests.rs
+++ b/core/shard/src/poll/completion_tests.rs
@@@ -30,11 -30,10 +30,11 @@@ use consensus::
encode_state_manifest,
};
use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest;
- use iggy_binary_protocol::requests::topics::{DeleteTopicRequest, PurgeTopicRequest};
+ use iggy_binary_protocol::requests::topics::DeleteTopicRequest;
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,