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 a9d3f03faeeac2514a3cbb74dac1849b59a52a4b Author: Hubert Gruszecki <[email protected]> AuthorDate: Tue Oct 13 18:27:20 2026 +0200 test(simulator): validate committed consumer state Membership history alone did not detect writes after owner transfer, and successful sends could disappear during a refused deletion. Check ordered owner installations and equally committed replica offsets; retain send effects until deletion succeeds. --- core/simulator/src/workload/mod.rs | 43 ++++++- core/simulator/src/workload/oracle.rs | 156 +++++++++++++++++++++++- core/simulator/src/workload/shadow.rs | 33 +++-- core/simulator/src/workload/state_checker.rs | 172 ++++++++++++++++++++++++--- 4 files changed, 370 insertions(+), 34 deletions(-) diff --git a/core/simulator/src/workload/mod.rs b/core/simulator/src/workload/mod.rs index e48023790..c64422898 100644 --- a/core/simulator/src/workload/mod.rs +++ b/core/simulator/src/workload/mod.rs @@ -762,10 +762,7 @@ impl Workload { // sampled across a fixture replacement can belong to the deleted // incarnation, whose state the shadow dropped with it. let current = header.operation.is_metadata() - || self - .shadow - .replacements(namespace) - .unchanged_since(entry.replacements); + || self.shadow.effect_is_current(namespace, entry.replacements); let effect = if current { ops::predicted_effect(&entry.input, &classified) } else { @@ -843,8 +840,8 @@ impl Workload { /// Fail the run if a group offset write committed from a client that never /// joined the group on an incarnation the write could reach. Only a member /// becomes an owner, and the write binds only an owner of its own client and - /// session. A past join on the same incarnation suffices: until its successor - /// installs, an owner that left can still commit. + /// session. This is only the membership check; the partition state checker + /// compares each offset with the preceding owner installation in its log. fn assert_group_writer_joined(&self, entry: &InFlight, header: &ReplyHeader) { let (namespace, group_id) = match &entry.input { InFlightInput::StoreConsumerOffset(input) @@ -2623,6 +2620,40 @@ mod tests { let _ = workload.on_reply(&reply_to(&send, unavailable, &[])); } + #[test] + fn given_a_send_committed_during_a_fixture_delete_when_delete_refuses_should_keep_its_effect() { + let fixture = IggyNamespace::new(1, 1, 0); + let (_, mut workload, send) = sample_only(Action::SendMessages, vec![fixture]); + let delete = sample_toggle(&mut workload, &bound_client(2)); + let _ = workload.on_reply(&reply_to(&send, 0, &[])); + let committed_messages = workload.shadow.sends_committed(fixture); + assert!( + committed_messages > 0, + "the pending delete must not erase a committed send" + ); + + let busy = ApplyReply::err(IggyError::LifecycleBusy.as_code()); + let _ = workload.on_reply(&reply_to(&delete, 0, &busy.to_reply_body())); + assert_eq!(workload.shadow.sends_committed(fixture), committed_messages); + } + + #[test] + fn given_a_send_committed_during_a_fixture_delete_when_delete_succeeds_should_clear_its_effect() + { + let fixture = IggyNamespace::new(1, 1, 0); + let (_, mut workload, send) = sample_only(Action::SendMessages, vec![fixture]); + let delete = sample_toggle(&mut workload, &bound_client(2)); + let _ = workload.on_reply(&reply_to(&send, 0, &[])); + assert!(workload.shadow.sends_committed(fixture) > 0); + + let _ = workload.on_reply(&reply_to( + &delete, + 0, + &ApplyReply::default().to_reply_body(), + )); + assert_eq!(workload.shadow.sends_committed(fixture), 0); + } + /// A send sampled before a fixture replacement can commit on the deleted /// incarnation, so it must not count toward the recreated one. #[test] diff --git a/core/simulator/src/workload/oracle.rs b/core/simulator/src/workload/oracle.rs index 590de04a7..e13b9b01a 100644 --- a/core/simulator/src/workload/oracle.rs +++ b/core/simulator/src/workload/oracle.rs @@ -48,6 +48,7 @@ use crate::workload::{ use crate::{Simulator, rebalancing_timeout_micros}; use consensus::{Consensus, MetadataHandle, PartitionsHandle, Status}; use iggy_binary_protocol::primitives::partition_history::ConsumerGroupOwner; +use iggy_common::ConsumerKind; use metadata::impls::metadata::StreamsFrontend; use server_common::sharding::IggyNamespace; use std::collections::{BTreeMap, BTreeSet}; @@ -562,6 +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); assert_no_stale_revocations(sim, leader, seed); let report = ConvergenceReport { @@ -765,8 +767,8 @@ fn assert_committed_metadata_agrees(sim: &Simulator, live: &[usize], seed: u64) compared } -/// Require the partition primary of every workload namespace to hold each owner -/// that leader metadata settled there. Returns how many owners it compared. +/// Require the partition primary to hold each owner that leader metadata settled, +/// and equally committed replicas to agree. Returns how many owners it compared. /// /// Metadata settles an owner only after a partition installed it, so a settled /// owner the primary lacks was lost or never installed. Metadata names no owner @@ -820,12 +822,82 @@ fn assert_settled_owners_installed( ns {ns:?}, but partition primary {primary} installed {installed:?} \ (seed={seed:#x})", ); + let primary_state = sim.partition_consensus_state(usize::from(primary), ns); + for replica in 0..sim.replica_count { + let index = usize::from(replica); + if sim.is_crashed(replica) + || sim.partition_incarnation(index, ns) + != sim.partition_incarnation(usize::from(primary), ns) + || sim + .partition_consensus_state(index, ns) + .map(|state| state.commit_min) + != primary_state.as_ref().map(|state| state.commit_min) + { + continue; + } + assert_eq!( + sim.installed_group_owner(index, ns, group_id), + installed, + "replica {replica} and primary {primary} installed different owners at the same committed prefix \ + (ns={ns:?}, group={group_id}, seed={seed:#x})", + ); + } compared += 1; } } compared } +fn assert_partition_offsets_agree(sim: &Simulator, workload: &Workload, live: &[usize], seed: u64) { + for &namespace in &workload.options.namespaces { + let mut committed = BTreeMap::new(); + for &replica in live { + let Some(partition) = sim.replicas[replica] + .partition_shard(namespace) + .plane + .partitions() + .get_by_ns(&namespace) + else { + continue; + }; + let mut offsets = BTreeMap::new(); + for kind in ConsumerKind::ALL { + for (consumer, offset) in partition.retained_consumer_offsets(kind) { + offsets.insert((kind.as_code(), consumer), offset); + } + } + let prefix = ( + partition.created_revision(), + partition.consensus().commit_min(), + ); + if let Some((other, expected)) = committed.get(&prefix) { + assert_eq!( + &offsets, expected, + "replicas {replica} and {other} have different consumer offsets at committed prefix {prefix:?} \ + (ns={namespace:?}, seed={seed:#x})" + ); + } + if workload.strict_outcome_oracle() + && workload.options.ack_quorum_ratio >= 1.0 + && sim.primary_index(namespace).map(usize::from) == Some(replica) + { + let expected: BTreeMap<_, _> = workload + .shadow + .consumer_offsets + .iter() + .filter(|((ns, _, _), _)| *ns == namespace) + .map(|((_, kind, consumer), offset)| ((*kind, *consumer), *offset)) + .collect(); + assert_eq!( + offsets, expected, + "consumer offsets differ from the serial acknowledged shadow (ns={namespace:?}, seed={seed:#x})" + ); + } + committed.insert(prefix, (replica, offsets)); + } + } +} + /// Fail if a revocation pending in the metadata of replica `leader` aged past /// [`PENDING_REVOCATION_TIMEOUTS_MAX`] rebalancing timeouts. pub(crate) fn assert_no_stale_revocations(sim: &Simulator, leader: usize, seed: u64) { @@ -906,3 +978,83 @@ fn shadow_metadata(shadow: &Shadow) -> CommittedMetadata { consumer_groups: shadow.consumer_group_names.iter().cloned().collect(), } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::packet::PacketSimulatorOptions; + use crate::workload::effect::Effect; + use crate::workload::options::WorkloadOptions; + + const SEED: u64 = 0x000F_F5E7; + const CONSUMER: u32 = 7; + const OFFSET: u64 = 9; + + fn namespace() -> IggyNamespace { + IggyNamespace::new(1, 1, 0) + } + + fn offset_fixture() -> (Simulator, Workload) { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + let mut sim = Simulator::new( + 3, + std::iter::once(1), + PacketSimulatorOptions { + node_count: 3, + client_count: 1, + seed: SEED, + ..Default::default() + }, + ); + sim.init_partition(namespace()); + let workload = Workload::new(WorkloadOptions::new(SEED, 3, vec![namespace()])); + (sim, workload) + } + + fn seed_offset(sim: &Simulator, replica: usize) { + sim.replicas[replica] + .partition_shard(namespace()) + .plane + .partitions() + .get_by_ns(&namespace()) + .unwrap() + .seed_recovered_consumer_offset(ConsumerKind::Consumer, CONSUMER, OFFSET, OFFSET); + } + + #[test] + #[should_panic(expected = "different consumer offsets at committed prefix")] + fn given_a_backup_offset_divergence_when_quiesced_should_fail_the_oracle() { + let (sim, workload) = offset_fixture(); + seed_offset(&sim, 1); + assert_partition_offsets_agree(&sim, &workload, &[0, 1, 2], SEED); + } + + #[test] + #[should_panic(expected = "consumer offsets differ from the serial acknowledged shadow")] + fn given_an_unacknowledged_offset_effect_when_quiesced_should_fail_the_shadow_oracle() { + let (sim, mut workload) = offset_fixture(); + workload.options.ack_quorum_ratio = 1.0; + for replica in 0..3 { + seed_offset(&sim, replica); + } + assert_partition_offsets_agree(&sim, &workload, &[0, 1, 2], SEED); + } + + #[test] + fn given_matching_acknowledged_offsets_when_quiesced_should_pass_the_oracle() { + let (sim, mut workload) = offset_fixture(); + workload.options.ack_quorum_ratio = 1.0; + for replica in 0..3 { + seed_offset(&sim, replica); + } + workload.shadow.apply(Effect::OffsetStored { + key: (namespace(), ConsumerKind::Consumer.as_code(), CONSUMER), + value: OFFSET, + }); + assert_partition_offsets_agree(&sim, &workload, &[0, 1, 2], SEED); + } +} diff --git a/core/simulator/src/workload/shadow.rs b/core/simulator/src/workload/shadow.rs index cb7e00f48..2752b02f8 100644 --- a/core/simulator/src/workload/shadow.rs +++ b/core/simulator/src/workload/shadow.rs @@ -93,14 +93,6 @@ impl Replacements { pub const fn may_replace(self, sampled: Self) -> bool { self.started > sampled.settled } - - /// Whether a success of a request that recorded `sampled` belongs to the - /// current incarnation. A delete replies before its fences install, so the - /// success of an earlier request can belong to the deleted one. - #[must_use] - pub const fn unchanged_since(self, sampled: Self) -> bool { - sampled.started == sampled.settled && sampled.started == self.started - } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -154,9 +146,9 @@ struct FixtureGroup { /// newest such join. `None` when the fate of a join is unknown, which can /// still commit on any later incarnation. /// - /// Never cleared: until its successor installs, an owner that left can still - /// commit, and a join sampled before a replacement can commit on the deleted - /// incarnation. + /// Membership history only: until its successor installs, an owner that left + /// can still commit. The partition state checker validates the ordered install + /// barrier separately, so this history does not authorize a stale owner. joined_at: HashMap<u128, Option<u64>>, /// The metadata ops at which the deletes of the topic were reached, oldest /// first. A delete drops the group with its members, so a join counts only on @@ -452,6 +444,25 @@ impl Shadow { .map_or_else(Replacements::default, |fixture| fixture.replacements) } + /// A pending delete can still refuse, so retain committed effects until its + /// success clears them. After a successful delete, late replies belong to the + /// old history until the replacement settles. + #[must_use] + pub fn effect_is_current(&self, namespace: IggyNamespace, sampled: Replacements) -> bool { + self.fixtures + .iter() + .find(|fixture| fixture.namespaces.contains(&namespace)) + .is_none_or(|fixture| { + sampled.settled == fixture.replacements.settled + && matches!( + fixture.state, + FixtureState::Live + | FixtureState::Deleting { .. } + | FixtureState::DeleteUnknown + ) + }) + } + /// Replacement counts summed over every fixture topic. #[must_use] pub fn replacement_totals(&self) -> Replacements { diff --git a/core/simulator/src/workload/state_checker.rs b/core/simulator/src/workload/state_checker.rs index 97b9628cc..7dded9dc0 100644 --- a/core/simulator/src/workload/state_checker.rs +++ b/core/simulator/src/workload/state_checker.rs @@ -36,8 +36,16 @@ //! non-vacuous: a chain nothing was compared against passes silently. use crate::Simulator; -use iggy_binary_protocol::PrepareHeader; +use consensus::PartitionsHandle; +use iggy_binary_protocol::primitives::consumer::KIND_CONSUMER_GROUP; +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::{HEADER_SIZE, Operation, PrepareHeader, WireDecode}; use journal::Journal; +use message_bus::is_auto_commit_client; use server_common::sharding::IggyNamespace; use std::collections::BTreeMap; @@ -68,6 +76,7 @@ struct PartitionProgress { incarnation: u64, replica_incarnation: u128, committed: u64, + owners: BTreeMap<u64, ConsumerGroupOwner>, } /// Canonical committed metadata and partition chains, accumulated across ticks. @@ -177,23 +186,25 @@ impl StateChecker { return; }; let replica_incarnation = sim.replicas[idx].metadata_incarnation; - let previous = self.partition_verified_upto.insert( - (replica_idx, namespace), - PartitionProgress { - incarnation, - replica_incarnation, - committed: state.commit_min, - }, - ); + let previous = self + .partition_verified_upto + .remove(&(replica_idx, namespace)); // Recheck the head even without progress, and the retained prefix after a // restart, replacement or lower recovered commit point. + let previous = previous.filter(|previous| { + previous.incarnation == incarnation + && previous.replica_incarnation == replica_incarnation + && previous.committed <= state.commit_min + }); let first = previous - .filter(|previous| { - previous.incarnation == incarnation - && previous.replica_incarnation == replica_incarnation - && previous.committed <= state.commit_min - }) + .as_ref() .map_or(1, |previous| previous.committed.max(1)); + let mut progress = PartitionProgress { + incarnation, + replica_incarnation, + committed: state.commit_min, + owners: previous.map_or_else(BTreeMap::new, |previous| previous.owners), + }; let Some(headers) = sim.partition_journaled_headers(idx, namespace, first..=state.commit_min) else { @@ -216,7 +227,34 @@ impl StateChecker { canonical.witnesses, canonical.checksum, ); canonical.witnesses.insert(replica_idx, replica_incarnation); + if matches!( + header.operation, + Operation::InstallConsumerGroupOwner + | Operation::StoreConsumerOffset + | Operation::DeleteConsumerOffset + ) { + let partition = sim.replicas[idx] + .partition_shard(namespace) + .plane + .partitions() + .get_by_ns(&namespace) + .expect("a journaled partition is hosted"); + let prepare = partition + .log + .journal() + .inner + .repair_entry(op) + .expect("a retained prepare header has its body"); + check_partition_owner( + &mut progress.owners, + &header, + &prepare.as_slice()[HEADER_SIZE..], + seed, + ); + } } + self.partition_verified_upto + .insert((replica_idx, namespace), progress); } /// Fold one replica's op into the canonical chain, or prove its absence @@ -343,6 +381,53 @@ 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>, + header: &PrepareHeader, + body: &[u8], + seed: u64, +) { + if header.operation == Operation::InstallConsumerGroupOwner { + let install = InstallConsumerGroupOwnerRequest::decode_from(body) + .expect("a committed owner installation decodes"); + owners.insert(install.group_id, install.owner); + return; + } + if is_auto_commit_client(header.client) { + return; + } + let (request, generation) = + split_prepared_offset(body).expect("a committed offset has its owner token"); + let (request, _) = + GetConsumerOffsetRequest::decode(request).expect("a committed offset prefix decodes"); + if request.consumer.kind != KIND_CONSUMER_GROUP { + return; + } + let group_id = u64::from( + request + .consumer + .id + .as_u32() + .expect("a prepared group id is numeric"), + ); + // The repair ring may have evicted the install before this checker started. + if let Some(owner) = owners.get(&group_id) { + assert!( + !owner.is_unassigned() + && owner.client_id == header.client + && owner.session == 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}", + header.client, + header.session, + header.op, + ); + } +} + /// Assert every live replica's committed metadata prefix agrees, op for op. /// /// The quiesce-time counterpart to [`StateChecker::check`]: that one folds ops in @@ -507,10 +592,67 @@ mod tests { use crate::workload::Workload; use crate::workload::invariants::Invariants; use crate::workload::options::WorkloadOptions; - use consensus::PartitionsHandle; + use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; + use iggy_binary_protocol::{AckLevel, WireConsumer, WireEncode, WireIdentifier}; 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 original = ConsumerGroupOwner { + client_id: 1, + session: 2, + generation: 3, + }; + let successor = ConsumerGroupOwner { + client_id: 4, + session: 5, + generation: 6, + }; + 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 = request.to_bytes().to_vec(); + body.extend_from_slice(&original.generation.to_le_bytes()); + for (op, owner) in [(1, original), (3, successor)] { + let install = InstallConsumerGroupOwnerRequest { + incarnation: 1, + group_id: 0, + owner, + metadata_op: op, + }; + check_partition_owner( + &mut owners, + &PrepareHeader { + operation: Operation::InstallConsumerGroupOwner, + op, + ..Default::default() + }, + &install.to_bytes(), + SEED, + ); + check_partition_owner( + &mut owners, + &PrepareHeader { + operation: Operation::StoreConsumerOffset, + client: original.client_id, + session: original.session, + op: op + 1, + ..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 {
