You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/13 20:34:52 UTC
(iggy) 04/06: test(simulator): exercise stale owners and strict offsets
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 23fac06d6cbe0e580780e8cd8c071edefd7ad6af
Author: Hubert Gruszecki <h....@gmail.com>
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 {