You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2026/09/28 20:55:31 UTC
(iggy) 01/01: fix(partitions): reject queued offsets after history resets
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 <pi...@gmail.com>
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::*;