This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/partition-routing in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 2bfbbaef0cd67ad0800c0208fa968198e7fb10e1 Author: Hubert Gruszecki <[email protected]> AuthorDate: Tue Oct 13 22:33:52 2026 +0200 test(simulator): exercise stale owners and strict offsets Random offsets did not exercise captured stale owners, and mixed-ack runs skipped comparison with the acknowledged offset shadow. Add a bounded lifecycle action and a same-session generation regression, check auto-commit generations and owner retirement, and require nonzero strict-shadow coverage in the serial quorum CI case. --- core/simulator/src/bin/workload-fuzz.rs | 5 +- core/simulator/src/lib.rs | 32 +++- core/simulator/src/workload/actions.rs | 1 + core/simulator/src/workload/mod.rs | 195 +++++++++++++++++++ core/simulator/src/workload/ops/mod.rs | 2 + .../src/workload/ops/store_stale_owner_offset.rs | 84 +++++++++ core/simulator/src/workload/options.rs | 4 +- core/simulator/src/workload/oracle.rs | 15 +- core/simulator/src/workload/shadow.rs | 65 ++++++- core/simulator/src/workload/state_checker.rs | 208 +++++++++++++++++++-- 10 files changed, 593 insertions(+), 18 deletions(-) diff --git a/core/simulator/src/bin/workload-fuzz.rs b/core/simulator/src/bin/workload-fuzz.rs index bbb1e184e..c8cab28e2 100644 --- a/core/simulator/src/bin/workload-fuzz.rs +++ b/core/simulator/src/bin/workload-fuzz.rs @@ -610,7 +610,7 @@ fn run_quiesce_phase( "quiesced and converged (leader-relative; entity oracle: {entity_oracle}; \ evictions={}; ops_compared={} partition_ops_compared={} \ deleted_partition_ops_compared={} replicas_compared={} namespaces_checked={} \ - owners_compared={})", + owners_compared={} offsets_compared_with_shadow={} stale_owner_stores={} owner_fence_refusals={})", workload.evictions(), convergence.ops_compared, convergence.partition_ops_compared, @@ -618,6 +618,9 @@ fn run_quiesce_phase( convergence.replicas_compared, convergence.namespaces_checked, convergence.owners_compared, + convergence.offsets_compared_with_shadow, + workload.stale_owner_stores(), + workload.owner_fence_refusals(), ); if args.require_entity_oracle && !workload.strict_outcome_oracle() { vacuous(format_args!( diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index d53ca225d..0290f5f05 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -7143,6 +7143,15 @@ mod tests { /// denied. #[test] fn shell_workload_drains_and_converges() { + shell_workload_with_ack_quorum_ratio(0.5); + } + + #[test] + fn given_serial_quorum_offsets_when_shell_workload_quiesces_should_compare_the_shadow() { + shell_workload_with_ack_quorum_ratio(1.0); + } + + fn shell_workload_with_ack_quorum_ratio(ack_quorum_ratio: f32) { use crate::workload::{ self, Workload, options::{ActionWeights, WorkloadOptions}, @@ -7182,6 +7191,7 @@ mod tests { let mut options = WorkloadOptions::new(seed, replica_count, vec![ns]); options.weights = ActionWeights::uniform(); + options.ack_quorum_ratio = ack_quorum_ratio; let mut wl = Workload::new(options); let clients = [client]; @@ -7213,7 +7223,27 @@ mod tests { "{}", oracle::quiesce_failure_report(&sim, &wl), ); - oracle::assert_converged(&sim, &mut wl); + let convergence = oracle::assert_converged(&sim, &mut wl); + if ack_quorum_ratio >= 1.0 { + assert!( + wl.auditor + .stats() + .commits(crate::workload::actions::Action::StoreConsumerOffset) + > 0, + "the strict CI cell must acknowledge an offset store" + ); + assert!( + convergence.offsets_compared_with_shadow > 0, + "the strict CI cell must compare its acknowledged offset shadow" + ); + println!( + "strict offset proof: stores={}, offsets_compared_with_shadow={}", + wl.auditor + .stats() + .commits(crate::workload::actions::Action::StoreConsumerOffset), + convergence.offsets_compared_with_shadow, + ); + } } /// A result-framed transport rejection is not a committed result. diff --git a/core/simulator/src/workload/actions.rs b/core/simulator/src/workload/actions.rs index 323a0c8a7..7bbde7d5b 100644 --- a/core/simulator/src/workload/actions.rs +++ b/core/simulator/src/workload/actions.rs @@ -51,6 +51,7 @@ pub enum Action { ReplaceFixtureTopic, JoinFixtureGroup, LeaveFixtureGroup, + StoreStaleOwnerOffset, } // Lock the discriminants of the first three variants. Inserting a new variant before diff --git a/core/simulator/src/workload/mod.rs b/core/simulator/src/workload/mod.rs index c64422898..cd60ecb5b 100644 --- a/core/simulator/src/workload/mod.rs +++ b/core/simulator/src/workload/mod.rs @@ -43,7 +43,9 @@ use crate::workload::ops::{ }; use actions::Action; use auditor::{OnReply, ServerAuditor}; +use consensus::MetadataHandle; use effect::{Effect, SimCommand}; +use iggy_binary_protocol::primitives::partition_history::PartitionContext; use iggy_binary_protocol::{ KIND_CONSUMER_GROUP, Operation, ReplyHeader, RoutedRequestHeader, result_code, }; @@ -161,6 +163,8 @@ pub struct Workload { /// Partition ops of deleted fixture incarnations witnessed by more than one /// replica, for the run summary. deleted_partition_ops_compared: usize, + stale_owner_stores: u64, + owner_fence_refusals: u64, /// Clients whose recovery or redirect handshake is in flight, with the replica /// and the tick it last went out on. Such a client submits nothing: its session /// is gone or moving, so anything it sent would be refused. @@ -207,6 +211,8 @@ impl Workload { evictions: 0, redirects: 0, deleted_partition_ops_compared: 0, + stale_owner_stores: 0, + owner_fence_refusals: 0, pending_handshakes: BTreeMap::new(), homes: BTreeMap::new(), samples_none: 0, @@ -555,6 +561,13 @@ impl Workload { self.samples_none += 1; return None; }; + if let InFlightInput::StoreStaleOwnerOffset(input) = &input { + if input.captured.owner.client_id != client.client_id() { + self.samples_none += 1; + return None; + } + self.stale_owner_stores += 1; + } let message = ops::build_message(client, &input); let header = message.header(); @@ -586,6 +599,43 @@ impl Workload { Some((target, message)) } + #[must_use] + pub const fn stale_owner_stores(&self) -> u64 { + self.stale_owner_stores + } + + #[must_use] + pub const fn owner_fence_refusals(&self) -> u64 { + self.owner_fence_refusals + } + + fn observe_fixture_owners(&mut self, sim: &Simulator) { + if self.options.weights.weight(Action::StoreStaleOwnerOffset) == 0 { + return; + } + for &namespace in &self.options.namespaces { + let Some(primary) = sim.primary_index(namespace).map(usize::from) else { + continue; + }; + let Some(incarnation) = sim.partition_incarnation(primary, namespace) else { + continue; + }; + let owner = sim + .installed_group_owner(primary, namespace, u64::from(shadow::FIXTURE_GROUP_ID)) + .unwrap_or_default(); + let context = PartitionContext { + incarnation, + owner_generation: owner.generation, + metadata_op: sim.replicas[primary].shards[0] + .plane + .metadata() + .applied_frontier() + .get(), + }; + self.shadow.observe_owner(namespace, owner, context); + } + } + /// Validate and apply a reply. Returns [`SimCommand`]s the driver /// must run against the simulator (e.g. `init_partition`); the /// auditor stays transport-agnostic. @@ -616,6 +666,18 @@ impl Workload { OnReply::Unknown => return Vec::new(), }; let namespace = IggyNamespace::from_raw(entry.request_namespace); + if entry.action == Action::StoreStaleOwnerOffset { + assert_ne!( + header.status, 0, + "a captured stale owner committed an offset: input={:?}, request={}, seed={:#x}", + entry.input, header.request, self.options.seed, + ); + if header.status + == IggyError::ConsumerGroupPartitionNotOwned(shadow::FIXTURE_GROUP_ID, 0).as_code() + { + self.owner_fence_refusals += 1; + } + } let attempts = self.outstanding.get(&key).map_or(1, |entry| entry.attempts); // A refusal proves only that its own attempt did not commit. let refusal = if attempts > 1 { @@ -1095,6 +1157,7 @@ fn drive_tick( // slot, so `build_request` would decline it anyway. resubmit_due(sim, workload); resubmit_due_handshakes(sim, workload, clients); + workload.observe_fixture_owners(sim); for client in clients { // A client whose session is gone submits nothing: the request would be // refused and evict it again, and the eviction would look like a fresh @@ -1429,6 +1492,7 @@ mod tests { use crate::partition_repair_driver_tests::is_commit_for; use consensus::{MetadataHandle, StateTransferStage, Status}; use iggy_binary_protocol::namespace::METADATA_GROUP; + use iggy_binary_protocol::primitives::partition_history::ConsumerGroupOwner; use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; use iggy_binary_protocol::{Command, WireDecode}; use metadata::impls::metadata::StreamsFrontend; @@ -3134,4 +3198,135 @@ mod tests { "no settled owner was compared" ); } + + #[test] + #[allow(clippy::too_many_lines)] + fn given_a_recaptured_assignment_when_storing_with_its_old_owner_should_refuse_in_the_workload() + { + const SEED: u64 = 0x57A1_E001; + const REPLICAS: u8 = 3; + const TRAFFIC_TICKS: u64 = 200; + let namespace = IggyNamespace::new(1, 1, 0); + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + let mut sim = Simulator::with_shards_shell( + usize::from(REPLICAS), + 1, + [1].into_iter(), + PacketSimulatorOptions { + node_count: REPLICAS, + client_count: 1, + seed: SEED, + ..Default::default() + }, + ); + sim.init_partition(namespace); + sim.seed_stream_topic_partition(namespace); + let clients = [SimClient::new(1)]; + sim.shell_login(&clients[0]); + let mut options = WorkloadOptions::new(SEED, REPLICAS, vec![namespace]); + options.ack_quorum_ratio = 1.0; + options.weights = ActionWeights::lifecycle(); + let mut workload = Workload::new(options); + let mut invariants = Invariants::new(); + create_fixture_groups(&mut sim, &mut workload, &clients[0]); + workload.options.weights = ActionWeights::new(&[(Action::SendMessages, 100)]); + run( + &mut sim, + &mut workload, + &clients, + TRAFFIC_TICKS, + u64::MAX, + &mut invariants, + ); + assert!(oracle::drive_to_quiesce( + &mut sim, + &mut workload, + &clients, + u64::from(crate::SETUP_TOTAL_STEPS), + &mut invariants, + )); + assert!(workload.shadow.sends_committed(namespace) > 0); + workload.options.weights = ActionWeights::new(&[(Action::StoreStaleOwnerOffset, 100)]); + let (stream, topic) = workload.shadow.fixture_names(0); + let mut original = None; + for join in [true, false, true] { + let request = if join { + clients[0].join_consumer_group(&stream, &topic, shadow::FIXTURE_GROUP_NAME) + } else { + clients[0].leave_consumer_group(&stream, &topic, shadow::FIXTURE_GROUP_NAME) + }; + let operation = request.header().operation; + let reply = sim.await_setup_reply( + &clients[0], + 0, + &request.into_generic(), + operation, + "owner handoff", + ); + assert_eq!(result_code(reply.body()), Some(0)); + assert!( + (0..crate::SETUP_TOTAL_STEPS).any(|_| { + sim.step(); + sim.installed_group_owner(0, namespace, u64::from(shadow::FIXTURE_GROUP_ID)) + .is_some_and(|owner| { + owner.is_unassigned() != join + && original.is_none_or(|old: ConsumerGroupOwner| { + owner.generation > old.generation + }) + }) + }), + "the requested owner handoff must install" + ); + if original.is_none() { + original = + sim.installed_group_owner(0, namespace, u64::from(shadow::FIXTURE_GROUP_ID)); + } + workload.observe_fixture_owners(&sim); + } + let current = sim + .installed_group_owner(0, namespace, u64::from(shadow::FIXTURE_GROUP_ID)) + .unwrap(); + let original = original.unwrap(); + assert_eq!(current.client_id, original.client_id); + assert_eq!(current.session, original.session); + assert!(current.generation > original.generation); + + run( + &mut sim, + &mut workload, + &clients, + TRAFFIC_TICKS, + u64::MAX, + &mut invariants, + ); + assert!(oracle::drive_to_quiesce( + &mut sim, + &mut workload, + &clients, + u64::from(crate::SETUP_TOTAL_STEPS), + &mut invariants, + )); + assert!( + workload.stale_owner_stores() > 0, + "no captured stale owner was submitted" + ); + assert!( + workload.owner_fence_refusals() > 0, + "no captured stale owner reached its fence" + ); + let convergence = oracle::assert_converged(&sim, &mut workload); + assert!(convergence.owners_compared > 0); + assert!(convergence.offsets_compared_with_shadow > 0); + println!( + "stale owner proof: submitted={}, fence_refusals={}, owners_compared={}, offsets_compared_with_shadow={}", + workload.stale_owner_stores(), + workload.owner_fence_refusals(), + convergence.owners_compared, + convergence.offsets_compared_with_shadow, + ); + } } diff --git a/core/simulator/src/workload/ops/mod.rs b/core/simulator/src/workload/ops/mod.rs index d046592dc..1424d60dc 100644 --- a/core/simulator/src/workload/ops/mod.rs +++ b/core/simulator/src/workload/ops/mod.rs @@ -49,6 +49,7 @@ pub mod leave_fixture_group; pub mod replace_fixture_topic; pub mod send_messages; pub mod store_consumer_offset; +pub mod store_stale_owner_offset; pub mod update_permissions; pub mod update_stream; pub mod update_topic; @@ -213,4 +214,5 @@ op_dispatch! { ReplaceFixtureTopic => replace_fixture_topic, JoinFixtureGroup => join_fixture_group, LeaveFixtureGroup => leave_fixture_group, + StoreStaleOwnerOffset => store_stale_owner_offset, } diff --git a/core/simulator/src/workload/ops/store_stale_owner_offset.rs b/core/simulator/src/workload/ops/store_stale_owner_offset.rs new file mode 100644 index 000000000..c46a79019 --- /dev/null +++ b/core/simulator/src/workload/ops/store_stale_owner_offset.rs @@ -0,0 +1,84 @@ +// 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. + +//! An offset store carrying authority captured before an owner handoff. + +use crate::client::SimClient; +use crate::workload::effect::Effect; +use crate::workload::options::WorkloadOptions; +use crate::workload::shadow::{CapturedOwner, FIXTURE_GROUP_ID, Shadow}; +use iggy_binary_protocol::{AckLevel, KIND_CONSUMER_GROUP, RoutedRequestHeader}; +use rand::RngExt; +use rand_xoshiro::Xoshiro256PlusPlus; +use server_common::Message; + +#[derive(Debug, Clone)] +pub struct Input { + pub captured: CapturedOwner, + pub offset: u64, +} + +#[derive(Copy, Clone, Eq, PartialEq, Debug)] +pub enum Outcome { + Refused, +} + +pub const OUTCOMES: &[Outcome] = &[Outcome::Refused]; + +pub fn sample( + shadow: &mut Shadow, + _outcome: Outcome, + prng: &mut Xoshiro256PlusPlus, + _options: &WorkloadOptions, +) -> Option<Input> { + let captured = shadow.pick_stale_owner(prng)?; + let high = shadow.sends_committed(captured.namespace); + (high > 0).then(|| Input { + captured, + offset: prng.random_range(0..high), + }) +} + +#[must_use] +pub fn build_message(client: &SimClient, input: &Input) -> Message<RoutedRequestHeader> { + let captured = input.captured; + client + .store_consumer_offset( + captured.namespace, + KIND_CONSUMER_GROUP, + FIXTURE_GROUP_ID, + input.offset, + AckLevel::Quorum, + ) + .transmute_header(|mut source, target: &mut RoutedRequestHeader| { + source.partition_incarnation = captured.context.incarnation; + source.owner_generation = captured.owner.generation; + source.minimum_metadata_op = captured.context.metadata_op; + source.session = captured.owner.session; + *target = source; + }) +} + +#[must_use] +pub const fn classify_reply(_code: u32) -> Outcome { + Outcome::Refused +} + +#[must_use] +pub const fn predicted_effect(_input: &Input, _outcome: Outcome) -> Effect { + Effect::None +} diff --git a/core/simulator/src/workload/options.rs b/core/simulator/src/workload/options.rs index 85a36ccd7..ad400e828 100644 --- a/core/simulator/src/workload/options.rs +++ b/core/simulator/src/workload/options.rs @@ -45,6 +45,7 @@ const UNIFORM_EXCLUDED: &[Action] = &[ Action::ReplaceFixtureTopic, Action::JoinFixtureGroup, Action::LeaveFixtureGroup, + Action::StoreStaleOwnerOffset, ]; /// Per-action sampling weights as percentages. Unlisted variants default @@ -94,11 +95,12 @@ impl ActionWeights { pub fn lifecycle() -> Self { Self::new(&[ (Action::SendMessages, 42), - (Action::StoreConsumerOffset, 25), + (Action::StoreConsumerOffset, 20), (Action::DeleteConsumerOffset, 8), (Action::ReplaceFixtureTopic, 5), (Action::JoinFixtureGroup, 12), (Action::LeaveFixtureGroup, 8), + (Action::StoreStaleOwnerOffset, 5), ]) } diff --git a/core/simulator/src/workload/oracle.rs b/core/simulator/src/workload/oracle.rs index e13b9b01a..ff49428a0 100644 --- a/core/simulator/src/workload/oracle.rs +++ b/core/simulator/src/workload/oracle.rs @@ -563,7 +563,7 @@ pub fn assert_converged(sim: &Simulator, workload: &mut Workload) -> Convergence let replicas_compared = assert_committed_metadata_agrees(sim, &live, seed); let owners_compared = assert_settled_owners_installed(sim, workload, leader, seed); - assert_partition_offsets_agree(sim, workload, &live, seed); + let offsets_compared_with_shadow = assert_partition_offsets_agree(sim, workload, &live, seed); assert_no_stale_revocations(sim, leader, seed); let report = ConvergenceReport { @@ -572,6 +572,7 @@ pub fn assert_converged(sim: &Simulator, workload: &mut Workload) -> Convergence replicas_compared, namespaces_checked, owners_compared, + offsets_compared_with_shadow, }; // Entity oracle: on a serial run the shadow must equal the committed @@ -635,6 +636,8 @@ pub struct ConvergenceReport { /// Consumer group owners settled in leader metadata that were compared with /// the owner their partition primary installed. pub owners_compared: usize, + /// Partition offset tables compared with the serial acknowledged shadow. + pub offsets_compared_with_shadow: usize, } /// Require a host and exactly one primary for every workload namespace committed @@ -848,7 +851,13 @@ fn assert_settled_owners_installed( compared } -fn assert_partition_offsets_agree(sim: &Simulator, workload: &Workload, live: &[usize], seed: u64) { +fn assert_partition_offsets_agree( + sim: &Simulator, + workload: &Workload, + live: &[usize], + seed: u64, +) -> usize { + let mut compared_with_shadow = 0; for &namespace in &workload.options.namespaces { let mut committed = BTreeMap::new(); for &replica in live { @@ -892,10 +901,12 @@ fn assert_partition_offsets_agree(sim: &Simulator, workload: &Workload, live: &[ offsets, expected, "consumer offsets differ from the serial acknowledged shadow (ns={namespace:?}, seed={seed:#x})" ); + compared_with_shadow += 1; } committed.insert(prefix, (replica, offsets)); } } + compared_with_shadow } /// Fail if a revocation pending in the metadata of replica `leader` aged past diff --git a/core/simulator/src/workload/shadow.rs b/core/simulator/src/workload/shadow.rs index 2752b02f8..eb961daa9 100644 --- a/core/simulator/src/workload/shadow.rs +++ b/core/simulator/src/workload/shadow.rs @@ -26,11 +26,12 @@ //! `IndexSet` for state we sample randomly (O(1) `get_index`). //! `HashMap` for pure-lookup state. +use iggy_binary_protocol::primitives::partition_history::{ConsumerGroupOwner, PartitionContext}; use indexmap::IndexSet; use metadata::stm::stream::Streams; use rand::RngExt; use rand_xoshiro::Xoshiro256PlusPlus; -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use server_common::sharding::IggyNamespace; @@ -45,6 +46,18 @@ pub const FIXTURE_GROUP_NAME: &str = "sim-fixture-group"; /// incarnation numbers its groups from 0. pub const FIXTURE_GROUP_ID: u32 = 0; +#[derive(Debug, Clone, Copy)] +pub struct CapturedOwner { + pub namespace: IggyNamespace, + pub owner: ConsumerGroupOwner, + pub context: PartitionContext, +} + +struct ObservedOwner { + captured: CapturedOwner, + current: ConsumerGroupOwner, +} + /// The half of a fixture replacement one request performs. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Toggle { @@ -243,6 +256,9 @@ pub struct Shadow { /// Namespaces of the fixture topics whose delete was reached since the last /// [`Self::take_deleted_namespaces`]. deleted_namespaces: Vec<IggyNamespace>, + /// One original token per fixture/client; only input selection reads these + /// observations. The independent committed-log oracle validates effects. + captured_owners: BTreeMap<(IggyNamespace, u128), ObservedOwner>, } impl Shadow { @@ -302,7 +318,52 @@ impl Shadow { fixtures, fixture_deletes_open: true, deleted_namespaces: Vec::new(), + captured_owners: BTreeMap::new(), + } + } + + pub(crate) fn observe_owner( + &mut self, + namespace: IggyNamespace, + owner: ConsumerGroupOwner, + context: PartitionContext, + ) { + self.captured_owners.retain(|&(ns, _), observed| { + ns != namespace || observed.captured.context.incarnation == context.incarnation + }); + for (&(ns, _), observed) in &mut self.captured_owners { + if ns == namespace { + observed.current = owner; + } + } + if !owner.is_unassigned() { + self.captured_owners + .entry((namespace, owner.client_id)) + .or_insert(ObservedOwner { + captured: CapturedOwner { + namespace, + owner, + context, + }, + current: owner, + }); + } + } + + pub(crate) fn pick_stale_owner(&self, prng: &mut Xoshiro256PlusPlus) -> Option<CapturedOwner> { + let stale = || { + self.captured_owners.values().filter(|observed| { + observed.current.generation > observed.captured.owner.generation + && self.namespaces_live.contains(&observed.captured.namespace) + }) + }; + let count = stale().count(); + if count == 0 { + return None; } + stale() + .nth(prng.random_range(0..count)) + .map(|observed| observed.captured) } /// Start a toggle of a fixture topic picked uniformly among those that can @@ -384,6 +445,8 @@ impl Shadow { .retain(|namespace, _| !namespaces.contains(namespace)); self.consumer_offsets .retain(|(namespace, _, _), _| !namespaces.contains(namespace)); + self.captured_owners + .retain(|(namespace, _), _| !namespaces.contains(namespace)); self.deleted_namespaces.extend(namespaces); if let Some(group) = &mut self.fixtures[index].group { group.exists = false; diff --git a/core/simulator/src/workload/state_checker.rs b/core/simulator/src/workload/state_checker.rs index 7dded9dc0..01afbe208 100644 --- a/core/simulator/src/workload/state_checker.rs +++ b/core/simulator/src/workload/state_checker.rs @@ -42,7 +42,9 @@ use iggy_binary_protocol::primitives::partition_history::ConsumerGroupOwner; use iggy_binary_protocol::requests::consumer_offsets::{ GetConsumerOffsetRequest, split_prepared_offset, }; -use iggy_binary_protocol::requests::partitions::InstallConsumerGroupOwnerRequest; +use iggy_binary_protocol::requests::partitions::{ + InstallConsumerGroupOwnerRequest, RetireConsumerGroupOwnersRequest, +}; use iggy_binary_protocol::{HEADER_SIZE, Operation, PrepareHeader, WireDecode}; use journal::Journal; use message_bus::is_auto_commit_client; @@ -76,7 +78,13 @@ struct PartitionProgress { incarnation: u64, replica_incarnation: u128, committed: u64, - owners: BTreeMap<u64, ConsumerGroupOwner>, + owners: PartitionOwners, +} + +#[derive(Debug, Default)] +struct PartitionOwners { + installed: BTreeMap<u64, ConsumerGroupOwner>, + retirement: Option<RetireConsumerGroupOwnersRequest>, } /// Canonical committed metadata and partition chains, accumulated across ticks. @@ -203,7 +211,7 @@ impl StateChecker { incarnation, replica_incarnation, committed: state.commit_min, - owners: previous.map_or_else(BTreeMap::new, |previous| previous.owners), + owners: previous.map_or_else(PartitionOwners::default, |previous| previous.owners), }; let Some(headers) = sim.partition_journaled_headers(idx, namespace, first..=state.commit_min) @@ -230,6 +238,7 @@ impl StateChecker { if matches!( header.operation, Operation::InstallConsumerGroupOwner + | Operation::RetireConsumerGroupOwners | Operation::StoreConsumerOffset | Operation::DeleteConsumerOffset ) { @@ -384,7 +393,7 @@ impl StateChecker { /// Membership is metadata state; the partition's ordered install barrier is /// what fences a departed member's later offset writes. fn check_partition_owner( - owners: &mut BTreeMap<u64, ConsumerGroupOwner>, + owners: &mut PartitionOwners, header: &PrepareHeader, body: &[u8], seed: u64, @@ -392,10 +401,28 @@ fn check_partition_owner( if header.operation == Operation::InstallConsumerGroupOwner { let install = InstallConsumerGroupOwnerRequest::decode_from(body) .expect("a committed owner installation decodes"); - owners.insert(install.group_id, install.owner); + if !owners + .retirement + .as_ref() + .is_some_and(|catalog| catalog.retires(install.group_id)) + { + owners.installed.insert(install.group_id, install.owner); + } return; } - if is_auto_commit_client(header.client) { + if header.operation == Operation::RetireConsumerGroupOwners { + let mut catalog = RetireConsumerGroupOwnersRequest::decode_from(body) + .expect("a committed owner retirement decodes"); + if let Some(previous) = &owners.retirement { + if catalog.metadata_op < previous.metadata_op { + return; + } + catalog + .live_group_ids + .retain(|group| !previous.retires(*group)); + } + owners.installed.retain(|group, _| !catalog.retires(*group)); + owners.retirement = Some(catalog); return; } let (request, generation) = @@ -412,12 +439,26 @@ fn check_partition_owner( .as_u32() .expect("a prepared group id is numeric"), ); + let auto_commit = is_auto_commit_client(header.client); + let cleanup = auto_commit && header.operation == Operation::DeleteConsumerOffset; + if owners + .retirement + .as_ref() + .is_some_and(|catalog| catalog.retires(group_id)) + { + assert!( + cleanup, + "group offset committed past its retirement fence: group={group_id}, op={}, seed={seed:#x}", + header.op, + ); + return; + } // The repair ring may have evicted the install before this checker started. - if let Some(owner) = owners.get(&group_id) { + if let Some(owner) = owners.installed.get(&group_id) { assert!( - !owner.is_unassigned() - && owner.client_id == header.client - && owner.session == header.session + (!owner.is_unassigned() || cleanup) + && (auto_commit + || (owner.client_id, owner.session) == (header.client, header.session)) && owner.generation == generation, "group offset committed past its owner fence: group={group_id}, owner={owner:?}, \ client={}, session={}, generation={generation}, op={}, seed={seed:#x}", @@ -592,15 +633,18 @@ mod tests { use crate::workload::Workload; use crate::workload::invariants::Invariants; use crate::workload::options::WorkloadOptions; - use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; + use iggy_binary_protocol::requests::consumer_offsets::{ + DeleteConsumerOffsetRequest, StoreConsumerOffsetRequest, + }; use iggy_binary_protocol::{AckLevel, WireConsumer, WireEncode, WireIdentifier}; + use message_bus::AUTO_COMMIT_CLIENT_ID; const SEED: u64 = 0x5C11; #[test] #[should_panic(expected = "group offset committed past its owner fence")] fn given_a_successor_install_when_the_departed_owner_commits_should_fail_the_oracle() { - let mut owners = BTreeMap::new(); + let mut owners = PartitionOwners::default(); let original = ConsumerGroupOwner { client_id: 1, session: 2, @@ -653,6 +697,146 @@ mod tests { } } + #[test] + #[should_panic(expected = "group offset committed past its owner fence")] + fn given_an_auto_commit_with_a_stale_generation_when_committed_should_fail_the_oracle() { + let mut owners = installed_owner(); + check_offset( + &mut owners, + AUTO_COMMIT_CLIENT_ID, + 0, + 6, + Operation::StoreConsumerOffset, + ); + } + + #[test] + #[should_panic(expected = "group offset committed past its owner fence")] + fn given_an_offset_with_a_stale_session_when_committed_should_fail_the_oracle() { + let mut owners = installed_owner(); + check_offset(&mut owners, 1, 3, 7, Operation::StoreConsumerOffset); + } + + #[test] + #[should_panic(expected = "group offset committed past its retirement fence")] + fn given_a_retired_group_when_an_auto_commit_store_commits_should_fail_the_oracle() { + let mut owners = retired_owner(); + check_offset( + &mut owners, + AUTO_COMMIT_CLIENT_ID, + 0, + 7, + Operation::StoreConsumerOffset, + ); + } + + #[test] + fn given_an_auto_commit_when_current_or_cleaning_retired_offsets_should_pass_the_oracle() { + let mut owners = installed_owner(); + check_offset( + &mut owners, + AUTO_COMMIT_CLIENT_ID, + 0, + 7, + Operation::StoreConsumerOffset, + ); + let mut owners = retired_owner(); + check_offset( + &mut owners, + AUTO_COMMIT_CLIENT_ID, + 0, + 6, + Operation::DeleteConsumerOffset, + ); + } + + fn installed_owner() -> PartitionOwners { + PartitionOwners { + installed: BTreeMap::from([( + 0, + ConsumerGroupOwner { + client_id: 1, + session: 2, + generation: 7, + }, + )]), + retirement: None, + } + } + + fn retired_owner() -> PartitionOwners { + let mut owners = installed_owner(); + owners.installed.insert( + 0, + ConsumerGroupOwner { + generation: 7, + ..Default::default() + }, + ); + for (metadata_op, live_group_ids) in [(8, vec![]), (9, vec![0])] { + let catalog = RetireConsumerGroupOwnersRequest { + incarnation: 1, + metadata_op, + next_group_id: 1, + live_group_ids, + }; + check_partition_owner( + &mut owners, + &PrepareHeader { + operation: Operation::RetireConsumerGroupOwners, + ..Default::default() + }, + &catalog.to_bytes(), + SEED, + ); + } + assert!(owners.installed.is_empty()); + owners + } + + fn check_offset( + owners: &mut PartitionOwners, + client: u128, + session: u64, + generation: u64, + operation: Operation, + ) { + let request = StoreConsumerOffsetRequest { + consumer: WireConsumer::consumer_group(WireIdentifier::numeric(0)), + stream_id: WireIdentifier::numeric(1), + topic_id: WireIdentifier::numeric(1), + partition_id: Some(0), + offset: 0, + ack: AckLevel::Quorum, + }; + let mut body = if operation == Operation::DeleteConsumerOffset { + DeleteConsumerOffsetRequest { + consumer: request.consumer, + stream_id: request.stream_id, + topic_id: request.topic_id, + partition_id: request.partition_id, + ack: request.ack, + } + .to_bytes() + .to_vec() + } else { + request.to_bytes().to_vec() + }; + body.extend_from_slice(&generation.to_le_bytes()); + check_partition_owner( + owners, + &PrepareHeader { + operation, + client, + session, + op: 10, + ..Default::default() + }, + &body, + SEED, + ); + } + /// A three-replica cluster with six committed metadata ops behind it. fn cluster_with_committed_ops() -> Simulator { server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
