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 <[email protected]> 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,
