This is an automated email from the ASF dual-hosted git repository. spetz pushed a commit to branch reject_queued_offsets in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 96c4bf7aab2ee4ed712101c841d15273e2f3cd53 Author: spetz <[email protected]> AuthorDate: Mon Sep 28 22:55:07 2026 +0200 fix(partitions): reject queued offsets after history resets --- core/consensus/src/impls.rs | 16 + core/partitions/src/iggy_partition.rs | 44 +- core/partitions/tests/consumer_offset_history.rs | 534 +++++++++++++++++++++ core/sdk/src/poll_routing.rs | 32 ++ .../simulator/src/consumer_offset_history_tests.rs | 220 +++++++++ core/simulator/src/lib.rs | 3 + 6 files changed, 846 insertions(+), 3 deletions(-) diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index e31680eec..9bc213554 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -283,6 +283,9 @@ pub struct AutoCommitRequestContext { pub struct RequestEntry { /// Automatic commit context owned by this entry until promotion or removal. auto_commit: Option<AutoCommitRequestContext>, + /// Offset writes must not cross a reset of the owner's message history. + /// Explicit stores keep their overwrite and client-dedup behavior. + consumer_offset_history: Option<PollHistoryId>, pub message: Message<RoutedRequestHeader>, /// When the request was parked, in microseconds from the consensus-injected /// clock ([`VsrConsensus::clock_realtime_micros`]). `0` until @@ -330,6 +333,18 @@ impl RequestEntry { self.auto_commit.as_ref() } + /// Bind an explicit offset mutation to the history at owner admission. + #[must_use] + pub const fn with_consumer_offset_history(mut self, history: PollHistoryId) -> Self { + self.consumer_offset_history = Some(history); + self + } + + #[must_use] + pub const fn consumer_offset_history(&self) -> Option<PollHistoryId> { + self.consumer_offset_history + } + /// Queued request on the network reply path: no in-process subscriber. #[must_use] pub const fn new(message: Message<RoutedRequestHeader>) -> Self { @@ -359,6 +374,7 @@ impl RequestEntry { ) -> Self { Self { auto_commit: None, + consumer_offset_history: None, message, received_at: 0, reply_sender, diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index c5e057881..96f9725e9 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -4548,9 +4548,13 @@ where // request room -> buffer; both full -> drop+warn (client retries // via read-timeout). if consensus.pipeline_is_full() { - let push_result = consensus.push_queued_request( - consensus::RequestEntry::with_sender(message, reply.take()), - ); + let entry = consensus::RequestEntry::with_sender(message, reply.take()); + let entry = if consumer_offset.is_some() { + entry.with_consumer_offset_history(self.poll_history) + } else { + entry + }; + let push_result = consensus.push_queued_request(entry); if let Err(mut refused) = push_result { emit_partition_diag( tracing::Level::WARN, @@ -4681,6 +4685,40 @@ where // Taken before the preflight so a refusal answers the parked waiter // instead of waking it with `Canceled`. let mut reply_sender = req.take_reply_sender(); + if req + .consumer_offset_history() + .is_some_and(|history| history != self.poll_history) + { + // A numeric offset can fit the replacement history. Refuse + // before projection, with a terminal error: transient errors + // let clients replay the old mutation into that new history. + let error = Self::parse_consumer_offset_request( + req.message.header().operation, + &req.message, + ) + .map_or( + IggyError::InvalidCommand, + |(_, consumer_id, offset, _)| { + offset.map_or_else( + || IggyError::ConsumerOffsetNotFound(consumer_id as usize), + IggyError::InvalidOffset, + ) + }, + ); + Self::send_partition_deny_or_log( + self.consensus(), + req.message.header(), + error.as_code(), + "queued offset history deny reply send failed", + reply_sender.take(), + ) + .await; + consecutive_denials += 1; + if consecutive_denials >= PROMOTION_DENIALS_MAX { + break; + } + continue; + } if !self .admit_reserved_send(&req.message, &mut reply_sender) .await diff --git a/core/partitions/tests/consumer_offset_history.rs b/core/partitions/tests/consumer_offset_history.rs new file mode 100644 index 000000000..35be0e1a1 --- /dev/null +++ b/core/partitions/tests/consumer_offset_history.rs @@ -0,0 +1,534 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use consensus::{LocalPipeline, Sequencer, VsrConsensus, oneshot_channel}; +use futures::FutureExt; +use iggy_binary_protocol::batch::{BatchIntegrity, decode_batch_slice_with}; +use iggy_binary_protocol::requests::consumer_offsets::{ + DeleteConsumerOffsetRequest, StoreConsumerOffsetRequest, +}; +use iggy_binary_protocol::{ + AckLevel, Command, Operation, PrepareOkHeader, ReplyHeader, RoutedRequestHeader, WireConsumer, + WireEncode, WireIdentifier, +}; +use iggy_common::{ConsumerKind, IggyByteSize, IggyError, PartitionStats, PollingStrategy}; +use message_bus::{AUTO_COMMIT_CLIENT_ID, IggyMessageBus}; +use partitions::{ + IggyPartition, IggyPartitions, Partition, PartitionPathLayout, PartitionsConfig, PollingArgs, + PollingConsumer, +}; +use server_common::Message; +use server_common::send_messages::{ + IggyMessage, IggyMessageHeader, IggyMessages, SendMessagesOwned, +}; +use server_common::sharding::{IggyNamespace, ShardId}; +use std::rc::Rc; +use std::sync::Arc; + +const CONSUMER_ID: u32 = 7; +const REPLACEMENT: &[u8] = b"replacement record"; +type TestPartition = IggyPartition<Rc<IggyMessageBus>>; + +#[compio::test] +async fn given_queued_offset_store_when_purged_should_not_skip_replacement_records() { + for client_id in [42, AUTO_COMMIT_CLIENT_ID] { + for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { + let (mut partition, config, _directory) = partition_with_old_record().await; + let consumer = polling_consumer(kind); + let (returned, delivered) = poll( + partition, + &config, + consumer, + &PollingArgs::new(PollingStrategy::first(), 1, false), + ); + partition = returned; + assert_eq!(delivered, vec![b"old record".to_vec()]); + assert_eq!(partition.get_consumer_offset(consumer), None); + + partition + .on_request(append(2, b"old pending prepare"), None) + .await; + assert!(partition.consensus().pipeline_is_full()); + partition.on_request(append(3, REPLACEMENT), None).await; + let (sender, receiver) = oneshot_channel(); + partition + .on_request(store(client_id, 1, kind, 0), Some(sender)) + .await; + assert_eq!(partition.consensus().request_queue_len(), 2); + + partition.purge(&config, 1).await.unwrap(); + assert_eq!(partition.purge_floor_op(), 2); + commit_pending(&mut partition, &config).await; + + assert_eq!( + partition.get_consumer_offset(consumer), + None, + "a store admitted before purge must not checkpoint replacement history ({client_id}, {kind:?})", + ); + let reply = receiver + .now_or_never() + .expect("the queued store must finish") + .unwrap(); + assert_eq!(reply.header().status, IggyError::InvalidOffset(0).as_code()); + assert_eq!( + partition.consensus().sequencer().current_sequence(), + 3, + "reject the obsolete store before assigning an operation" + ); + let (partition, next) = poll( + partition, + &config, + consumer, + &PollingArgs::new(PollingStrategy::next(), 10, false), + ); + let (_, first) = poll( + partition, + &config, + PollingConsumer::Consumer(8, 0), + &PollingArgs::new(PollingStrategy::first(), 10, false), + ); + assert_eq!( + first, + vec![REPLACEMENT.to_vec()], + "the unrelated queued publish survives" + ); + assert_eq!(next, first, "Next must not skip the replacement record"); + } + } +} + +#[compio::test] +async fn given_queued_automatic_commit_when_purged_should_keep_replacement_records_unread() { + let (mut partition, config, _directory) = partition_with_old_record().await; + partition + .on_request(append(2, b"old pending prepare"), None) + .await; + partition.on_request(append(3, REPLACEMENT), None).await; + let consumer = polling_consumer(ConsumerKind::Consumer); + let (mut partition, delivered) = poll( + partition, + &config, + consumer, + &PollingArgs::new(PollingStrategy::first(), 1, true), + ); + assert_eq!(delivered, vec![b"old record".to_vec()]); + assert_eq!(partition.consensus().request_queue_len(), 2); + partition.purge(&config, 1).await.unwrap(); + assert_eq!(partition.consensus().request_queue_len(), 1); + commit_pending(&mut partition, &config).await; + assert_eq!(partition.get_consumer_offset(consumer), None); + let (_, next) = poll( + partition, + &config, + consumer, + &PollingArgs::new(PollingStrategy::next(), 10, false), + ); + assert_eq!(next, vec![REPLACEMENT.to_vec()]); +} + +#[compio::test] +async fn given_queued_explicit_store_when_history_is_unchanged_should_allow_rewind() { + for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { + let (mut partition, config, _directory) = partition_with_old_record().await; + partition + .on_request(append(2, b"second record"), None) + .await; + commit_pending(&mut partition, &config).await; + let (sender, receiver) = oneshot_channel(); + partition + .on_request(store(42, 1, kind, 1), Some(sender)) + .await; + commit_pending(&mut partition, &config).await; + assert_success(receiver); + assert_eq!( + partition.get_consumer_offset(polling_consumer(kind)), + Some(1) + ); + + partition.on_request(append(3, b"third record"), None).await; + let (sender, receiver) = oneshot_channel(); + partition + .on_request(store(42, 2, kind, 0), Some(sender)) + .await; + assert_eq!(partition.consensus().request_queue_len(), 1); + commit_pending(&mut partition, &config).await; + assert_success(receiver); + assert_eq!( + partition.get_consumer_offset(polling_consumer(kind)), + Some(0), + "history protection must not turn an explicit store into a monotone automatic commit" + ); + } +} + +#[compio::test] +async fn given_queued_offset_delete_when_history_changes_should_reject_without_replay() { + for purge in [false, true] { + for kind in [ConsumerKind::Consumer, ConsumerKind::ConsumerGroup] { + let (mut partition, config, _directory) = partition_with_old_record().await; + let (sender, receiver) = oneshot_channel(); + partition + .on_request(store(42, 1, kind, 0), Some(sender)) + .await; + commit_pending(&mut partition, &config).await; + assert_success(receiver); + partition + .on_request(append(2, b"pending prepare"), None) + .await; + partition.on_request(append(3, REPLACEMENT), None).await; + let (sender, receiver) = oneshot_channel(); + partition.on_request(delete(kind, 2), Some(sender)).await; + assert_eq!(partition.consensus().request_queue_len(), 2); + if purge { + partition.purge(&config, 1).await.unwrap(); + } + commit_pending(&mut partition, &config).await; + let reply = receiver + .now_or_never() + .expect("the delete must finish") + .unwrap(); + assert_eq!( + reply.header().status, + if purge { + IggyError::ConsumerOffsetNotFound(CONSUMER_ID as usize).as_code() + } else { + 0 + } + ); + assert_eq!( + partition.consensus().sequencer().current_sequence(), + if purge { 4 } else { 5 } + ); + assert_eq!(partition.get_consumer_offset(polling_consumer(kind)), None); + } + } +} + +#[compio::test] +async fn given_many_obsolete_offsets_when_purged_should_finish_rejections_and_resume_queued_writes() +{ + let (mut partition, config, _directory) = partition_with_old_record().await; + partition + .on_request(append(2, b"old pending prepare"), None) + .await; + let mut receivers = Vec::new(); + for request_id in 1..=6 { + let (sender, receiver) = oneshot_channel(); + partition + .on_request( + store(42, request_id, ConsumerKind::Consumer, 0), + Some(sender), + ) + .await; + receivers.push(receiver); + } + partition.on_request(append(3, REPLACEMENT), None).await; + partition.purge(&config, 1).await.unwrap(); + commit_pending(&mut partition, &config).await; + assert!( + partition.consensus().request_queue_len() > 0, + "the rejection budget must leave work for a later owner turn" + ); + assert!(partition.queued_requests_ready()); + for _ in 0..8 { + partition.resume_queued_requests().await; + commit_pending(&mut partition, &config).await; + if partition.consensus().request_queue_len() == 0 { + break; + } + } + assert_eq!(partition.consensus().request_queue_len(), 0); + assert_eq!(partition.consensus().sequencer().current_sequence(), 3); + for receiver in receivers { + let reply = receiver + .now_or_never() + .expect("every obsolete store gets a reply") + .unwrap(); + assert_eq!(reply.header().status, IggyError::InvalidOffset(0).as_code()); + } + let (_, next) = poll( + partition, + &config, + polling_consumer(ConsumerKind::Consumer), + &PollingArgs::new(PollingStrategy::next(), 10, false), + ); + assert_eq!(next, vec![REPLACEMENT.to_vec()]); +} + +#[compio::test] +async fn given_queued_delete_when_replacement_checkpoint_commits_should_preserve_new_progress() { + let kind = ConsumerKind::Consumer; + let (mut partition, config, _directory) = partition_with_old_record().await; + let (sender, receiver) = oneshot_channel(); + partition + .on_request(store(42, 1, kind, 0), Some(sender)) + .await; + commit_pending(&mut partition, &config).await; + assert_success(receiver); + partition + .on_request(append(2, b"old pending prepare"), None) + .await; + partition.on_request(append(3, REPLACEMENT), None).await; + let mut obsolete_stores = Vec::new(); + for request_id in 2..=5 { + let (sender, receiver) = oneshot_channel(); + partition + .on_request(store(42, request_id, kind, 0), Some(sender)) + .await; + obsolete_stores.push(receiver); + } + let (sender, deleted) = oneshot_channel(); + partition.on_request(delete(kind, 6), Some(sender)).await; + partition.purge(&config, 1).await.unwrap(); + commit_pending(&mut partition, &config).await; + assert_eq!( + partition.consensus().request_queue_len(), + 1, + "the rejection budget leaves the old delete for a later turn" + ); + + // A new owner turn can admit a valid store while obsolete queued work + // waits for the next bounded promotion turn. + let (sender, stored) = oneshot_channel(); + partition + .on_request(store(43, 1, kind, 0), Some(sender)) + .await; + commit_pending(&mut partition, &config).await; + assert_success(stored); + let reply = deleted + .now_or_never() + .expect("the old delete must finish") + .unwrap(); + assert_eq!( + reply.header().status, + IggyError::ConsumerOffsetNotFound(CONSUMER_ID as usize).as_code() + ); + assert_eq!( + partition.get_consumer_offset(polling_consumer(kind)), + Some(0), + "the obsolete delete must not remove the replacement checkpoint" + ); + assert_eq!(partition.consensus().sequencer().current_sequence(), 5); + for receiver in obsolete_stores { + let reply = receiver + .now_or_never() + .expect("the obsolete store must finish") + .unwrap(); + assert_eq!(reply.header().status, IggyError::InvalidOffset(0).as_code()); + } +} + +#[allow(clippy::future_not_send)] +async fn partition_with_old_record() -> (TestPartition, PartitionsConfig, tempfile::TempDir) { + let directory = tempfile::tempdir().unwrap(); + let segment_size = IggyByteSize::from(1_048_576_u64); + let config = PartitionsConfig { + messages_required_to_save: 100, + size_of_messages_required_to_save: segment_size, + validate_checksum: true, + segment_size, + preallocate_segments: false, + encryptor: None, + path_layout: PartitionPathLayout::default(), + }; + let consensus = VsrConsensus::new( + 1, + 0, + 3, + namespace().inner(), + Rc::new(IggyMessageBus::new(0)), + LocalPipeline::with_capacities(1, 8), + ); + consensus.init(); + let mut partition = IggyPartition::with_in_memory_storage( + Arc::new(PartitionStats::default()), + consensus, + segment_size, + ); + partition.set_partition_dir(directory.path().to_string_lossy().into_owned()); + partition.on_request(append(1, b"old record"), None).await; + commit_pending(&mut partition, &config).await; + assert_eq!(partition.consensus().commit_min(), 1); + (partition, config, directory) +} + +// Drive the actual commit and queue-promotion paths with deterministic quorum +// completion. Remote replicas and timing do not decide when a prepare advances. +#[allow(clippy::future_not_send)] +async fn commit_pending(partition: &mut TestPartition, config: &PartitionsConfig) { + for _ in 0..16 { + process_self_acknowledgments(partition, config).await; + let current = partition.consensus().sequencer().current_sequence(); + if partition.consensus().commit_min() == current { + return; + } + partition.consensus().advance_commit_max(current); + partition.commit_journal(config).await; + } + panic!("the bounded request queue did not drain"); +} + +#[allow(clippy::future_not_send)] +async fn process_self_acknowledgments(partition: &mut TestPartition, config: &PartitionsConfig) { + let mut queued = Vec::new(); + partition.consensus().drain_loopback_into(&mut queued); + for message in queued { + let mut typed = Message::<PrepareOkHeader>::new(message.as_slice().len()); + typed.as_mut_slice().copy_from_slice(message.as_slice()); + partition.on_ack(typed, config).await; + } +} + +fn poll( + partition: TestPartition, + config: &PartitionsConfig, + consumer: PollingConsumer, + args: &PollingArgs, +) -> (TestPartition, Vec<Vec<u8>>) { + let owner = IggyPartitions::new(ShardId::new(0), config.clone()); + owner.insert(namespace(), partition); + let read = owner + .build_poll_snapshot(&namespace(), consumer, args) + .unwrap() + .execute_resident(); + let completion = owner.complete_poll(&namespace(), read).unwrap(); + assert!( + completion.replication.is_none(), + "automatic commits in this fixture must queue" + ); + let payloads = completion + .fragments + .iter() + .flat_map(|fragment| { + let batch = + decode_batch_slice_with(fragment.as_slice(), BatchIntegrity::Verify).unwrap(); + batch + .iter_with_offsets() + .map(|record| record.message.payload.to_vec()) + .collect::<Vec<_>>() + }) + .collect(); + (owner.remove(&namespace()).unwrap(), payloads) +} + +fn namespace() -> IggyNamespace { + IggyNamespace::new(1, 1, 0) +} + +const fn polling_consumer(kind: ConsumerKind) -> PollingConsumer { + match kind { + ConsumerKind::Consumer => PollingConsumer::Consumer(CONSUMER_ID as usize, 0), + ConsumerKind::ConsumerGroup => PollingConsumer::ConsumerGroup(CONSUMER_ID as usize, 0), + } +} + +fn append(request: u64, payload: &[u8]) -> Message<RoutedRequestHeader> { + let mut messages = IggyMessages::with_capacity(1); + messages.push(IggyMessage { + header: IggyMessageHeader { + id: u128::from(request), + payload_length: u32::try_from(payload.len()).unwrap(), + ..Default::default() + }, + payload: payload.to_vec().into(), + user_headers: None, + }); + SendMessagesOwned::from_messages(namespace(), &messages) + .unwrap() + .encode_request(RoutedRequestHeader { + command: Command::Request, + operation: Operation::SendMessages, + client: 1, + session: 1, + request, + group: namespace().inner(), + ..Default::default() + }) + .unwrap() +} + +fn store( + client: u128, + request: u64, + kind: ConsumerKind, + offset: u64, +) -> Message<RoutedRequestHeader> { + let body = StoreConsumerOffsetRequest { + consumer: WireConsumer { + kind: kind.as_code(), + id: WireIdentifier::numeric(CONSUMER_ID), + }, + stream_id: WireIdentifier::numeric(1), + topic_id: WireIdentifier::numeric(1), + partition_id: Some(0), + offset, + ack: AckLevel::Quorum, + } + .to_bytes(); + let size = size_of::<RoutedRequestHeader>() + body.len(); + let mut message = Message::<RoutedRequestHeader>::new(size); + message.as_mut_slice()[size_of::<RoutedRequestHeader>()..].copy_from_slice(&body); + message.transmute_header(|_, header: &mut RoutedRequestHeader| { + *header = RoutedRequestHeader { + command: Command::Request, + operation: Operation::StoreConsumerOffset, + size: u32::try_from(size).unwrap(), + client, + session: 1, + request, + group: namespace().inner(), + ..Default::default() + }; + }) +} + +fn delete(kind: ConsumerKind, request: u64) -> Message<RoutedRequestHeader> { + let body = DeleteConsumerOffsetRequest { + consumer: WireConsumer { + kind: kind.as_code(), + id: WireIdentifier::numeric(CONSUMER_ID), + }, + stream_id: WireIdentifier::numeric(1), + topic_id: WireIdentifier::numeric(1), + partition_id: Some(0), + ack: AckLevel::Quorum, + } + .to_bytes(); + let size = size_of::<RoutedRequestHeader>() + body.len(); + let mut message = Message::<RoutedRequestHeader>::new(size); + message.as_mut_slice()[size_of::<RoutedRequestHeader>()..].copy_from_slice(&body); + message.transmute_header(|_, header: &mut RoutedRequestHeader| { + *header = RoutedRequestHeader { + command: Command::Request, + operation: Operation::DeleteConsumerOffset, + size: u32::try_from(size).unwrap(), + client: 42, + session: 1, + request, + group: namespace().inner(), + ..Default::default() + }; + }) +} + +fn assert_success(receiver: consensus::Receiver<Message<ReplyHeader>>) { + let reply = receiver + .now_or_never() + .expect("the store must finish") + .unwrap(); + assert_eq!(reply.header().status, 0); +} diff --git a/core/sdk/src/poll_routing.rs b/core/sdk/src/poll_routing.rs index 5056e775e..d73c4a1ca 100644 --- a/core/sdk/src/poll_routing.rs +++ b/core/sdk/src/poll_routing.rs @@ -957,6 +957,38 @@ mod tests { } } + #[tokio::test] + async fn given_terminal_offset_errors_when_routed_should_not_recover_or_replay() { + for (code, error) in [ + (STORE_CONSUMER_OFFSET_CODE, IggyError::InvalidOffset(10)), + ( + DELETE_CONSUMER_OFFSET_CODE, + IggyError::ConsumerOffsetNotFound(7), + ), + ] { + let expected_code = error.as_code(); + let (router, coordinator, request) = fixture([ + ( + Channel::Coordinator, + GET_CONSUMER_OFFSET_ROUTING_CODE, + Ok(routing()), + ), + ( + Channel::Data, + ATTACH_CONSUMER_SESSION_CODE, + Ok(Bytes::new()), + ), + (Channel::Data, code, Err(error)), + ]); + let error = routed_request(&router, &coordinator, &request, code) + .await + .expect_err("obsolete offset mutations must be returned to the caller"); + assert_eq!(error.as_code(), expected_code); + assert_eq!(coordinator.script.connections.load(Ordering::Relaxed), 1); + assert!(coordinator.script.exchanges.lock().unwrap().is_empty()); + } + } + #[tokio::test] async fn acknowledged_metadata_changes_refresh_cached_routes_and_attachments() { let (router, coordinator, request) = fixture([ diff --git a/core/simulator/src/consumer_offset_history_tests.rs b/core/simulator/src/consumer_offset_history_tests.rs new file mode 100644 index 000000000..a4a7f476b --- /dev/null +++ b/core/simulator/src/consumer_offset_history_tests.rs @@ -0,0 +1,220 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Queue-history coverage under seeded scheduling and delayed replica replies. +//! The partition integration test covers actual purge. Here the existing log +//! reload seam changes history while real shard queues and replication run. + +use crate::Simulator; +use crate::client::SimClient; +use crate::packet::{Packet, PacketSimulatorOptions, ProcessId}; +use crate::tests::submit_and_wait_for_reply; +use bytes::Bytes; +use consensus::{PIPELINE_PREPARE_QUEUE_MAX, PartitionsHandle, Sequencer}; +use iggy_binary_protocol::{AckLevel, Command, PrepareOkHeader}; +use iggy_common::{ConsumerKind, IggyByteSize, IggyError}; +use partitions::{Partition, PollingConsumer, RetainedPartitionState}; +use server_common::sharding::IggyNamespace; +use std::mem; + +#[test] +fn given_queued_offset_when_owner_history_is_reloaded_should_reject_across_seeded_schedules() { + for seed in 0x0FF5_E700..0x0FF5_E708 { + let first = history_reload_trace(seed); + let replay = history_reload_trace(seed); + assert_eq!( + first, replay, + "seed {seed:#x} must replay the same schedule" + ); + } +} + +#[allow(clippy::too_many_lines)] +fn history_reload_trace(seed: u64) -> u64 { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: IggyByteSize::from(0_u64), + bucket_capacity: 1, + }); + let namespace = IggyNamespace::new(1, 1, 0); + let mut simulator = Simulator::with_shards( + 3, + 2, + [1, 2].into_iter(), + PacketSimulatorOptions { + node_count: 3, + client_count: 2, + seed, + ..PacketSimulatorOptions::default() + }, + ); + simulator.init_partition(namespace); + let producer = SimClient::new(1); + let consumer = SimClient::new(2); + simulator.register_client_with_primary(&producer); + simulator.register_client_with_primary(&consumer); + let reply = submit_and_wait_for_reply( + &mut simulator, + 1, + 0, + producer.send_messages(namespace, &[Bytes::from_static(b"old record")]), + ); + assert_eq!(reply.header().status, 0); + for _ in 0..20 { + simulator.step(); + } + + // Keep the primary alive and its prepare pipeline full by withholding only + // this partition's remote acknowledgments. Other control traffic still runs. + for backup in 1..3 { + *simulator + .network + .link_drop_packet_fn(ProcessId::Replica(backup), ProcessId::Replica(0)) = + Some(drop_partition_acknowledgment); + } + for _ in 0..PIPELINE_PREPARE_QUEUE_MAX { + simulator.submit_request( + 1, + 0, + producer + .send_messages(namespace, &[Bytes::from_static(b"pending record")]) + .into_generic(), + ); + } + assert!( + (0..100).any(|_| { + simulator.step(); + simulator.replicas[0] + .partition_shard(namespace) + .plane + .partitions() + .with_partition(&namespace, |partition| { + partition.consensus().pipeline_is_full() + }) + == Some(true) + }), + "seed {seed:#x}: the prepare pipeline never filled" + ); + let request = consumer.store_consumer_offset( + namespace, + ConsumerKind::Consumer.as_code(), + 7, + 0, + AckLevel::Quorum, + ); + let request_id = request.header().request; + simulator.submit_request(2, 0, request.into_generic()); + assert!( + (0..100).any(|_| { + simulator.step(); + simulator.replicas[0] + .partition_shard(namespace) + .plane + .partitions() + .with_partition(&namespace, |partition| { + partition.consensus().request_queue_len() + }) + == Some(1) + }), + "seed {seed:#x}: the offset store never entered the request queue" + ); + simulator.run_pumps(); + + // Reinstall the retained log only while every pump is quiescent. This + // production helper invalidates the same history identity as purge, while + // retaining bytes avoids real filesystem timing in the seeded executor. + let owner = simulator.replicas[0].partition_shard(namespace); + let partition = owner.plane.partitions().get_mut_by_ns(&namespace).unwrap(); + let offsets = partition.offsets(); + let retained = RetainedPartitionState { + consumer_offsets: partition.retained_consumer_offsets(ConsumerKind::Consumer), + consumer_group_offsets: partition.retained_consumer_offsets(ConsumerKind::ConsumerGroup), + durable_offset: offsets.commit_offset, + write_offset: offsets.write_offset, + offset_space_used: partition.offset_space_used(), + log: mem::take(&mut partition.log), + }; + partition.adopt_retained_log(retained); + let expected_operation = partition.consensus().sequencer().current_sequence(); + assert_eq!( + partition.consensus().view(), + 0, + "the test must not recover through election" + ); + + for backup in 1..3 { + *simulator + .network + .link_drop_packet_fn(ProcessId::Replica(backup), ProcessId::Replica(0)) = None; + } + let mut observed_status = None; + let converged = (0..800).any(|_| { + for reply in simulator.step() { + if reply.header().client == 2 && reply.header().request == request_id { + observed_status = Some(reply.header().status); + } + } + observed_status.is_some() + && simulator.replicas.iter().all(|replica| { + replica + .partition_shard(namespace) + .plane + .partitions() + .with_partition(&namespace, |partition| { + partition.consensus().commit_min() == expected_operation + }) + == Some(true) + }) + }); + assert_eq!( + observed_status, + Some(IggyError::InvalidOffset(0).as_code()), + "seed {seed:#x}: obsolete progress must return a terminal rejection" + ); + assert!( + converged, + "seed {seed:#x}: unrelated prepared writes must converge" + ); + for replica in &simulator.replicas { + let partitions = replica.partition_shard(namespace).plane.partitions(); + assert_eq!( + partitions + .consumer_offset_read(&namespace, PollingConsumer::Consumer(7, 0)) + .unwrap() + .0, + None + ); + assert_eq!( + partitions.with_partition(&namespace, |partition| partition + .consensus() + .sequencer() + .current_sequence()), + Some(expected_operation), + "the rejected offset must not receive a replicated operation" + ); + } + simulator.executor.schedule_hash() +} + +fn drop_partition_acknowledgment(packet: &Packet) -> bool { + if packet.message.header().command != Command::PrepareOk { + return false; + } + let header: &PrepareOkHeader = + bytemuck::checked::from_bytes(&packet.message.as_slice()[..size_of::<PrepareOkHeader>()]); + header.group == IggyNamespace::new(1, 1, 0).inner() +} diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index cae0418c3..5acb01651 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -1681,6 +1681,9 @@ fn materialise_partition( } } +#[cfg(test)] +mod consumer_offset_history_tests; + #[cfg(test)] mod tests { use super::*;
