This is an automated email from the ASF dual-hosted git repository. spetz pushed a commit to branch offset_hardening in repository https://gitbox.apache.org/repos/asf/iggy.git
commit afd204e80667d65cace99ee953f1bb785e8139f2 Author: spetz <[email protected]> AuthorDate: Sat Sep 5 20:56:24 2026 +0200 fix tests --- core/partitions/src/iggy_partition.rs | 84 +++++++++++++++++++++++++++--- core/sdk/src/quic/quic_client.rs | 5 +- core/sdk/src/tcp/tcp_client.rs | 8 ++- core/sdk/src/websocket/websocket_client.rs | 5 +- core/server/src/dispatch/partition.rs | 10 +++- 5 files changed, 102 insertions(+), 10 deletions(-) diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 4b5b23127..2c19ac694 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -4659,6 +4659,12 @@ where let (frozen_batches, index_bytes, flush_index, batch_count, committed_info, chunk_len) = { let segment = self.log.active_segment(); let mut file_position = segment.size.as_bytes_u64(); + let persisted_end = if file_position == 0 { + segment.start_offset.checked_sub(1) + } else { + Some(segment.end_offset) + } + .max(self.recovered_durable_offset); let mut flush_index = None; let mut frozen = Vec::with_capacity(entries.len()); let mut batch_count = 0u32; @@ -4707,13 +4713,13 @@ where if message_count == 0 { continue; } - // A repaired batch at or below the boot-time recovered - // durable offset is already IN the segments this replica - // recovered; persisting it again would append duplicate - // bytes past the segment end. Evict it without writing. - // Live traffic always sits above the (immutable) line. + // Flush can run ahead of the bounded commit walk. Repair + // may re-journal an evicted batch above commit_min even + // after this process persisted it. Include the current + // segment frontier, not just the boot recovery frontier, + // so that replay cannot append a second copy. let batch_end = batch.header.base_offset + u64::from(message_count) - 1; - if let Some(durable) = self.recovered_durable_offset + if let Some(durable) = persisted_end && batch_end <= durable { continue; @@ -11423,6 +11429,72 @@ mod tests { ); } + #[compio::test] + async fn given_flushed_repair_ahead_of_commit_min_when_replayed_should_persist_only_new_batches() + { + let dir = tempfile::tempdir().expect("temp dir"); + let log_path = dir.path().join("segment.log"); + let index_path = dir.path().join("segment.index"); + let mut fixture = PersistFixture::new( + log_path.to_str().expect("utf-8 path"), + index_path.to_str().expect("utf-8 path"), + ) + .await; + let partition = &mut fixture.partition; + partition.log.journal().inner.set_repair_retention(true); + partition.repair = Some(armed_session(4, 0, None)); + let prepares: Vec<_> = (1..=4) + .map(|op| repaired_send_prepare(op, 0, u128::from(op)).into_frozen()) + .collect(); + let replay = |op: usize| { + let bytes = prepares[op - 1].as_slice(); + let mut message = Message::<PrepareHeader>::new(bytes.len()); + message.as_mut_slice().copy_from_slice(bytes); + message + }; + for op in 1..=3 { + partition.apply_repaired_prepare(replay(op)).await; + } + partition.consensus().advance_commit_max(3); + partition + .flush_committed_messages(&repair_config()) + .await + .expect("flush before the commit walk catches up"); + assert_eq!(partition.consensus().commit_min(), 0); + assert_eq!(partition.recovered_durable_offset, None); + let original = std::fs::read(&log_path).expect("read initial segment"); + let original_index = std::fs::read(&index_path).expect("read initial index"); + + for op in 2..=3 { + partition.apply_repaired_prepare(replay(op)).await; + } + partition + .flush_committed_messages(&repair_config()) + .await + .expect("flush replayed batches"); + assert_eq!(std::fs::read(&log_path).unwrap(), original); + assert_eq!(std::fs::read(&index_path).unwrap(), original_index); + + let next = replay(4); + let mut expected = original; + expected.extend_from_slice(&next.as_slice()[size_of::<PrepareHeader>()..]); + partition.apply_repaired_prepare(next).await; + partition.consensus().advance_commit_max(4); + partition + .flush_committed_messages(&repair_config()) + .await + .expect("flush the new batch"); + assert_eq!(std::fs::read(&log_path).unwrap(), expected); + // The commit walk needs resident headers after the earlier flush. + for op in 1..=4 { + partition.apply_repaired_prepare(replay(op)).await; + } + partition.commit_journal(&repair_config()).await; + assert_eq!(partition.consensus().commit_min(), 4); + assert!(partition.fatal.is_none()); + assert_eq!(std::fs::read(&log_path).unwrap(), expected); + } + #[cfg(target_os = "linux")] #[compio::test] async fn given_a_persist_failure_on_a_committed_op_should_fence_the_partition_not_panic() { diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs index 5dcadc1d6..fec2358e7 100644 --- a/core/sdk/src/quic/quic_client.rs +++ b/core/sdk/src/quic/quic_client.rs @@ -218,7 +218,10 @@ impl BinaryTransport for QuicClient { } else if let Some(next) = roster_walk.as_mut().and_then(RosterWalk::next) { (next, true) } else { - break; + // A single-node roster can still be converging a newly + // committed partition. Retry this explicitly unadmitted + // request on the current endpoint within the same budget. + (current, false) }; loop { diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs index 154670833..6d292dbc5 100644 --- a/core/sdk/src/tcp/tcp_client.rs +++ b/core/sdk/src/tcp/tcp_client.rs @@ -1324,7 +1324,13 @@ impl TcpClient { // never reach the rest of the roster. (next, true) } else { - return Err(IggyError::TransientNotAccepted); + // A one-node roster has nowhere else to walk while a + // freshly committed partition is still materialising. + // The server explicitly did not admit this request, so + // keep retrying the current endpoint within the existing + // overall deadline rather than surfacing a transient + // solely because the roster contains no alternative. + (current, false) }; loop { diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs index 2a22683e0..b3b9cb97e 100644 --- a/core/sdk/src/websocket/websocket_client.rs +++ b/core/sdk/src/websocket/websocket_client.rs @@ -211,7 +211,10 @@ impl BinaryTransport for WebSocketClient { } else if let Some(next) = roster_walk.as_mut().and_then(RosterWalk::next) { (next, true) } else { - break; + // A single-node roster can still be converging a newly + // committed partition. Retry this explicitly unadmitted + // request on the current endpoint within the same budget. + (current, false) }; loop { diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs index beda4d87a..3ec2d806e 100644 --- a/core/server/src/dispatch/partition.rs +++ b/core/server/src/dispatch/partition.rs @@ -457,7 +457,15 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>( operation = ?header.operation, "partition request with unresolved namespace; replying denied" ); - send_deny_reply(shard, transport_client_id, &header, error.as_code()).await; + let status = if matches!( + header.operation, + Operation::StoreConsumerOffset | Operation::DeleteConsumerOffset + ) { + error.as_code() + } else { + IggyError::ResourceNotFound(String::new()).as_code() + }; + send_deny_reply(shard, transport_client_id, &header, status).await; return; } };
