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/05/14 12:36:54 UTC
(iggy) branch chaos updated (f5d875ba5 -> cb77a3767)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch chaos
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit f5d875ba5 lab
new cb77a3767 lab
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (f5d875ba5)
\
N -- N -- N refs/heads/chaos (cb77a3767)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
core/chaos/src/ops/generate.rs | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
(iggy) 01/01: lab
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch chaos
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit cb77a376771f7fe2bf36f1ebefc2ba4c0b66a39b
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Feb 13 21:13:59 2026 +0100
lab
---
.gitignore | 1 +
Cargo.lock | 21 +
Cargo.toml | 2 +
DEPENDENCIES.md | 1 +
core/chaos/Cargo.toml | 46 ++
core/chaos/src/args.rs | 193 ++++++++
core/chaos/src/client.rs | 96 ++++
core/chaos/src/error.rs | 47 ++
core/chaos/src/eventually.rs | 130 +++++
core/chaos/src/in_flight.rs | 60 +++
core/chaos/src/invariants.rs | 573 +++++++++++++++++++++++
core/chaos/src/main.rs | 117 +++++
core/chaos/src/ops/classify.rs | 322 +++++++++++++
core/chaos/src/ops/execute.rs | 379 +++++++++++++++
core/chaos/src/ops/generate.rs | 281 +++++++++++
core/chaos/src/ops/mod.rs | 458 ++++++++++++++++++
core/chaos/src/ops/shadow_update.rs | 173 +++++++
core/chaos/src/payload.rs | 151 ++++++
core/chaos/src/perfetto.rs | 312 ++++++++++++
core/chaos/src/replay.rs | 346 ++++++++++++++
core/chaos/src/report.rs | 139 ++++++
core/chaos/src/runner.rs | 420 +++++++++++++++++
core/chaos/src/safe_name.rs | 114 +++++
core/chaos/src/scenarios/concurrent_crud.rs | 93 ++++
core/chaos/src/scenarios/consumer_group_churn.rs | 194 ++++++++
core/chaos/src/scenarios/kit.rs | 117 +++++
core/chaos/src/scenarios/lane_builder.rs | 91 ++++
core/chaos/src/scenarios/mixed_workload.rs | 76 +++
core/chaos/src/scenarios/mod.rs | 208 ++++++++
core/chaos/src/scenarios/segment_rotation.rs | 80 ++++
core/chaos/src/scenarios/stale_consumers.rs | 270 +++++++++++
core/chaos/src/shadow.rs | 276 +++++++++++
core/chaos/src/trace.rs | 211 +++++++++
core/chaos/src/violation.rs | 34 ++
core/chaos/src/watermark.rs | 84 ++++
core/chaos/src/worker.rs | 394 ++++++++++++++++
36 files changed, 6510 insertions(+)
diff --git a/.gitignore b/.gitignore
index 7c8d4d1dd..36ce66d37 100644
--- a/.gitignore
+++ b/.gitignore
@@ -8,6 +8,7 @@ target*
local_data*
local_state*
test_logs/
+chaos_logs/
/bench_*
/sdk/errors_table/
/rust-clippy-results.sarif
diff --git a/Cargo.lock b/Cargo.lock
index d7ffd97b4..6ae02db1c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4550,6 +4550,27 @@ dependencies = [
"zip",
]
+[[package]]
+name = "iggy-chaos"
+version = "0.1.0"
+dependencies = [
+ "ahash 0.8.12",
+ "async-trait",
+ "bytes",
+ "chrono",
+ "clap",
+ "crc32fast",
+ "iggy",
+ "iggy_common",
+ "rand 0.10.0",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tokio",
+ "tracing",
+ "tracing-subscriber",
+]
+
[[package]]
name = "iggy-cli"
version = "0.10.3-edge.1"
diff --git a/Cargo.toml b/Cargo.toml
index e306352c1..91613b9be 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -26,6 +26,7 @@ members = [
"core/bench/report",
"core/bench/runner",
"core/binary_protocol",
+ "core/chaos",
"core/cli",
"core/clock",
"core/common",
@@ -125,6 +126,7 @@ configs = { path = "core/configs", version = "0.1.0" }
configs_derive = { path = "core/configs_derive", version = "0.1.0" }
consensus = { path = "core/consensus" }
console-subscriber = "0.5.0"
+crc32fast = "1.4.2"
crossbeam = "0.8.4"
ctor = "0.6.3"
ctrlc = { version = "3.5", features = ["termination"] }
diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md
index a9e14e6d4..ccb3040ab 100644
--- a/DEPENDENCIES.md
+++ b/DEPENDENCIES.md
@@ -391,6 +391,7 @@ idna_adapter: 1.2.1, "Apache-2.0 OR MIT",
iggy: 0.8.3-edge.1, "Apache-2.0",
iggy-bench: 0.3.3-edge.1, "Apache-2.0",
iggy-bench-dashboard-server: 0.5.1-edge.1, "Apache-2.0",
+iggy-chaos: 0.1.0, "Apache-2.0",
iggy-cli: 0.10.3-edge.1, "Apache-2.0",
iggy-connectors: 0.2.4-edge.1, "Apache-2.0",
iggy-mcp: 0.2.3-edge.1, "Apache-2.0",
diff --git a/core/chaos/Cargo.toml b/core/chaos/Cargo.toml
new file mode 100644
index 000000000..9189c67b9
--- /dev/null
+++ b/core/chaos/Cargo.toml
@@ -0,0 +1,46 @@
+# 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.
+
+[package]
+name = "iggy-chaos"
+version = "0.1.0"
+edition = "2024"
+license = "Apache-2.0"
+repository = "https://github.com/apache/iggy"
+homepage = "https://iggy.apache.org"
+description = "Chaos stress testing CLI for Iggy message streaming platform"
+
+[[bin]]
+name = "iggy-chaos"
+path = "src/main.rs"
+
+[dependencies]
+ahash = { workspace = true }
+async-trait = { workspace = true }
+bytes = { workspace = true }
+chrono = { workspace = true }
+clap = { workspace = true }
+crc32fast = { workspace = true }
+iggy = { workspace = true }
+iggy_common = { workspace = true }
+rand = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
diff --git a/core/chaos/src/args.rs b/core/chaos/src/args.rs
new file mode 100644
index 000000000..5a308376f
--- /dev/null
+++ b/core/chaos/src/args.rs
@@ -0,0 +1,193 @@
+/* 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 clap::{Parser, Subcommand, ValueEnum};
+use iggy_common::IggyDuration;
+use std::path::PathBuf;
+use std::str::FromStr;
+
+/// Default message payload size in bytes.
+const DEFAULT_MESSAGE_SIZE: u32 = 256;
+
+#[derive(Parser)]
+#[command(
+ name = "iggy-chaos",
+ about = "Chaos stress testing CLI for Iggy message streaming platform"
+)]
+pub struct IggyLabArgs {
+ #[command(subcommand)]
+ pub command: Command,
+}
+
+#[derive(Subcommand)]
+pub enum Command {
+ /// Run a chaos scenario against a live Iggy server
+ Run(RunArgs),
+ /// Replay a recorded trace against a live Iggy server
+ Replay(ReplayArgs),
+ /// List available scenarios
+ List,
+ /// Show detailed description of a scenario
+ Explain {
+ /// Scenario name to explain
+ scenario: ScenarioName,
+ },
+ /// Convert trace.jsonl to Perfetto trace format for visualization
+ Viz(VizArgs),
+}
+
+#[derive(Parser)]
+pub struct RunArgs {
+ /// Which scenario to run
+ pub scenario: ScenarioName,
+
+ /// Server address (host:port)
+ #[arg(long, default_value = "127.0.0.1:8090")]
+ pub server_address: String,
+
+ /// Transport protocol
+ #[arg(long, default_value = "tcp")]
+ pub transport: Transport,
+
+ /// PRNG seed for reproducibility (random if omitted)
+ #[arg(long)]
+ pub seed: Option<u64>,
+
+ /// How long to run the chaos phase
+ #[arg(long, default_value = "30s", value_parser = IggyDuration::from_str)]
+ pub duration: IggyDuration,
+
+ /// Stop after this many total ops (across all workers). Duration still acts as safety timeout.
+ #[arg(long)]
+ pub ops: Option<u64>,
+
+ /// Resource name prefix for safety isolation.
+ /// With `--prefix lab-`, stable streams use `lab-s-` and churn streams use `lab-c-`.
+ #[arg(long, default_value = "lab-")]
+ pub prefix: String,
+
+ /// Directory for trace artifacts (stdout-only if omitted)
+ #[arg(long)]
+ pub output_dir: Option<PathBuf>,
+
+ /// Number of concurrent worker tasks
+ #[arg(long, default_value = "8", value_parser = clap::value_parser!(u32).range(1..))]
+ pub workers: u32,
+
+ /// Payload size in bytes per message
+ #[arg(long, default_value = "256", value_parser = clap::value_parser!(u32).range(1..))]
+ pub message_size: u32,
+
+ /// Messages per send batch
+ #[arg(long, default_value = "10")]
+ pub messages_per_batch: u32,
+
+ /// Continue running after ServerBug classification instead of stopping
+ #[arg(long)]
+ pub no_fail_fast: bool,
+
+ /// Delete stale lab-prefixed resources before starting
+ #[arg(long)]
+ pub force_cleanup: bool,
+
+ /// Skip post-run resource deletion
+ #[arg(long)]
+ pub no_cleanup: bool,
+
+ /// Skip post-run invariant verification
+ #[arg(long)]
+ pub skip_post_run_verify: bool,
+}
+
+#[derive(Parser)]
+pub struct ReplayArgs {
+ /// Directory containing trace-worker-*.jsonl files
+ pub trace_dir: PathBuf,
+
+ /// Server address (host:port)
+ #[arg(long, default_value = "127.0.0.1:8090")]
+ pub server_address: String,
+
+ /// Transport protocol
+ #[arg(long, default_value = "tcp")]
+ pub transport: Transport,
+
+ /// Payload size in bytes per message
+ #[arg(long, default_value_t = DEFAULT_MESSAGE_SIZE)]
+ pub message_size: u32,
+
+ /// Resource name prefix for cleanup
+ #[arg(long, default_value = "lab-")]
+ pub prefix: String,
+
+ /// Stop on first concerning divergence
+ #[arg(long)]
+ pub fail_fast: bool,
+
+ /// Delete stale lab-prefixed resources before replaying
+ #[arg(long)]
+ pub force_cleanup: bool,
+}
+
+#[derive(Parser)]
+pub struct VizArgs {
+ /// Bundle directory containing trace.jsonl
+ pub bundle_dir: PathBuf,
+}
+
+/// Maps to TransportProtocol but implements clap::ValueEnum
+#[derive(Debug, Clone, Copy, ValueEnum)]
+pub enum Transport {
+ Tcp,
+ Quic,
+ Http,
+ #[value(alias = "ws")]
+ WebSocket,
+}
+
+#[derive(Debug, Clone, Copy, ValueEnum)]
+pub enum ScenarioName {
+ ConcurrentCrud,
+ SegmentRotation,
+ MixedWorkload,
+ StaleConsumers,
+ ConsumerGroupChurn,
+}
+
+impl std::fmt::Display for Transport {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::Tcp => write!(f, "tcp"),
+ Self::Quic => write!(f, "quic"),
+ Self::Http => write!(f, "http"),
+ Self::WebSocket => write!(f, "websocket"),
+ }
+ }
+}
+
+impl std::fmt::Display for ScenarioName {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::ConcurrentCrud => write!(f, "concurrent-crud"),
+ Self::SegmentRotation => write!(f, "segment-rotation"),
+ Self::MixedWorkload => write!(f, "mixed-workload"),
+ Self::StaleConsumers => write!(f, "stale-consumers"),
+ Self::ConsumerGroupChurn => write!(f, "consumer-group-churn"),
+ }
+ }
+}
diff --git a/core/chaos/src/client.rs b/core/chaos/src/client.rs
new file mode 100644
index 000000000..b856347f1
--- /dev/null
+++ b/core/chaos/src/client.rs
@@ -0,0 +1,96 @@
+/* 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 crate::args::Transport;
+use crate::scenarios::NamespacePrefixes;
+use iggy::prelude::*;
+use std::str::FromStr;
+use std::sync::Arc;
+use tracing::warn;
+
+pub async fn create_client(
+ server_address: &str,
+ transport: Transport,
+) -> Result<IggyClient, IggyError> {
+ let client = match transport {
+ Transport::Tcp => IggyClient::builder()
+ .with_tcp()
+ .with_server_address(server_address.to_owned())
+ .build()?,
+ Transport::Quic => IggyClient::builder()
+ .with_quic()
+ .with_server_address(server_address.to_owned())
+ .build()?,
+ Transport::Http => {
+ let api_url = format!("http://{server_address}");
+ IggyClient::builder()
+ .with_http()
+ .with_api_url(api_url)
+ .build()?
+ }
+ Transport::WebSocket => IggyClient::builder()
+ .with_websocket()
+ .with_server_address(server_address.to_owned())
+ .build()?,
+ };
+ client.connect().await?;
+ client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await?;
+ Ok(client)
+}
+
+/// Create a TCP client with a very long heartbeat interval (1 hour).
+///
+/// When connected to a server with `heartbeat.enabled = true` and a short
+/// `heartbeat.interval` (e.g. 2s), this client will be evicted after
+/// `server_interval * 1.2` because it never sends pings within the threshold.
+#[allow(dead_code)]
+pub async fn create_stale_client(server_address: &str) -> Result<IggyClient, IggyError> {
+ let config = TcpClientConfig {
+ server_address: server_address.to_string(),
+ heartbeat_interval: IggyDuration::from_str("1h").unwrap(),
+ ..TcpClientConfig::default()
+ };
+ let tcp = TcpClient::create(Arc::new(config))?;
+ Client::connect(&tcp).await?;
+ let client = IggyClient::create(ClientWrapper::Tcp(tcp), None, None);
+ client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await?;
+ Ok(client)
+}
+
+/// Delete all streams whose name starts with `prefix`.
+pub async fn cleanup_prefixed(client: &IggyClient, prefix: &str) {
+ let Ok(streams) = client.get_streams().await else {
+ return;
+ };
+ for stream in streams.iter().filter(|s| s.name.starts_with(prefix)) {
+ let id = Identifier::numeric(stream.id).unwrap();
+ if let Err(e) = client.delete_stream(&id).await {
+ warn!("Cleanup: failed to delete stream '{}': {e}", stream.name);
+ }
+ }
+}
+
+/// Delete all streams in both stable and churn namespaces.
+pub async fn cleanup_all(client: &IggyClient, prefixes: &NamespacePrefixes) {
+ cleanup_prefixed(client, &prefixes.stable).await;
+ cleanup_prefixed(client, &prefixes.churn).await;
+}
diff --git a/core/chaos/src/error.rs b/core/chaos/src/error.rs
new file mode 100644
index 000000000..ef0ddb670
--- /dev/null
+++ b/core/chaos/src/error.rs
@@ -0,0 +1,47 @@
+/* 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 iggy_common::IggyError;
+
+#[derive(Debug, thiserror::Error)]
+pub enum LabError {
+ #[error(transparent)]
+ Iggy(#[from] IggyError),
+
+ #[error("operation timed out: {context}")]
+ Timeout { context: String },
+
+ #[error("{context}: {source}")]
+ Io {
+ context: String,
+ source: std::io::Error,
+ },
+
+ #[error("trace parse error at {path}:{line}: {message}")]
+ TraceParse {
+ path: String,
+ line: usize,
+ message: String,
+ },
+
+ #[error("invalid identifier: {0}")]
+ InvalidIdentifier(String),
+
+ #[error("prefix safety violation: {0}")]
+ PrefixViolation(String),
+}
diff --git a/core/chaos/src/eventually.rs b/core/chaos/src/eventually.rs
new file mode 100644
index 000000000..2d24fff47
--- /dev/null
+++ b/core/chaos/src/eventually.rs
@@ -0,0 +1,130 @@
+/* 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 crate::error::LabError;
+use iggy::prelude::*;
+use iggy_common::IggyError;
+use std::time::Duration;
+use tokio::time::{Instant, interval};
+
+const DEFAULT_POLL_TICK: Duration = Duration::from_millis(100);
+
+/// Poll `probe` until it returns `Some(T)`, or fail after `timeout`.
+///
+/// On each tick, calls `probe()`:
+/// - `Ok(Some(v))` → returns `v`
+/// - `Ok(None)` → retries
+/// - `Err(e)` where `e` is retryable → retries
+/// - `Err(e)` otherwise → propagates immediately
+#[allow(dead_code)]
+pub async fn eventually<F, Fut, T>(
+ timeout: Duration,
+ tick: Duration,
+ mut probe: F,
+) -> Result<T, LabError>
+where
+ F: FnMut() -> Fut,
+ Fut: std::future::Future<Output = Result<Option<T>, LabError>>,
+{
+ let deadline = Instant::now() + timeout;
+ let mut ticker = interval(tick);
+ loop {
+ ticker.tick().await;
+ if Instant::now() >= deadline {
+ return Err(LabError::Timeout {
+ context: "eventually: condition not met within timeout".into(),
+ });
+ }
+ match probe().await {
+ Ok(Some(v)) => return Ok(v),
+ Ok(None) => continue,
+ Err(e) if is_retryable(&e) => continue,
+ Err(e) => return Err(e),
+ }
+ }
+}
+
+fn is_retryable(err: &LabError) -> bool {
+ match err {
+ LabError::Iggy(e) => matches!(
+ e,
+ IggyError::ResourceNotFound(_)
+ | IggyError::ConsumerGroupIdNotFound(_, _)
+ | IggyError::ConsumerGroupNameNotFound(_, _)
+ | IggyError::ConsumerGroupMemberNotFound(_, _, _)
+ | IggyError::Disconnected
+ | IggyError::NotConnected
+ | IggyError::CannotEstablishConnection
+ | IggyError::Unauthenticated
+ ),
+ _ => false,
+ }
+}
+
+/// Wait until `get_consumer_group()` reports exactly `expected` members.
+#[allow(dead_code)]
+pub async fn await_members_count(
+ client: &IggyClient,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ group_id: &Identifier,
+ expected: u32,
+ timeout: Duration,
+) -> Result<ConsumerGroupDetails, LabError> {
+ eventually(timeout, DEFAULT_POLL_TICK, || async {
+ let details = client
+ .get_consumer_group(stream_id, topic_id, group_id)
+ .await?;
+ match details {
+ Some(d) if d.members_count == expected => Ok(Some(d)),
+ _ => Ok(None),
+ }
+ })
+ .await
+}
+
+/// Wait until rebalance is complete: every member has partitions assigned and
+/// the total assigned partitions equals the topic's partition count.
+#[allow(dead_code)]
+pub async fn await_rebalance_complete(
+ client: &IggyClient,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ group_id: &Identifier,
+ timeout: Duration,
+) -> Result<ConsumerGroupDetails, LabError> {
+ eventually(timeout, DEFAULT_POLL_TICK, || async {
+ let details = client
+ .get_consumer_group(stream_id, topic_id, group_id)
+ .await?;
+ let Some(d) = details else {
+ return Ok(None);
+ };
+ if d.members_count == 0 {
+ return Ok(None);
+ }
+ let all_have_partitions = d.members.iter().all(|m| m.partitions_count > 0);
+ let total_assigned: u32 = d.members.iter().map(|m| m.partitions_count).sum();
+ if all_have_partitions && total_assigned == d.partitions_count {
+ Ok(Some(d))
+ } else {
+ Ok(None)
+ }
+ })
+ .await
+}
diff --git a/core/chaos/src/in_flight.rs b/core/chaos/src/in_flight.rs
new file mode 100644
index 000000000..db8de79b3
--- /dev/null
+++ b/core/chaos/src/in_flight.rs
@@ -0,0 +1,60 @@
+/* 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 std::collections::HashMap;
+use std::sync::Mutex;
+
+/// Tracks resource keys currently being mutated by any worker.
+///
+/// Bridges the TOCTOU gap between a destructive op executing on the server
+/// and its shadow state update arriving: the classifier checks in-flight keys
+/// alongside shadow state so that a concurrent not-found error during this
+/// window is classified as `ExpectedConcurrent` rather than `ServerBug`.
+///
+/// Uses reference counting so that overlapping ops on the same key (e.g.
+/// PurgeTopic + DeleteTopic on the same topic) don't prematurely remove
+/// coverage when the first one completes.
+pub struct InFlightOps {
+ keys: Mutex<HashMap<String, usize>>,
+}
+
+impl InFlightOps {
+ pub fn new() -> Self {
+ Self {
+ keys: Mutex::new(HashMap::new()),
+ }
+ }
+
+ pub fn register(&self, key: &str) {
+ *self.keys.lock().unwrap().entry(key.to_owned()).or_insert(0) += 1;
+ }
+
+ pub fn deregister(&self, key: &str) {
+ let mut keys = self.keys.lock().unwrap();
+ if let Some(count) = keys.get_mut(key) {
+ *count -= 1;
+ if *count == 0 {
+ keys.remove(key);
+ }
+ }
+ }
+
+ pub fn contains(&self, key: &str) -> bool {
+ self.keys.lock().unwrap().contains_key(key)
+ }
+}
diff --git a/core/chaos/src/invariants.rs b/core/chaos/src/invariants.rs
new file mode 100644
index 000000000..a2df39647
--- /dev/null
+++ b/core/chaos/src/invariants.rs
@@ -0,0 +1,573 @@
+/* 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 crate::args::Transport;
+use crate::client::create_client;
+use crate::ops::{Op, OpDetail, OpOutcome};
+use crate::scenarios::{Namespace, NamespacePrefixes};
+use crate::shadow::ShadowState;
+use crate::violation::ViolationCategory;
+use crate::watermark::WatermarkState;
+use iggy::prelude::*;
+use serde::Serialize;
+use std::collections::HashMap;
+use std::time::Duration;
+use tokio::task::JoinSet;
+use tracing::info;
+
+pub type PartitionKey = (String, String, u32);
+
+#[derive(Debug, Clone, Serialize)]
+pub struct TraceAnchor {
+ pub seq: u64,
+ pub t_us: u64,
+ pub worker: u32,
+ pub lane: &'static str,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct InvariantViolation {
+ pub category: ViolationCategory,
+ pub kind: &'static str,
+ pub description: String,
+ pub context: String,
+ #[serde(flatten, skip_serializing_if = "Option::is_none")]
+ pub trace_anchor: Option<TraceAnchor>,
+}
+
+impl InvariantViolation {
+ pub fn with_anchor(mut self, anchor: TraceAnchor) -> Self {
+ self.trace_anchor = Some(anchor);
+ self
+ }
+}
+
+/// Check that polled offsets are monotonically increasing per partition.
+/// Offset regressions after a recent purge are expected and not violations.
+pub fn check_offset_monotonicity(
+ last_offsets: &mut HashMap<PartitionKey, u64>,
+ op: &Op,
+ outcome: &OpOutcome,
+ shadow: &ShadowState,
+) -> Result<(), InvariantViolation> {
+ let (stream, topic, partition) = match op {
+ Op::PollMessages {
+ stream,
+ topic,
+ partition,
+ ..
+ } => (stream, topic, partition),
+ _ => return Ok(()),
+ };
+ let offset = match outcome {
+ OpOutcome::Success {
+ detail: Some(OpDetail::Polled { offset, .. }),
+ } => *offset,
+ _ => return Ok(()),
+ };
+
+ let key = (stream.clone(), topic.clone(), *partition);
+ if let Some(&last) = last_offsets.get(&key)
+ && offset < last
+ {
+ if shadow.was_recently_purged(stream, topic) {
+ last_offsets.insert(key, offset);
+ return Ok(());
+ }
+ return Err(InvariantViolation {
+ category: ViolationCategory::Invariant,
+ kind: "offset_regression",
+ description: format!("Offset went backwards: {last} -> {offset}"),
+ context: format!("stream={stream}, topic={topic}, partition={partition}"),
+ trace_anchor: None,
+ });
+ }
+ last_offsets.insert(key, offset);
+ Ok(())
+}
+
+/// Post-run: verify server state matches shadow state for all prefixed resources.
+/// Stable-namespace streams are checked strictly (mismatches are real bugs).
+/// Churn-namespace streams are checked leniently (mismatches from concurrent races are expected).
+pub async fn post_run_verify(
+ client: &IggyClient,
+ shadow: &ShadowState,
+ prefixes: &NamespacePrefixes,
+) -> Vec<InvariantViolation> {
+ let mut violations = Vec::new();
+
+ let streams = match client.get_streams().await {
+ Ok(s) => s,
+ Err(e) => {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::Infrastructure,
+ kind: "post_run_fetch_failed",
+ description: format!("Failed to get streams: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ return violations;
+ }
+ };
+
+ let server_streams: HashMap<&str, &Stream> = streams
+ .iter()
+ .filter(|s| prefixes.matches(&s.name))
+ .map(|s| (s.name.as_str(), s))
+ .collect();
+
+ // Check all shadow streams exist on server (strict for stable, lenient for churn)
+ for (name, shadow_stream) in &shadow.streams {
+ if !server_streams.contains_key(name.as_str())
+ && shadow_stream.namespace == Namespace::Stable
+ {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::ShadowMismatch,
+ kind: "missing_stable_stream",
+ description: format!("Shadow has stable stream '{name}' but server does not"),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ }
+ }
+
+ // Check no extra prefixed streams on server
+ for &name in server_streams.keys() {
+ if !shadow.streams.contains_key(name) {
+ let is_stable = prefixes
+ .namespace_of(name)
+ .is_some_and(|ns| ns == Namespace::Stable);
+ if is_stable {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::ShadowMismatch,
+ kind: "extra_stable_stream",
+ description: format!("Server has stable stream '{name}' not in shadow"),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ }
+ // Churn: shadow tracking lag from concurrent ops - skip
+ }
+ }
+
+ // Deep-check topics for each matching stream
+ for (name, shadow_stream) in &shadow.streams {
+ let Some(server_stream) = server_streams.get(name.as_str()) else {
+ continue;
+ };
+
+ let is_stable = shadow_stream.namespace == Namespace::Stable;
+
+ let stream_id = Identifier::numeric(server_stream.id).unwrap();
+ let Ok(Some(details)) = client.get_stream(&stream_id).await else {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::Infrastructure,
+ kind: "stream_details_fetch_failed",
+ description: format!("Failed to get details for stream '{name}'"),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ continue;
+ };
+
+ let server_topics: HashMap<&str, &Topic> = details
+ .topics
+ .iter()
+ .map(|t| (t.name.as_str(), t))
+ .collect();
+
+ for topic_name in shadow_stream.topics.keys() {
+ if !server_topics.contains_key(topic_name.as_str()) && is_stable {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::ShadowMismatch,
+ kind: "missing_stable_topic",
+ description: format!(
+ "Shadow has topic '{topic_name}' in stable stream '{name}' but server does not"
+ ),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ }
+ }
+
+ for &topic_name in server_topics.keys() {
+ if !shadow_stream.topics.contains_key(topic_name) && is_stable {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::ShadowMismatch,
+ kind: "extra_stable_topic",
+ description: format!(
+ "Server has topic '{topic_name}' in stable stream '{name}' not in shadow"
+ ),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ }
+ }
+ }
+
+ violations
+}
+
+/// Post-run: verify that server-reported message counts do not exceed send watermarks.
+/// Detects phantom reads (server returns messages that were never sent) and purge
+/// failures (server retained data that should have been removed).
+pub async fn watermark_verify(
+ client: &IggyClient,
+ watermarks: &WatermarkState,
+ shadow: &ShadowState,
+) -> Vec<InvariantViolation> {
+ let mut violations = Vec::new();
+
+ // For each purged (stream, topic), verify server's message count doesn't
+ // exceed post-purge sends
+ for (stream, topic) in watermarks.purge_history() {
+ if !shadow.stream_exists(stream) {
+ continue;
+ }
+ let stream_id = match Identifier::from_str_value(stream) {
+ Ok(id) => id,
+ Err(_) => continue,
+ };
+
+ let topics_to_check: Vec<String> = match topic {
+ Some(t) => vec![t.clone()],
+ None => shadow
+ .streams
+ .get(stream)
+ .map(|s| s.topics.keys().cloned().collect())
+ .unwrap_or_default(),
+ };
+
+ for topic_name in &topics_to_check {
+ if !shadow.topic_exists(stream, topic_name) {
+ continue;
+ }
+ let topic_id = match Identifier::from_str_value(topic_name) {
+ Ok(id) => id,
+ Err(_) => continue,
+ };
+
+ let details = match client.get_topic(&stream_id, &topic_id).await {
+ Ok(Some(d)) => d,
+ _ => continue,
+ };
+
+ // Sum post-purge sends across all partitions for this topic
+ let post_purge_sends: u64 = (1..=details.partitions_count)
+ .map(|p| watermarks.send_watermark(stream, topic_name, p))
+ .sum();
+
+ if details.messages_count > post_purge_sends && post_purge_sends > 0 {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::ShadowMismatch,
+ kind: "purge_data_retained",
+ description: format!(
+ "Server reports {} messages for {stream}/{topic_name} \
+ but only {post_purge_sends} were sent after purge",
+ details.messages_count
+ ),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ }
+ }
+ }
+
+ violations
+}
+
+/// Cross-client consistency: spawn a fresh client and compare get_streams results.
+pub async fn cross_client_consistency(
+ addr: &str,
+ transport: Transport,
+ prefixes: &NamespacePrefixes,
+) -> Vec<InvariantViolation> {
+ let mut violations = Vec::new();
+
+ let client_a = match create_client(addr, transport).await {
+ Ok(c) => c,
+ Err(e) => {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::Infrastructure,
+ kind: "cross_client_connect_failed",
+ description: format!("Client A: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ return violations;
+ }
+ };
+ let client_b = match create_client(addr, transport).await {
+ Ok(c) => c,
+ Err(e) => {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::Infrastructure,
+ kind: "cross_client_connect_failed",
+ description: format!("Client B: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ return violations;
+ }
+ };
+
+ let (a_result, b_result) = tokio::join!(client_a.get_streams(), client_b.get_streams());
+
+ let (streams_a, streams_b) = match (a_result, b_result) {
+ (Ok(a), Ok(b)) => (a, b),
+ (Err(e), _) | (_, Err(e)) => {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::Infrastructure,
+ kind: "cross_client_fetch_failed",
+ description: format!("Failed to get streams: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ return violations;
+ }
+ };
+
+ let names_a: std::collections::HashSet<&str> = streams_a
+ .iter()
+ .filter(|s| prefixes.matches(&s.name))
+ .map(|s| s.name.as_str())
+ .collect();
+ let names_b: std::collections::HashSet<&str> = streams_b
+ .iter()
+ .filter(|s| prefixes.matches(&s.name))
+ .map(|s| s.name.as_str())
+ .collect();
+
+ if names_a != names_b {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::CrossClient,
+ kind: "cross_client_inconsistency",
+ description: format!(
+ "Client A sees {} streams, Client B sees {}",
+ names_a.len(),
+ names_b.len()
+ ),
+ context: format!(
+ "only_in_a={:?}, only_in_b={:?}",
+ names_a.difference(&names_b).collect::<Vec<_>>(),
+ names_b.difference(&names_a).collect::<Vec<_>>()
+ ),
+ trace_anchor: None,
+ });
+ }
+
+ violations
+}
+
+/// Shared consumer group health check: joins fresh members, verifies rebalance,
+/// checks partition assignment uniqueness, polls concurrently, asserts sentinel delivery.
+pub struct ConsumerGroupHealthCheck<'a> {
+ pub label: &'a str,
+ pub server_address: &'a str,
+ pub transport: Transport,
+ pub admin_client: &'a IggyClient,
+ pub stream_id: &'a Identifier,
+ pub topic_id: &'a Identifier,
+ pub group_id: &'a Identifier,
+ pub fresh_member_count: u32,
+ pub rebalance_timeout: Duration,
+ pub sentinel_payload: &'a [u8],
+}
+
+impl ConsumerGroupHealthCheck<'_> {
+ pub async fn run(&self) -> Vec<InvariantViolation> {
+ match self.run_inner().await {
+ Ok(violations) => violations,
+ Err(v) => vec![v],
+ }
+ }
+
+ async fn run_inner(&self) -> Result<Vec<InvariantViolation>, InvariantViolation> {
+ let label = self.label;
+ let mut violations = Vec::new();
+
+ info!(
+ "Verify [{label}]: joining {} fresh TCP members",
+ self.fresh_member_count
+ );
+ let mut fresh_clients = Vec::with_capacity(self.fresh_member_count as usize);
+ for i in 0..self.fresh_member_count {
+ let client = create_client(self.server_address, self.transport)
+ .await
+ .map_err(|e| InvariantViolation {
+ category: ViolationCategory::Infrastructure,
+ kind: "fresh_client_create_failed",
+ description: format!("[{label}] Failed to create fresh client #{i}: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ })?;
+ client
+ .join_consumer_group(self.stream_id, self.topic_id, self.group_id)
+ .await
+ .map_err(|e| InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "fresh_join_failed",
+ description: format!("[{label}] Fresh client #{i} failed to join group: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ })?;
+ fresh_clients.push(client);
+ }
+
+ crate::eventually::await_members_count(
+ self.admin_client,
+ self.stream_id,
+ self.topic_id,
+ self.group_id,
+ self.fresh_member_count,
+ self.rebalance_timeout,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "fresh_members_never_joined",
+ description: format!(
+ "[{label}] Expected {} fresh members, timed out: {e}",
+ self.fresh_member_count
+ ),
+ context: String::new(),
+ trace_anchor: None,
+ })?;
+
+ let details = crate::eventually::await_rebalance_complete(
+ self.admin_client,
+ self.stream_id,
+ self.topic_id,
+ self.group_id,
+ self.rebalance_timeout,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "rebalance_incomplete",
+ description: format!(
+ "[{label}] Rebalance did not complete within {:?}: {e}",
+ self.rebalance_timeout
+ ),
+ context: String::new(),
+ trace_anchor: None,
+ })?;
+
+ info!(
+ "Verify [{label}]: rebalance complete - {} members, {} partitions",
+ details.members_count, details.partitions_count
+ );
+
+ let mut partition_owners: HashMap<u32, u32> = HashMap::new();
+ for member in &details.members {
+ for &partition in &member.partitions {
+ if let Some(&prev_owner) = partition_owners.get(&partition) {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "duplicate_partition_assignment",
+ description: format!(
+ "[{label}] Partition {partition} assigned to multiple members"
+ ),
+ context: format!("member_id={} and member_id={prev_owner}", member.id),
+ trace_anchor: None,
+ });
+ } else {
+ partition_owners.insert(partition, member.id);
+ }
+ }
+ }
+
+ info!(
+ "Verify [{label}]: polling concurrently from {} fresh members",
+ self.fresh_member_count
+ );
+ let mut poll_tasks = JoinSet::new();
+ for client in fresh_clients {
+ let s_id = self.stream_id.clone();
+ let t_id = self.topic_id.clone();
+ let g_id = self.group_id.clone();
+ poll_tasks.spawn(async move {
+ let consumer = Consumer::group(g_id);
+ client
+ .poll_messages(
+ &s_id,
+ &t_id,
+ None,
+ &consumer,
+ &PollingStrategy::next(),
+ 10,
+ false,
+ )
+ .await
+ });
+ }
+
+ let sentinel = self.sentinel_payload;
+ let mut sentinel_receivers = 0u32;
+ while let Some(result) = poll_tasks.join_next().await {
+ match result {
+ Ok(Ok(polled)) => {
+ if polled.messages.iter().any(|m| m.payload == sentinel) {
+ sentinel_receivers += 1;
+ }
+ }
+ Ok(Err(e)) => {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "poll_failed",
+ description: format!("[{label}] Fresh member poll failed: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ }
+ Err(e) => {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::Infrastructure,
+ kind: "poll_task_panicked",
+ description: format!("[{label}] Poll task panicked: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ }
+ }
+ }
+
+ if sentinel_receivers == 0 {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "sentinel_lost",
+ description: format!("[{label}] No fresh member received the sentinel message"),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ } else if sentinel_receivers > 1 {
+ violations.push(InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "duplicate_sentinel_delivery",
+ description: format!(
+ "[{label}] {sentinel_receivers} members received the sentinel (expected exactly 1)"
+ ),
+ context: String::new(),
+ trace_anchor: None,
+ });
+ }
+
+ info!("Verify [{label}]: health check passed");
+ Ok(violations)
+ }
+}
diff --git a/core/chaos/src/main.rs b/core/chaos/src/main.rs
new file mode 100644
index 000000000..2c505dc6e
--- /dev/null
+++ b/core/chaos/src/main.rs
@@ -0,0 +1,117 @@
+/* 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.
+ */
+
+mod args;
+mod client;
+mod error;
+mod eventually;
+mod in_flight;
+mod invariants;
+mod ops;
+mod payload;
+mod perfetto;
+mod replay;
+mod report;
+mod runner;
+mod safe_name;
+mod scenarios;
+mod shadow;
+mod trace;
+mod violation;
+mod watermark;
+mod worker;
+
+use args::{Command, IggyLabArgs};
+use clap::Parser;
+use error::LabError;
+use replay::LabReplay;
+use runner::LabRunner;
+use tracing_subscriber::EnvFilter;
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+
+#[tokio::main]
+async fn main() -> Result<(), LabError> {
+ let args = IggyLabArgs::parse();
+
+ match args.command {
+ Command::List => {
+ println!("Available scenarios:\n");
+ for (name, description) in scenarios::list_scenarios() {
+ let first_line = description.lines().next().unwrap_or(description);
+ println!(" {name:<25} {first_line}");
+ }
+ }
+ Command::Explain { scenario } => {
+ let s = scenarios::create_scenario(scenario);
+ println!("{}\n", s.name());
+ println!("{}", s.describe());
+ }
+ Command::Run(run_args) => {
+ init_tracing();
+
+ let runner = LabRunner::new(run_args);
+ let outcome = runner.run().await?;
+
+ if !outcome.passed {
+ eprintln!("Run ID: {}", outcome.run_id);
+ std::process::exit(1);
+ }
+ }
+ Command::Viz(viz_args) => match perfetto::convert(&viz_args.bundle_dir) {
+ Ok(stats) => {
+ let out = viz_args.bundle_dir.join("trace.perfetto.json");
+ println!(
+ "Wrote {} events to {} ({} unmatched outcomes)",
+ stats.events,
+ out.display(),
+ stats.unmatched_outcomes,
+ );
+ }
+ Err(e) => {
+ eprintln!("Failed to convert trace: {e}");
+ std::process::exit(1);
+ }
+ },
+ Command::Replay(replay_args) => {
+ init_tracing();
+
+ let outcome = LabReplay::new(replay_args).run().await?;
+
+ println!(
+ "Replayed {} ops - {} divergences ({} concerning)",
+ outcome.total_ops, outcome.divergences_total, outcome.divergences_concerning
+ );
+
+ if outcome.divergences_concerning > 0 {
+ std::process::exit(1);
+ }
+ }
+ }
+
+ Ok(())
+}
+
+fn init_tracing() {
+ let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
+
+ tracing_subscriber::registry()
+ .with(env_filter)
+ .with(tracing_subscriber::fmt::layer().with_ansi(true))
+ .init();
+}
diff --git a/core/chaos/src/ops/classify.rs b/core/chaos/src/ops/classify.rs
new file mode 100644
index 000000000..79bfa69d3
--- /dev/null
+++ b/core/chaos/src/ops/classify.rs
@@ -0,0 +1,322 @@
+/* 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 super::{ErrorClass, Op};
+use crate::in_flight::InFlightOps;
+use crate::shadow::ShadowState;
+use iggy_common::IggyError;
+
+impl ErrorClass {
+ /// Shadow-free classifier for contexts without shadow state (e.g. replay).
+ /// Treats all resource-not-found as concurrent races, connection errors as
+ /// transient, everything else as a server bug.
+ pub fn coarse(err: &IggyError) -> Self {
+ match err {
+ IggyError::Disconnected
+ | IggyError::NotConnected
+ | IggyError::CannotEstablishConnection
+ | IggyError::Unauthenticated
+ | IggyError::Unauthorized
+ | IggyError::TcpError
+ | IggyError::QuicError
+ | IggyError::StaleClient
+ | IggyError::ClientShutdown
+ | IggyError::ConnectionClosed
+ | IggyError::WebSocketError
+ | IggyError::WebSocketConnectionError
+ | IggyError::WebSocketCloseError
+ | IggyError::WebSocketReceiveError
+ | IggyError::WebSocketSendError
+ | IggyError::HttpError(_)
+ | IggyError::HttpResponseError(_, _)
+ | IggyError::FeatureUnavailable => Self::Transient,
+
+ IggyError::StreamNameAlreadyExists(_)
+ | IggyError::TopicNameAlreadyExists(_, _)
+ | IggyError::ConsumerGroupNameAlreadyExists(_, _)
+ | IggyError::StreamNameNotFound(_)
+ | IggyError::TopicNameNotFound(_, _)
+ | IggyError::StreamIdNotFound(_)
+ | IggyError::TopicIdNotFound(_, _)
+ | IggyError::PartitionNotFound(_, _, _)
+ | IggyError::SegmentNotFound
+ | IggyError::ConsumerGroupIdNotFound(_, _)
+ | IggyError::ConsumerGroupNameNotFound(_, _)
+ | IggyError::ConsumerGroupMemberNotFound(_, _, _)
+ | IggyError::ResourceNotFound(_)
+ | IggyError::InvalidPartitionsCount
+ | IggyError::TooManyPartitions
+ | IggyError::NoPartitions(_, _) => Self::ExpectedConcurrent,
+
+ _ => Self::ServerBug,
+ }
+ }
+}
+
+impl Op {
+ /// Classify an IggyError using shadow state and in-flight tracking to distinguish
+ /// between concurrent races (benign), server bugs, and transient failures.
+ pub fn classify_error(
+ &self,
+ err: &IggyError,
+ state: &ShadowState,
+ in_flight: &InFlightOps,
+ ) -> ErrorClass {
+ match err {
+ IggyError::Disconnected
+ | IggyError::NotConnected
+ | IggyError::CannotEstablishConnection
+ | IggyError::Unauthenticated
+ | IggyError::Unauthorized
+ | IggyError::TcpError
+ | IggyError::QuicError
+ | IggyError::StaleClient
+ | IggyError::ClientShutdown
+ | IggyError::ConnectionClosed
+ | IggyError::WebSocketError
+ | IggyError::WebSocketConnectionError
+ | IggyError::WebSocketCloseError
+ | IggyError::WebSocketReceiveError
+ | IggyError::WebSocketSendError
+ | IggyError::HttpError(_)
+ | IggyError::HttpResponseError(_, _)
+ | IggyError::FeatureUnavailable => ErrorClass::Transient,
+
+ IggyError::StreamNameNotFound(name) => {
+ if state.was_recently_deleted(name)
+ || !state.stream_exists(name)
+ || in_flight.contains(name)
+ {
+ ErrorClass::ExpectedConcurrent
+ } else {
+ ErrorClass::ServerBug
+ }
+ }
+
+ IggyError::StreamIdNotFound(_) => {
+ if let Some(stream_name) = self.stream_name()
+ && (state.was_recently_deleted(stream_name)
+ || !state.stream_exists(stream_name)
+ || in_flight.contains(stream_name))
+ {
+ return ErrorClass::ExpectedConcurrent;
+ }
+ ErrorClass::ServerBug
+ }
+
+ IggyError::TopicNameNotFound(_, _) | IggyError::TopicIdNotFound(_, _) => {
+ if let (Some(s), Some(t)) = (self.stream_name(), self.topic_name())
+ && (state.was_recently_deleted(&format!("{s}/{t}"))
+ || state.was_recently_deleted(s)
+ || !state.topic_exists(s, t)
+ || in_flight.contains(&format!("{s}/{t}"))
+ || in_flight.contains(s))
+ {
+ return ErrorClass::ExpectedConcurrent;
+ }
+ ErrorClass::ServerBug
+ }
+
+ IggyError::StreamNameAlreadyExists(name) => {
+ if state.stream_exists(name) {
+ ErrorClass::ExpectedConcurrent
+ } else {
+ ErrorClass::ServerBug
+ }
+ }
+
+ IggyError::TopicNameAlreadyExists(_, _) => {
+ if let (Some(s), Some(t)) = (self.stream_name(), self.topic_name())
+ && state.topic_exists(s, t)
+ {
+ return ErrorClass::ExpectedConcurrent;
+ }
+ ErrorClass::ServerBug
+ }
+
+ IggyError::ConsumerGroupNameAlreadyExists(_, _) => {
+ if let (Some(s), Some(t)) = (self.stream_name(), self.topic_name()) {
+ let group_name = match self {
+ Op::CreateConsumerGroup { name, .. } => Some(name.as_str()),
+ _ => None,
+ };
+ if let Some(gn) = group_name
+ && state
+ .get_topic(s, t)
+ .is_some_and(|t| t.consumer_groups.contains(gn))
+ {
+ return ErrorClass::ExpectedConcurrent;
+ }
+ }
+ ErrorClass::ServerBug
+ }
+
+ IggyError::PartitionNotFound(_, _, _)
+ | IggyError::SegmentNotFound
+ | IggyError::ConsumerGroupIdNotFound(_, _)
+ | IggyError::ConsumerGroupNameNotFound(_, _)
+ | IggyError::ConsumerGroupMemberNotFound(_, _, _)
+ | IggyError::ResourceNotFound(_)
+ | IggyError::InvalidPartitionsCount
+ | IggyError::TooManyPartitions
+ | IggyError::NoPartitions(_, _) => ErrorClass::ExpectedConcurrent,
+
+ _ => ErrorClass::ServerBug,
+ }
+ }
+
+ pub(crate) fn stream_name(&self) -> Option<&str> {
+ match self {
+ Op::CreateStream { name } | Op::DeleteStream { name } | Op::PurgeStream { name } => {
+ Some(name)
+ }
+ Op::CreateTopic { stream, .. }
+ | Op::DeleteTopic { stream, .. }
+ | Op::PurgeTopic { stream, .. }
+ | Op::SendMessages { stream, .. }
+ | Op::PollMessages { stream, .. }
+ | Op::DeleteSegments { stream, .. }
+ | Op::CreatePartitions { stream, .. }
+ | Op::DeletePartitions { stream, .. }
+ | Op::CreateConsumerGroup { stream, .. }
+ | Op::DeleteConsumerGroup { stream, .. }
+ | Op::JoinConsumerGroup { stream, .. }
+ | Op::LeaveConsumerGroup { stream, .. }
+ | Op::GetConsumerGroup { stream, .. }
+ | Op::PollGroupMessages { stream, .. }
+ | Op::StoreConsumerOffset { stream, .. }
+ | Op::GetStreamDetails { stream }
+ | Op::GetTopicDetails { stream, .. } => Some(stream),
+ Op::GetStreams | Op::GetStats => None,
+ }
+ }
+
+ pub(crate) fn topic_name(&self) -> Option<&str> {
+ match self {
+ Op::CreateStream { .. }
+ | Op::DeleteStream { .. }
+ | Op::PurgeStream { .. }
+ | Op::GetStreams
+ | Op::GetStats
+ | Op::GetStreamDetails { .. } => None,
+ Op::CreateTopic { name, .. } => Some(name),
+ Op::DeleteTopic { topic, .. }
+ | Op::PurgeTopic { topic, .. }
+ | Op::SendMessages { topic, .. }
+ | Op::PollMessages { topic, .. }
+ | Op::DeleteSegments { topic, .. }
+ | Op::CreatePartitions { topic, .. }
+ | Op::DeletePartitions { topic, .. }
+ | Op::CreateConsumerGroup { topic, .. }
+ | Op::DeleteConsumerGroup { topic, .. }
+ | Op::JoinConsumerGroup { topic, .. }
+ | Op::LeaveConsumerGroup { topic, .. }
+ | Op::GetConsumerGroup { topic, .. }
+ | Op::PollGroupMessages { topic, .. }
+ | Op::StoreConsumerOffset { topic, .. }
+ | Op::GetTopicDetails { topic, .. } => Some(topic),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iggy_common::Identifier;
+
+ fn id(n: u32) -> Identifier {
+ Identifier::numeric(n).unwrap()
+ }
+
+ /// Verify that every explicitly-handled IggyError variant in `ErrorClass::coarse`
+ /// is classified as intended (not falling through to the `_ => ServerBug` wildcard).
+ /// If a new SDK variant is added to a match arm, this test catches it at compile time.
+ #[test]
+ fn coarse_explicit_variants_not_server_bug() {
+ let transient_variants: Vec<IggyError> = vec![
+ IggyError::Disconnected,
+ IggyError::NotConnected,
+ IggyError::CannotEstablishConnection,
+ IggyError::Unauthenticated,
+ IggyError::Unauthorized,
+ IggyError::TcpError,
+ IggyError::QuicError,
+ IggyError::StaleClient,
+ IggyError::ClientShutdown,
+ IggyError::ConnectionClosed,
+ IggyError::WebSocketError,
+ IggyError::WebSocketConnectionError,
+ IggyError::WebSocketCloseError,
+ IggyError::WebSocketReceiveError,
+ IggyError::WebSocketSendError,
+ IggyError::HttpError(String::new()),
+ IggyError::HttpResponseError(0, String::new()),
+ IggyError::FeatureUnavailable,
+ ];
+ for err in &transient_variants {
+ assert_eq!(
+ ErrorClass::coarse(err),
+ ErrorClass::Transient,
+ "Expected Transient for {err}"
+ );
+ }
+
+ let expected_concurrent_variants: Vec<IggyError> = vec![
+ IggyError::StreamNameAlreadyExists(String::new()),
+ IggyError::TopicNameAlreadyExists(String::new(), id(1)),
+ IggyError::ConsumerGroupNameAlreadyExists(String::new(), id(1)),
+ IggyError::StreamNameNotFound(String::new()),
+ IggyError::TopicNameNotFound(String::new(), String::new()),
+ IggyError::StreamIdNotFound(id(1)),
+ IggyError::TopicIdNotFound(id(1), id(1)),
+ IggyError::PartitionNotFound(0, id(1), id(1)),
+ IggyError::SegmentNotFound,
+ IggyError::ConsumerGroupIdNotFound(id(1), id(1)),
+ IggyError::ConsumerGroupNameNotFound(String::new(), id(1)),
+ IggyError::ConsumerGroupMemberNotFound(0, id(1), id(1)),
+ IggyError::ResourceNotFound(String::new()),
+ IggyError::InvalidPartitionsCount,
+ IggyError::TooManyPartitions,
+ IggyError::NoPartitions(id(1), id(1)),
+ ];
+ for err in &expected_concurrent_variants {
+ assert_eq!(
+ ErrorClass::coarse(err),
+ ErrorClass::ExpectedConcurrent,
+ "Expected ExpectedConcurrent for {err}"
+ );
+ }
+ }
+
+ /// Verify that unhandled variants fall through to ServerBug in coarse classification.
+ #[test]
+ fn coarse_unhandled_is_server_bug() {
+ let unhandled = vec![
+ IggyError::InvalidCommand,
+ IggyError::InvalidFormat,
+ IggyError::CannotCreateStream(0),
+ ];
+ for err in &unhandled {
+ assert_eq!(
+ ErrorClass::coarse(err),
+ ErrorClass::ServerBug,
+ "Expected ServerBug for unhandled {err}"
+ );
+ }
+ }
+}
diff --git a/core/chaos/src/ops/execute.rs b/core/chaos/src/ops/execute.rs
new file mode 100644
index 000000000..39223cc08
--- /dev/null
+++ b/core/chaos/src/ops/execute.rs
@@ -0,0 +1,379 @@
+/* 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 super::{Op, OpDetail};
+use crate::error::LabError;
+use crate::payload::{self, PayloadGenerator};
+use crate::scenarios::NamespacePrefixes;
+use bytes::Bytes;
+use iggy::prelude::*;
+use iggy_common::IggyError;
+use tracing::warn;
+
+fn ident(value: &str) -> Result<Identifier, LabError> {
+ Identifier::from_str_value(value).map_err(|_| LabError::InvalidIdentifier(value.to_owned()))
+}
+
+fn ident_numeric(value: u32) -> Result<Identifier, LabError> {
+ Identifier::numeric(value).map_err(|_| LabError::InvalidIdentifier(value.to_string()))
+}
+
+impl Op {
+ pub async fn execute(
+ &self,
+ client: &IggyClient,
+ payload_gen: &mut PayloadGenerator,
+ prefixes: &NamespacePrefixes,
+ ) -> Result<Option<OpDetail>, LabError> {
+ match self.stream_name() {
+ Some(name) if prefixes.matches(name) => {}
+ Some(name) => return Err(LabError::PrefixViolation(name.to_owned())),
+ None => match self {
+ Op::GetStreams | Op::GetStats => {}
+ _ => {
+ return Err(LabError::PrefixViolation(format!(
+ "op {} has no stream_name but is not allowlisted",
+ self.kind_tag()
+ )));
+ }
+ },
+ }
+
+ match self {
+ Op::CreateStream { name } => {
+ let details = client.create_stream(name).await?;
+ Ok(Some(OpDetail::Created { id: details.id }))
+ }
+
+ Op::DeleteStream { name } => {
+ client.delete_stream(&ident(name)?).await?;
+ Ok(None)
+ }
+
+ Op::PurgeStream { name } => {
+ client.purge_stream(&ident(name)?).await?;
+ Ok(None)
+ }
+
+ Op::CreateTopic {
+ stream,
+ name,
+ partitions,
+ } => {
+ let stream_id = ident(stream)?;
+ let details = client
+ .create_topic(
+ &stream_id,
+ name,
+ *partitions,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::Unlimited,
+ )
+ .await?;
+ Ok(Some(OpDetail::Created { id: details.id }))
+ }
+
+ Op::DeleteTopic { stream, topic } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ client.delete_topic(&stream_id, &topic_id).await?;
+ Ok(None)
+ }
+
+ Op::PurgeTopic { stream, topic } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ client.purge_topic(&stream_id, &topic_id).await?;
+ Ok(None)
+ }
+
+ Op::SendMessages {
+ stream,
+ topic,
+ partition,
+ count,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ let partitioning = Partitioning::partition_id(*partition);
+ let mut messages: Vec<IggyMessage> = Vec::with_capacity(*count as usize);
+ for _ in 0..*count {
+ let payload = payload_gen.next();
+ messages.push(
+ IggyMessage::builder()
+ .payload(Bytes::copy_from_slice(payload))
+ .build()
+ .unwrap(),
+ );
+ }
+ client
+ .send_messages(&stream_id, &topic_id, &partitioning, &mut messages)
+ .await?;
+ Ok(Some(OpDetail::Sent { count: *count }))
+ }
+
+ Op::PollMessages {
+ stream,
+ topic,
+ partition,
+ count,
+ consumer_id,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ let consumer = Consumer::new(ident_numeric(*consumer_id)?);
+ let strategy = PollingStrategy::next();
+ let polled = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(*partition),
+ &consumer,
+ &strategy,
+ *count,
+ false,
+ )
+ .await?;
+ verify_polled_payloads(&polled);
+ Ok(Some(OpDetail::Polled {
+ received: polled.messages.len(),
+ offset: polled.current_offset,
+ }))
+ }
+
+ Op::DeleteSegments {
+ stream,
+ topic,
+ partition,
+ count,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ client
+ .delete_segments(&stream_id, &topic_id, *partition, *count)
+ .await?;
+ Ok(None)
+ }
+
+ Op::CreatePartitions {
+ stream,
+ topic,
+ count,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ client
+ .create_partitions(&stream_id, &topic_id, *count)
+ .await?;
+ Ok(None)
+ }
+
+ Op::DeletePartitions {
+ stream,
+ topic,
+ count,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ client
+ .delete_partitions(&stream_id, &topic_id, *count)
+ .await?;
+ Ok(None)
+ }
+
+ Op::CreateConsumerGroup {
+ stream,
+ topic,
+ name,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ let details = client
+ .create_consumer_group(&stream_id, &topic_id, name)
+ .await?;
+ Ok(Some(OpDetail::Created { id: details.id }))
+ }
+
+ Op::DeleteConsumerGroup {
+ stream,
+ topic,
+ name,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ let group_id = ident(name)?;
+ client
+ .delete_consumer_group(&stream_id, &topic_id, &group_id)
+ .await?;
+ Ok(None)
+ }
+
+ Op::JoinConsumerGroup {
+ stream,
+ topic,
+ group,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ let group_id = ident(group)?;
+ client
+ .join_consumer_group(&stream_id, &topic_id, &group_id)
+ .await?;
+ Ok(None)
+ }
+
+ Op::LeaveConsumerGroup {
+ stream,
+ topic,
+ group,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ let group_id = ident(group)?;
+ client
+ .leave_consumer_group(&stream_id, &topic_id, &group_id)
+ .await?;
+ Ok(None)
+ }
+
+ Op::GetConsumerGroup {
+ stream,
+ topic,
+ group,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ let group_id = ident(group)?;
+ match client
+ .get_consumer_group(&stream_id, &topic_id, &group_id)
+ .await?
+ {
+ Some(details) => Ok(Some(OpDetail::GroupInfo {
+ members: details.members_count,
+ partitions: details.partitions_count,
+ })),
+ None => {
+ Err(IggyError::ConsumerGroupNameNotFound(group.clone(), topic_id).into())
+ }
+ }
+ }
+
+ Op::PollGroupMessages {
+ stream,
+ topic,
+ group,
+ count,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ let consumer = Consumer::group(ident(group)?);
+ let strategy = PollingStrategy::next();
+ let polled = client
+ .poll_messages(
+ &stream_id, &topic_id, None, &consumer, &strategy, *count, true,
+ )
+ .await?;
+ verify_polled_payloads(&polled);
+ Ok(Some(OpDetail::GroupPolled {
+ received: polled.messages.len(),
+ partition_id: polled.partition_id,
+ offset: polled.current_offset,
+ }))
+ }
+
+ Op::StoreConsumerOffset {
+ stream,
+ topic,
+ partition,
+ offset,
+ consumer_id,
+ } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ let consumer = Consumer::new(ident_numeric(*consumer_id)?);
+ client
+ .store_consumer_offset(
+ &consumer,
+ &stream_id,
+ &topic_id,
+ Some(*partition),
+ *offset,
+ )
+ .await?;
+ Ok(None)
+ }
+
+ Op::GetStreams => {
+ let streams = client.get_streams().await?;
+ Ok(Some(OpDetail::StreamList {
+ count: streams.len(),
+ }))
+ }
+
+ Op::GetStreamDetails { stream } => {
+ let stream_id = ident(stream)?;
+ match client.get_stream(&stream_id).await? {
+ Some(details) => Ok(Some(OpDetail::StreamInfo {
+ topics: details.topics.len(),
+ })),
+ None => Err(IggyError::StreamNameNotFound(stream.clone()).into()),
+ }
+ }
+
+ Op::GetTopicDetails { stream, topic } => {
+ let stream_id = ident(stream)?;
+ let topic_id = ident(topic)?;
+ match client.get_topic(&stream_id, &topic_id).await? {
+ Some(details) => Ok(Some(OpDetail::TopicInfo {
+ partitions: details.partitions_count,
+ })),
+ None => Err(IggyError::TopicNameNotFound(stream.clone(), topic.clone()).into()),
+ }
+ }
+
+ Op::GetStats => {
+ let stats = client.get_stats().await?;
+ Ok(Some(OpDetail::Stats {
+ streams: stats.streams_count,
+ topics: stats.topics_count,
+ partitions: stats.partitions_count,
+ }))
+ }
+ }
+ }
+}
+
+/// Best-effort CRC verification of polled messages. Warns on corruption but does not
+/// fail the operation - messages produced by scenario setup (sentinels) or external
+/// sources won't have the chaos header format and are silently skipped.
+fn verify_polled_payloads(polled: &PolledMessages) {
+ for msg in &polled.messages {
+ if msg.payload.len() < 16 {
+ continue;
+ }
+ if let Err(e) = payload::verify_payload(&msg.payload) {
+ warn!(
+ partition_id = polled.partition_id,
+ offset = msg.header.offset,
+ "Payload corruption detected: {e}"
+ );
+ }
+ }
+}
diff --git a/core/chaos/src/ops/generate.rs b/core/chaos/src/ops/generate.rs
new file mode 100644
index 000000000..9ac1bb3f0
--- /dev/null
+++ b/core/chaos/src/ops/generate.rs
@@ -0,0 +1,281 @@
+/* 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 super::{Op, OpKind};
+use crate::safe_name::SafeResourceName;
+use crate::scenarios::{NamespacePrefixes, OriginScope, TargetNamespace};
+use crate::shadow::ShadowState;
+use rand::{Rng, RngExt};
+
+const DEFAULT_PARTITIONS: u32 = 3;
+
+fn pick_random_group(
+ state: &ShadowState,
+ rng: &mut impl Rng,
+ scope: OriginScope,
+ namespace: TargetNamespace,
+) -> Option<(String, String, String)> {
+ let (stream, topic) = state.random_topic(rng, scope, namespace)?;
+ let groups = &state.streams[stream].topics[topic].consumer_groups;
+ if groups.is_empty() {
+ return None;
+ }
+ let idx = rng.random_range(0..groups.len());
+ let group = groups.iter().nth(idx)?.to_owned();
+ Some((stream.to_owned(), topic.to_owned(), group))
+}
+
+impl Op {
+ /// Generate a concrete Op from an OpKind, resolving targets from the shadow state.
+ /// Returns None if the precondition for this kind cannot be met.
+ ///
+ /// Resource-destructive ops (delete/purge stream/topic) are forced to target
+ /// churn-namespace resources regardless of the lane's configured namespace.
+ #[allow(clippy::too_many_arguments)]
+ pub fn generate(
+ kind: OpKind,
+ state: &ShadowState,
+ rng: &mut impl Rng,
+ prefixes: &NamespacePrefixes,
+ namespace: TargetNamespace,
+ messages_per_batch: u32,
+ scope: OriginScope,
+ consumer_id: u32,
+ ) -> Option<Self> {
+ let effective_namespace = if kind.is_resource_destructive() {
+ TargetNamespace::Churn
+ } else {
+ namespace
+ };
+
+ match kind {
+ OpKind::CreateStream => {
+ let prefix = prefixes.create_prefix(namespace);
+ let name = SafeResourceName::random(prefix, rng);
+ Some(Op::CreateStream {
+ name: name.into_inner(),
+ })
+ }
+
+ OpKind::DeleteStream => {
+ let name = state.random_stream(rng, scope, effective_namespace)?;
+ Some(Op::DeleteStream {
+ name: name.to_owned(),
+ })
+ }
+
+ OpKind::PurgeStream => {
+ let name = state.random_stream(rng, scope, effective_namespace)?;
+ Some(Op::PurgeStream {
+ name: name.to_owned(),
+ })
+ }
+
+ OpKind::CreateTopic => {
+ let stream = state.random_stream(rng, scope, effective_namespace)?;
+ let name = SafeResourceName::random(&prefixes.base, rng);
+ Some(Op::CreateTopic {
+ stream: stream.to_owned(),
+ name: name.into_inner(),
+ partitions: DEFAULT_PARTITIONS,
+ })
+ }
+
+ OpKind::DeleteTopic => {
+ let (stream, topic) = state.random_topic(rng, scope, effective_namespace)?;
+ Some(Op::DeleteTopic {
+ stream: stream.to_owned(),
+ topic: topic.to_owned(),
+ })
+ }
+
+ OpKind::PurgeTopic => {
+ let (stream, topic) = state.random_topic(rng, scope, effective_namespace)?;
+ Some(Op::PurgeTopic {
+ stream: stream.to_owned(),
+ topic: topic.to_owned(),
+ })
+ }
+
+ OpKind::SendMessages => {
+ let (stream, topic, partitions) =
+ state.random_topic_with_partitions(rng, scope, effective_namespace)?;
+ if partitions == 0 {
+ return None;
+ }
+ let partition = rng.random_range(0..partitions);
+ Some(Op::SendMessages {
+ stream: stream.to_owned(),
+ topic: topic.to_owned(),
+ partition,
+ count: messages_per_batch,
+ })
+ }
+
+ OpKind::PollMessages => {
+ let (stream, topic, partitions) =
+ state.random_topic_with_partitions(rng, scope, effective_namespace)?;
+ if partitions == 0 {
+ return None;
+ }
+ let partition = rng.random_range(0..partitions);
+ Some(Op::PollMessages {
+ stream: stream.to_owned(),
+ topic: topic.to_owned(),
+ partition,
+ count: messages_per_batch,
+ consumer_id,
+ })
+ }
+
+ OpKind::DeleteSegments => {
+ let (stream, topic, partitions) =
+ state.random_topic_with_partitions(rng, scope, effective_namespace)?;
+ if partitions == 0 {
+ return None;
+ }
+ let partition = rng.random_range(0..partitions);
+ Some(Op::DeleteSegments {
+ stream: stream.to_owned(),
+ topic: topic.to_owned(),
+ partition,
+ count: 1,
+ })
+ }
+
+ OpKind::CreatePartitions => {
+ let (stream, topic) = state.random_topic(rng, scope, effective_namespace)?;
+ Some(Op::CreatePartitions {
+ stream: stream.to_owned(),
+ topic: topic.to_owned(),
+ count: 1,
+ })
+ }
+
+ OpKind::DeletePartitions => {
+ let (stream, topic, partitions) =
+ state.random_topic_with_partitions(rng, scope, effective_namespace)?;
+ if partitions <= 1 {
+ return None;
+ }
+ Some(Op::DeletePartitions {
+ stream: stream.to_owned(),
+ topic: topic.to_owned(),
+ count: 1,
+ })
+ }
+
+ OpKind::CreateConsumerGroup => {
+ let (stream, topic) = state.random_topic(rng, scope, effective_namespace)?;
+ let name = SafeResourceName::random(&prefixes.base, rng);
+ Some(Op::CreateConsumerGroup {
+ stream: stream.to_owned(),
+ topic: topic.to_owned(),
+ name: name.into_inner(),
+ })
+ }
+
+ OpKind::DeleteConsumerGroup => {
+ let (stream, topic, group) =
+ pick_random_group(state, rng, scope, effective_namespace)?;
+ Some(Op::DeleteConsumerGroup {
+ stream,
+ topic,
+ name: group,
+ })
+ }
+
+ OpKind::JoinConsumerGroup => {
+ let (stream, topic, group) =
+ pick_random_group(state, rng, scope, effective_namespace)?;
+ Some(Op::JoinConsumerGroup {
+ stream,
+ topic,
+ group,
+ })
+ }
+
+ OpKind::LeaveConsumerGroup => {
+ let (stream, topic, group) =
+ pick_random_group(state, rng, scope, effective_namespace)?;
+ Some(Op::LeaveConsumerGroup {
+ stream,
+ topic,
+ group,
+ })
+ }
+
+ OpKind::GetConsumerGroup => {
+ let (stream, topic, group) =
+ pick_random_group(state, rng, scope, effective_namespace)?;
+ Some(Op::GetConsumerGroup {
+ stream,
+ topic,
+ group,
+ })
+ }
+
+ OpKind::PollGroupMessages => {
+ let (stream, topic, group) =
+ pick_random_group(state, rng, scope, effective_namespace)?;
+ Some(Op::PollGroupMessages {
+ stream,
+ topic,
+ group,
+ count: messages_per_batch,
+ })
+ }
+
+ OpKind::StoreConsumerOffset => {
+ let (stream, topic, partitions) =
+ state.random_topic_with_partitions(rng, scope, effective_namespace)?;
+ if partitions == 0 {
+ return None;
+ }
+ let partition = rng.random_range(0..partitions);
+ let offset = rng.random_range(0..1000);
+ Some(Op::StoreConsumerOffset {
+ stream: stream.to_owned(),
+ topic: topic.to_owned(),
+ partition,
+ offset,
+ consumer_id,
+ })
+ }
+
+ OpKind::GetStreams => Some(Op::GetStreams),
+
+ OpKind::GetStreamDetails => {
+ let stream = state.random_stream(rng, scope, effective_namespace)?;
+ Some(Op::GetStreamDetails {
+ stream: stream.to_owned(),
+ })
+ }
+
+ OpKind::GetTopicDetails => {
+ let (stream, topic) = state.random_topic(rng, scope, effective_namespace)?;
+ Some(Op::GetTopicDetails {
+ stream: stream.to_owned(),
+ topic: topic.to_owned(),
+ })
+ }
+
+ OpKind::GetStats => Some(Op::GetStats),
+ }
+ }
+}
diff --git a/core/chaos/src/ops/mod.rs b/core/chaos/src/ops/mod.rs
new file mode 100644
index 000000000..32a05e514
--- /dev/null
+++ b/core/chaos/src/ops/mod.rs
@@ -0,0 +1,458 @@
+/* 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.
+ */
+
+pub mod classify;
+pub mod execute;
+pub mod generate;
+pub mod shadow_update;
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum Op {
+ CreateStream {
+ name: String,
+ },
+ DeleteStream {
+ name: String,
+ },
+ PurgeStream {
+ name: String,
+ },
+ CreateTopic {
+ stream: String,
+ name: String,
+ partitions: u32,
+ },
+ DeleteTopic {
+ stream: String,
+ topic: String,
+ },
+ PurgeTopic {
+ stream: String,
+ topic: String,
+ },
+ SendMessages {
+ stream: String,
+ topic: String,
+ partition: u32,
+ count: u32,
+ },
+ PollMessages {
+ stream: String,
+ topic: String,
+ partition: u32,
+ count: u32,
+ consumer_id: u32,
+ },
+ DeleteSegments {
+ stream: String,
+ topic: String,
+ partition: u32,
+ count: u32,
+ },
+ CreatePartitions {
+ stream: String,
+ topic: String,
+ count: u32,
+ },
+ DeletePartitions {
+ stream: String,
+ topic: String,
+ count: u32,
+ },
+ CreateConsumerGroup {
+ stream: String,
+ topic: String,
+ name: String,
+ },
+ DeleteConsumerGroup {
+ stream: String,
+ topic: String,
+ name: String,
+ },
+ JoinConsumerGroup {
+ stream: String,
+ topic: String,
+ group: String,
+ },
+ LeaveConsumerGroup {
+ stream: String,
+ topic: String,
+ group: String,
+ },
+ GetConsumerGroup {
+ stream: String,
+ topic: String,
+ group: String,
+ },
+ PollGroupMessages {
+ stream: String,
+ topic: String,
+ group: String,
+ count: u32,
+ },
+ StoreConsumerOffset {
+ stream: String,
+ topic: String,
+ partition: u32,
+ offset: u64,
+ consumer_id: u32,
+ },
+ GetStreams,
+ GetStreamDetails {
+ stream: String,
+ },
+ GetTopicDetails {
+ stream: String,
+ topic: String,
+ },
+ GetStats,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum OpDetail {
+ Created {
+ id: u32,
+ },
+ Sent {
+ count: u32,
+ },
+ Polled {
+ received: usize,
+ offset: u64,
+ },
+ GroupPolled {
+ received: usize,
+ partition_id: u32,
+ offset: u64,
+ },
+ GroupInfo {
+ members: u32,
+ partitions: u32,
+ },
+ StreamList {
+ count: usize,
+ },
+ StreamInfo {
+ topics: usize,
+ },
+ TopicInfo {
+ partitions: u32,
+ },
+ Stats {
+ streams: u32,
+ topics: u32,
+ partitions: u32,
+ },
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "status")]
+pub enum OpOutcome {
+ Success {
+ detail: Option<OpDetail>,
+ },
+ Error {
+ class: ErrorClass,
+ error: String,
+ context: String,
+ },
+}
+
+impl OpOutcome {
+ pub fn success_from(detail: Option<OpDetail>) -> Self {
+ Self::Success { detail }
+ }
+
+ pub fn error(class: ErrorClass, error: impl Into<String>, context: impl Into<String>) -> Self {
+ Self::Error {
+ class,
+ error: error.into(),
+ context: context.into(),
+ }
+ }
+
+ pub fn is_success(&self) -> bool {
+ matches!(self, Self::Success { .. })
+ }
+
+ pub fn error_class(&self) -> Option<ErrorClass> {
+ match self {
+ Self::Error { class, .. } => Some(*class),
+ Self::Success { .. } => None,
+ }
+ }
+
+ pub fn result_tag(&self) -> &'static str {
+ match self {
+ Self::Success { .. } => "ok",
+ Self::Error { class, .. } => class.tag(),
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ErrorClass {
+ /// Another worker deleted the resource concurrently - benign
+ ExpectedConcurrent,
+ /// Server returned something that contradicts shadow state
+ ServerBug,
+ /// Transient network/timeout error - retry-safe
+ Transient,
+}
+
+impl ErrorClass {
+ pub fn tag(self) -> &'static str {
+ match self {
+ Self::ExpectedConcurrent => "expected_error",
+ Self::ServerBug => "server_bug",
+ Self::Transient => "transient",
+ }
+ }
+}
+
+/// Selects which kind of operation to generate (before target resolution).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[allow(dead_code)]
+pub enum OpKind {
+ CreateStream,
+ DeleteStream,
+ PurgeStream,
+ CreateTopic,
+ DeleteTopic,
+ PurgeTopic,
+ SendMessages,
+ PollMessages,
+ DeleteSegments,
+ CreatePartitions,
+ DeletePartitions,
+ CreateConsumerGroup,
+ DeleteConsumerGroup,
+ JoinConsumerGroup,
+ LeaveConsumerGroup,
+ GetConsumerGroup,
+ PollGroupMessages,
+ StoreConsumerOffset,
+ GetStreams,
+ GetStreamDetails,
+ GetTopicDetails,
+ GetStats,
+}
+
+impl OpKind {
+ pub fn is_creative(self) -> bool {
+ matches!(
+ self,
+ Self::CreateStream
+ | Self::CreateTopic
+ | Self::CreatePartitions
+ | Self::CreateConsumerGroup
+ )
+ }
+
+ pub fn is_destructive(self) -> bool {
+ matches!(
+ self,
+ Self::DeleteStream
+ | Self::DeleteTopic
+ | Self::PurgeStream
+ | Self::PurgeTopic
+ | Self::DeleteSegments
+ | Self::DeletePartitions
+ | Self::DeleteConsumerGroup
+ )
+ }
+
+ /// Operations that remove or purge entire streams/topics.
+ /// Always restricted to churn namespace at generation time.
+ pub fn is_resource_destructive(self) -> bool {
+ matches!(
+ self,
+ Self::DeleteStream | Self::DeleteTopic | Self::PurgeStream | Self::PurgeTopic
+ )
+ }
+}
+
+impl Op {
+ pub fn kind_tag(&self) -> &'static str {
+ match self {
+ Self::CreateStream { .. } => "create_stream",
+ Self::DeleteStream { .. } => "delete_stream",
+ Self::PurgeStream { .. } => "purge_stream",
+ Self::CreateTopic { .. } => "create_topic",
+ Self::DeleteTopic { .. } => "delete_topic",
+ Self::PurgeTopic { .. } => "purge_topic",
+ Self::SendMessages { .. } => "send_messages",
+ Self::PollMessages { .. } => "poll_messages",
+ Self::DeleteSegments { .. } => "delete_segments",
+ Self::CreatePartitions { .. } => "create_partitions",
+ Self::DeletePartitions { .. } => "delete_partitions",
+ Self::CreateConsumerGroup { .. } => "create_consumer_group",
+ Self::DeleteConsumerGroup { .. } => "delete_consumer_group",
+ Self::JoinConsumerGroup { .. } => "join_consumer_group",
+ Self::LeaveConsumerGroup { .. } => "leave_consumer_group",
+ Self::GetConsumerGroup { .. } => "get_consumer_group",
+ Self::PollGroupMessages { .. } => "poll_group_messages",
+ Self::StoreConsumerOffset { .. } => "store_consumer_offset",
+ Self::GetStreams => "get_streams",
+ Self::GetStreamDetails { .. } => "get_stream_details",
+ Self::GetTopicDetails { .. } => "get_topic_details",
+ Self::GetStats => "get_stats",
+ }
+ }
+
+ /// Compact human-readable target context for traces and error messages.
+ pub fn context(&self) -> String {
+ match self {
+ Self::CreateStream { name }
+ | Self::DeleteStream { name }
+ | Self::PurgeStream { name } => format!("stream={name}"),
+
+ Self::CreateTopic {
+ stream,
+ name,
+ partitions,
+ } => format!("stream={stream} topic={name} partitions={partitions}"),
+
+ Self::DeleteTopic { stream, topic } | Self::PurgeTopic { stream, topic } => {
+ format!("stream={stream} topic={topic}")
+ }
+
+ Self::SendMessages {
+ stream,
+ topic,
+ partition,
+ count,
+ } => format!("stream={stream} topic={topic} partition={partition} count={count}"),
+
+ Self::PollMessages {
+ stream,
+ topic,
+ partition,
+ count,
+ consumer_id,
+ } => format!(
+ "stream={stream} topic={topic} partition={partition} count={count} consumer={consumer_id}"
+ ),
+
+ Self::DeleteSegments {
+ stream,
+ topic,
+ partition,
+ count,
+ } => format!("stream={stream} topic={topic} partition={partition} count={count}"),
+
+ Self::CreatePartitions {
+ stream,
+ topic,
+ count,
+ }
+ | Self::DeletePartitions {
+ stream,
+ topic,
+ count,
+ } => format!("stream={stream} topic={topic} count={count}"),
+
+ Self::CreateConsumerGroup {
+ stream,
+ topic,
+ name,
+ }
+ | Self::DeleteConsumerGroup {
+ stream,
+ topic,
+ name,
+ } => format!("stream={stream} topic={topic} group={name}"),
+
+ Self::JoinConsumerGroup {
+ stream,
+ topic,
+ group,
+ }
+ | Self::LeaveConsumerGroup {
+ stream,
+ topic,
+ group,
+ }
+ | Self::GetConsumerGroup {
+ stream,
+ topic,
+ group,
+ } => format!("stream={stream} topic={topic} group={group}"),
+
+ Self::PollGroupMessages {
+ stream,
+ topic,
+ group,
+ count,
+ } => format!("stream={stream} topic={topic} group={group} count={count}"),
+
+ Self::StoreConsumerOffset {
+ stream,
+ topic,
+ partition,
+ offset,
+ consumer_id,
+ } => format!(
+ "stream={stream} topic={topic} partition={partition} offset={offset} consumer={consumer_id}"
+ ),
+
+ Self::GetStreams => "all".into(),
+ Self::GetStreamDetails { stream } => format!("stream={stream}"),
+ Self::GetTopicDetails { stream, topic } => {
+ format!("stream={stream} topic={topic}")
+ }
+ Self::GetStats => "server".into(),
+ }
+ }
+
+ /// Resource key for ops that destroy resource visibility (delete/purge).
+ /// Used by `InFlightOps` to bridge the TOCTOU gap between server execution
+ /// and shadow state update.
+ pub fn in_flight_key(&self) -> Option<String> {
+ match self {
+ Self::DeleteStream { name } | Self::PurgeStream { name } => Some(name.clone()),
+ Self::DeleteTopic { stream, topic } | Self::PurgeTopic { stream, topic } => {
+ Some(format!("{stream}/{topic}"))
+ }
+ _ => None,
+ }
+ }
+
+ /// Whether this op mutates shadow state on success.
+ /// Non-mutating ops (data-plane reads/writes, gets) can skip the write lock.
+ pub fn mutates_shadow(&self) -> bool {
+ matches!(
+ self,
+ Self::CreateStream { .. }
+ | Self::DeleteStream { .. }
+ | Self::PurgeStream { .. }
+ | Self::CreateTopic { .. }
+ | Self::DeleteTopic { .. }
+ | Self::PurgeTopic { .. }
+ | Self::CreatePartitions { .. }
+ | Self::DeletePartitions { .. }
+ | Self::CreateConsumerGroup { .. }
+ | Self::DeleteConsumerGroup { .. }
+ )
+ }
+}
diff --git a/core/chaos/src/ops/shadow_update.rs b/core/chaos/src/ops/shadow_update.rs
new file mode 100644
index 000000000..b5773cb85
--- /dev/null
+++ b/core/chaos/src/ops/shadow_update.rs
@@ -0,0 +1,173 @@
+/* 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 super::{Op, OpOutcome};
+use crate::shadow::{ResourceOrigin, ShadowState};
+use iggy_common::IggyError;
+
+impl Op {
+ /// Update the shadow state to reflect a completed operation.
+ /// Only mutates on success - failed ops leave shadow unchanged.
+ pub fn update_shadow(&self, state: &mut ShadowState, outcome: &OpOutcome) {
+ if !outcome.is_success() {
+ return;
+ }
+
+ match self {
+ Op::CreateStream { name } => {
+ state.apply_create_stream(name.clone(), ResourceOrigin::Dynamic);
+ }
+ Op::DeleteStream { name } => {
+ state.apply_delete_stream(name);
+ }
+ Op::PurgeStream { name } => {
+ state.record_purge(name.clone(), None);
+ }
+ Op::CreateTopic {
+ stream,
+ name,
+ partitions,
+ } => {
+ state.apply_create_topic(
+ stream,
+ name.clone(),
+ *partitions,
+ ResourceOrigin::Dynamic,
+ );
+ }
+ Op::DeleteTopic { stream, topic } => {
+ state.apply_delete_topic(stream, topic);
+ }
+ Op::PurgeTopic { stream, topic } => {
+ state.record_purge(stream.clone(), Some(topic.clone()));
+ }
+ Op::SendMessages { .. } => {}
+ Op::PollMessages { .. } => {}
+ Op::DeleteSegments { .. } => {}
+ Op::CreatePartitions {
+ stream,
+ topic,
+ count,
+ } => {
+ state.apply_create_partitions(stream, topic, *count);
+ }
+ Op::DeletePartitions {
+ stream,
+ topic,
+ count,
+ } => {
+ state.apply_delete_partitions(stream, topic, *count);
+ }
+ Op::CreateConsumerGroup {
+ stream,
+ topic,
+ name,
+ } => {
+ state.apply_create_consumer_group(stream, topic, name.clone());
+ }
+ Op::DeleteConsumerGroup {
+ stream,
+ topic,
+ name,
+ } => {
+ state.apply_delete_consumer_group(stream, topic, name);
+ }
+ Op::StoreConsumerOffset { .. }
+ | Op::JoinConsumerGroup { .. }
+ | Op::LeaveConsumerGroup { .. }
+ | Op::GetConsumerGroup { .. }
+ | Op::PollGroupMessages { .. }
+ | Op::GetStreams
+ | Op::GetStreamDetails { .. }
+ | Op::GetTopicDetails { .. }
+ | Op::GetStats => {}
+ }
+
+ state.prune_tombstones();
+ }
+
+ /// On `ExpectedConcurrent` errors, converge shadow toward the server's
+ /// authoritative response. All mutations here are idempotent:
+ ///
+ /// - Delete ops with "not found": remove from shadow if still present
+ /// - Create ops with "already exists": insert into shadow if missing
+ pub fn converge_shadow(&self, err: &IggyError, state: &mut ShadowState) {
+ match (self, err) {
+ (
+ Op::DeleteStream { name },
+ IggyError::StreamNameNotFound(_) | IggyError::StreamIdNotFound(_),
+ ) => {
+ state.apply_delete_stream(name);
+ }
+ (
+ Op::DeleteTopic { stream, topic },
+ IggyError::TopicNameNotFound(_, _) | IggyError::TopicIdNotFound(_, _),
+ ) => {
+ state.apply_delete_topic(stream, topic);
+ }
+ (
+ Op::DeleteConsumerGroup {
+ stream,
+ topic,
+ name,
+ },
+ IggyError::ConsumerGroupNameNotFound(_, _)
+ | IggyError::ConsumerGroupIdNotFound(_, _),
+ ) => {
+ state.apply_delete_consumer_group(stream, topic, name);
+ }
+
+ (Op::CreateStream { name }, IggyError::StreamNameAlreadyExists(_)) => {
+ if !state.stream_exists(name) {
+ state.apply_create_stream(name.clone(), ResourceOrigin::Dynamic);
+ }
+ }
+ (
+ Op::CreateTopic {
+ stream,
+ name,
+ partitions,
+ },
+ IggyError::TopicNameAlreadyExists(_, _),
+ ) => {
+ if !state.topic_exists(stream, name) {
+ state.apply_create_topic(
+ stream,
+ name.clone(),
+ *partitions,
+ ResourceOrigin::Dynamic,
+ );
+ }
+ }
+ (
+ Op::CreateConsumerGroup {
+ stream,
+ topic,
+ name,
+ },
+ IggyError::ConsumerGroupNameAlreadyExists(_, _),
+ ) => {
+ state.apply_create_consumer_group(stream, topic, name.clone());
+ }
+
+ _ => {}
+ }
+
+ state.prune_tombstones();
+ }
+}
diff --git a/core/chaos/src/payload.rs b/core/chaos/src/payload.rs
new file mode 100644
index 000000000..15dd36d32
--- /dev/null
+++ b/core/chaos/src/payload.rs
@@ -0,0 +1,151 @@
+/* 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.
+ */
+
+//! Deterministic checksummed payloads for chaos testing.
+//!
+//! Layout: `[worker_id: u32 LE][seq: u64 LE][crc32: u32 LE][fill...]`
+//!
+//! The CRC covers the entire payload *except* the CRC field itself (bytes 0..12 + 16..).
+//! Fill bytes are a deterministic pattern derived from `seq` to avoid RNG overhead
+//! while still varying content across messages.
+
+const HEADER_SIZE: usize = 16;
+// [0..4] worker_id: u32 LE
+// [4..12] seq: u64 LE
+// [12..16] crc32: u32 LE
+
+/// Generates deterministic, self-verifiable payloads.
+///
+/// Reuses a single buffer across calls - zero heap allocation per message
+/// (caller copies into `Bytes`).
+pub struct PayloadGenerator {
+ worker_id: u32,
+ seq: u64,
+ buf: Vec<u8>,
+}
+
+impl PayloadGenerator {
+ pub fn new(worker_id: u32, msg_size: u32) -> Self {
+ let size = (msg_size as usize).max(HEADER_SIZE);
+ Self {
+ worker_id,
+ seq: 0,
+ buf: vec![0u8; size],
+ }
+ }
+
+ /// Write next payload into the internal buffer and return a reference.
+ pub fn next(&mut self) -> &[u8] {
+ let seq = self.seq;
+ self.seq += 1;
+
+ self.buf[0..4].copy_from_slice(&self.worker_id.to_le_bytes());
+ self.buf[4..12].copy_from_slice(&seq.to_le_bytes());
+
+ // Deterministic fill derived from seq
+ let fill_byte = (seq & 0xFF) as u8;
+ for b in &mut self.buf[HEADER_SIZE..] {
+ *b = fill_byte;
+ }
+
+ // CRC covers everything except the CRC field itself (bytes 12..16)
+ let mut hasher = crc32fast::Hasher::new();
+ hasher.update(&self.buf[..12]);
+ hasher.update(&self.buf[HEADER_SIZE..]);
+ let crc = hasher.finalize();
+ self.buf[12..16].copy_from_slice(&crc.to_le_bytes());
+
+ &self.buf
+ }
+}
+
+/// Verify a payload's integrity. Returns `Ok(())` if valid, or a description of the failure.
+pub fn verify_payload(data: &[u8]) -> Result<(), String> {
+ if data.len() < HEADER_SIZE {
+ return Err(format!(
+ "payload too small: {} bytes (minimum {HEADER_SIZE})",
+ data.len()
+ ));
+ }
+
+ let stored_crc = u32::from_le_bytes(data[12..16].try_into().unwrap());
+
+ let mut hasher = crc32fast::Hasher::new();
+ hasher.update(&data[..12]);
+ hasher.update(&data[HEADER_SIZE..]);
+ let computed_crc = hasher.finalize();
+
+ if stored_crc != computed_crc {
+ let worker_id = u32::from_le_bytes(data[0..4].try_into().unwrap());
+ let seq = u64::from_le_bytes(data[4..12].try_into().unwrap());
+ return Err(format!(
+ "CRC mismatch: stored={stored_crc:#010x}, computed={computed_crc:#010x} \
+ (worker={worker_id}, seq={seq})"
+ ));
+ }
+
+ Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn generated_payload_verifies() {
+ let mut pg = PayloadGenerator::new(42, 64);
+ for _ in 0..100 {
+ let payload = pg.next();
+ verify_payload(payload).expect("payload should verify");
+ }
+ }
+
+ #[test]
+ fn corrupted_payload_fails_verification() {
+ let mut pg = PayloadGenerator::new(1, 32);
+ let payload = pg.next().to_vec();
+ let mut corrupted = payload.clone();
+ corrupted[HEADER_SIZE] ^= 0xFF;
+ assert!(verify_payload(&corrupted).is_err());
+ }
+
+ #[test]
+ fn minimum_size_payload() {
+ let mut pg = PayloadGenerator::new(0, 4);
+ let payload = pg.next();
+ assert_eq!(payload.len(), HEADER_SIZE);
+ verify_payload(payload).expect("minimum-size payload should verify");
+ }
+
+ #[test]
+ fn sequential_payloads_differ() {
+ let mut pg = PayloadGenerator::new(0, 32);
+ let p1 = pg.next().to_vec();
+ let p2 = pg.next().to_vec();
+ assert_ne!(p1, p2);
+ }
+
+ #[test]
+ fn different_workers_differ() {
+ let mut pg1 = PayloadGenerator::new(0, 32);
+ let mut pg2 = PayloadGenerator::new(1, 32);
+ let p1 = pg1.next().to_vec();
+ let p2 = pg2.next().to_vec();
+ assert_ne!(p1, p2);
+ }
+}
diff --git a/core/chaos/src/perfetto.rs b/core/chaos/src/perfetto.rs
new file mode 100644
index 000000000..a06e56fbb
--- /dev/null
+++ b/core/chaos/src/perfetto.rs
@@ -0,0 +1,312 @@
+/* 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.
+ */
+
+//! Converts `trace.jsonl` to Chrome Trace Event format for Perfetto UI visualization.
+//!
+//! Each intent/outcome pair becomes a complete duration event ("X") with
+//! `lane` mapped to process (pid) and `worker` to thread (tid).
+
+use serde::Serialize;
+use std::collections::HashMap;
+use std::fs::File;
+use std::io::{self, BufRead, BufReader, BufWriter};
+use std::path::Path;
+
+pub struct ConvertStats {
+ pub events: usize,
+ pub unmatched_outcomes: usize,
+}
+
+struct PendingIntent {
+ op_name: String,
+ context: Option<String>,
+ worker: u32,
+ lane: String,
+}
+
+#[derive(Serialize)]
+struct ChromeTrace {
+ #[serde(rename = "traceEvents")]
+ trace_events: Vec<ChromeEvent>,
+ #[serde(rename = "displayTimeUnit")]
+ display_time_unit: &'static str,
+}
+
+#[derive(Serialize)]
+struct ChromeEvent {
+ name: String,
+ cat: String,
+ ph: &'static str,
+ ts: u64,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ dur: Option<u64>,
+ pid: u32,
+ tid: u32,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ args: Option<serde_json::Value>,
+}
+
+/// Convert `trace.jsonl` in `bundle_dir` to `trace.perfetto.json`.
+pub fn convert(bundle_dir: &Path) -> io::Result<ConvertStats> {
+ let input_path = bundle_dir.join("trace.jsonl");
+ let output_path = bundle_dir.join("trace.perfetto.json");
+
+ let file = File::open(&input_path)?;
+ let reader = BufReader::new(file);
+
+ let mut pending: HashMap<u64, PendingIntent> = HashMap::new();
+ let mut events: Vec<ChromeEvent> = Vec::new();
+ // lane_name → assigned pid
+ let mut lane_pids: HashMap<String, u32> = HashMap::new();
+ let mut unmatched_outcomes: u64 = 0;
+
+ for line in reader.lines() {
+ let line = line?;
+ if line.is_empty() {
+ continue;
+ }
+
+ let entry: serde_json::Value = match serde_json::from_str(&line) {
+ Ok(v) => v,
+ Err(_) => continue,
+ };
+
+ let Some(seq) = entry["seq"].as_u64() else {
+ continue;
+ };
+ let Some(phase) = entry["phase"].as_str() else {
+ continue;
+ };
+
+ let lane = entry["lane"].as_str().unwrap_or("default");
+ let worker = entry["worker"].as_u64().unwrap_or(0) as u32;
+
+ let next_pid = lane_pids.len() as u32 + 1;
+ lane_pids.entry(lane.to_owned()).or_insert(next_pid);
+
+ match phase {
+ "intent" => {
+ let op_name = entry["op"]["type"]
+ .as_str()
+ .map(camel_to_snake)
+ .unwrap_or_default();
+ let context = op_context(&entry["op"]);
+
+ pending.insert(
+ seq,
+ PendingIntent {
+ op_name,
+ context,
+ worker,
+ lane: lane.to_owned(),
+ },
+ );
+ }
+ "outcome" => {
+ let latency_us = entry["latency_us"].as_u64().unwrap_or(0);
+ let t_us = entry["t_us"].as_u64().unwrap_or(0);
+ let start_us = t_us.saturating_sub(latency_us);
+ let result = entry["result"].as_str().unwrap_or("unknown");
+
+ if let Some(intent) = pending.remove(&seq) {
+ let pid = lane_pids[&intent.lane];
+
+ let mut args = serde_json::Map::new();
+ args.insert("result".into(), serde_json::Value::String(result.into()));
+ args.insert("seq".into(), serde_json::Value::Number(seq.into()));
+ if let Some(ctx) = &intent.context {
+ args.insert("context".into(), serde_json::Value::String(ctx.clone()));
+ }
+ if let Some(err) = entry["error"].as_str() {
+ args.insert("error".into(), serde_json::Value::String(err.into()));
+ }
+
+ events.push(ChromeEvent {
+ name: intent.op_name,
+ cat: "op".into(),
+ ph: "X",
+ ts: start_us,
+ dur: Some(latency_us),
+ pid,
+ tid: intent.worker,
+ args: Some(serde_json::Value::Object(args)),
+ });
+ } else {
+ unmatched_outcomes += 1;
+ }
+ }
+ _ => {}
+ }
+ }
+
+ // Sort by start timestamp for consistent output
+ events.sort_by_key(|e| e.ts);
+
+ // Emit metadata events for lane and worker names
+ let mut meta_events: Vec<ChromeEvent> = Vec::new();
+
+ for (lane_name, pid) in &lane_pids {
+ meta_events.push(ChromeEvent {
+ name: "process_name".into(),
+ cat: String::new(),
+ ph: "M",
+ ts: 0,
+ dur: None,
+ pid: *pid,
+ tid: 0,
+ args: Some(serde_json::json!({ "name": lane_name })),
+ });
+ }
+
+ // Collect unique (pid, tid) pairs for thread name metadata
+ let mut worker_threads: HashMap<(u32, u32), ()> = HashMap::new();
+ for event in &events {
+ worker_threads.entry((event.pid, event.tid)).or_default();
+ }
+ for (pid, tid) in worker_threads.keys() {
+ meta_events.push(ChromeEvent {
+ name: "thread_name".into(),
+ cat: String::new(),
+ ph: "M",
+ ts: 0,
+ dur: None,
+ pid: *pid,
+ tid: *tid,
+ args: Some(serde_json::json!({ "name": format!("worker-{tid}") })),
+ });
+ }
+
+ meta_events.append(&mut events);
+
+ let trace = ChromeTrace {
+ trace_events: meta_events,
+ display_time_unit: "us",
+ };
+
+ let out_file = File::create(&output_path)?;
+ let writer = BufWriter::new(out_file);
+ serde_json::to_writer(writer, &trace).map_err(io::Error::other)?;
+
+ Ok(ConvertStats {
+ events: trace.trace_events.len(),
+ unmatched_outcomes: unmatched_outcomes as usize,
+ })
+}
+
+/// Extract a compact context label from the op's JSON fields (excluding "type").
+fn op_context(op: &serde_json::Value) -> Option<String> {
+ let obj = op.as_object()?;
+ let mut parts = Vec::new();
+ for (key, val) in obj {
+ if key == "type" {
+ continue;
+ }
+ match val {
+ serde_json::Value::String(s) => parts.push(format!("{key}={s}")),
+ serde_json::Value::Number(n) => parts.push(format!("{key}={n}")),
+ _ => {}
+ }
+ }
+ if parts.is_empty() {
+ None
+ } else {
+ Some(parts.join(" "))
+ }
+}
+
+/// Convert CamelCase to snake_case.
+fn camel_to_snake(s: &str) -> String {
+ let mut result = String::with_capacity(s.len() + 4);
+ for (i, ch) in s.chars().enumerate() {
+ if ch.is_uppercase() {
+ if i > 0 {
+ result.push('_');
+ }
+ result.push(ch.to_ascii_lowercase());
+ } else {
+ result.push(ch);
+ }
+ }
+ result
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_camel_to_snake() {
+ assert_eq!(camel_to_snake("CreateStream"), "create_stream");
+ assert_eq!(camel_to_snake("SendMessages"), "send_messages");
+ assert_eq!(camel_to_snake("GetStats"), "get_stats");
+ assert_eq!(camel_to_snake("PollGroupMessages"), "poll_group_messages");
+ }
+
+ #[test]
+ fn test_convert_roundtrip() {
+ let dir = std::env::temp_dir().join("perfetto_test");
+ let _ = std::fs::create_dir_all(&dir);
+
+ let trace = r#"{"seq":1,"t_us":1000,"worker":0,"lane":"crud","phase":"intent","op":{"type":"CreateStream","name":"s1"}}
+{"seq":1,"t_us":1200,"worker":0,"lane":"crud","phase":"outcome","result":"ok","latency_us":200}
+{"seq":2,"t_us":1500,"worker":1,"lane":"data","phase":"intent","op":{"type":"SendMessages","stream":"s1","topic":"t1","partition":1,"count":10}}
+{"seq":2,"t_us":1800,"worker":1,"lane":"data","phase":"outcome","result":"ok","latency_us":300}
+"#;
+ std::fs::write(dir.join("trace.jsonl"), trace).unwrap();
+
+ let stats = convert(&dir).unwrap();
+ assert_eq!(stats.unmatched_outcomes, 0);
+ // 2 duration events + metadata events (2 lanes + 2 workers)
+ assert!(stats.events >= 2);
+
+ let output: serde_json::Value =
+ serde_json::from_reader(File::open(dir.join("trace.perfetto.json")).unwrap()).unwrap();
+ assert_eq!(output["displayTimeUnit"], "us");
+
+ let events = output["traceEvents"].as_array().unwrap();
+ let duration_events: Vec<_> = events.iter().filter(|e| e["ph"] == "X").collect();
+ assert_eq!(duration_events.len(), 2);
+
+ // First duration event: create_stream starting at 1000, dur 200
+ assert_eq!(duration_events[0]["name"], "create_stream");
+ assert_eq!(duration_events[0]["ts"], 1000);
+ assert_eq!(duration_events[0]["dur"], 200);
+
+ // Second: send_messages starting at 1500, dur 300
+ assert_eq!(duration_events[1]["name"], "send_messages");
+ assert_eq!(duration_events[1]["ts"], 1500);
+ assert_eq!(duration_events[1]["dur"], 300);
+
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+
+ #[test]
+ fn test_unmatched_outcome() {
+ let dir = std::env::temp_dir().join("perfetto_test_unmatched");
+ let _ = std::fs::create_dir_all(&dir);
+
+ let trace = r#"{"seq":99,"t_us":5000,"worker":0,"lane":"crud","phase":"outcome","result":"ok","latency_us":100}
+"#;
+ std::fs::write(dir.join("trace.jsonl"), trace).unwrap();
+
+ let stats = convert(&dir).unwrap();
+ assert_eq!(stats.unmatched_outcomes, 1);
+
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+}
diff --git a/core/chaos/src/replay.rs b/core/chaos/src/replay.rs
new file mode 100644
index 000000000..3e640fd5d
--- /dev/null
+++ b/core/chaos/src/replay.rs
@@ -0,0 +1,346 @@
+/* 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 crate::args::ReplayArgs;
+use crate::client::{cleanup_all, create_client};
+use crate::error::LabError;
+use crate::ops::{ErrorClass, Op, OpDetail, OpOutcome};
+use crate::payload::PayloadGenerator;
+use crate::scenarios::NamespacePrefixes;
+use serde::Deserialize;
+use std::collections::HashMap;
+use std::fmt;
+use std::fs::File;
+use std::io::{BufRead, BufReader};
+use tracing::{error, info, warn};
+
+pub struct LabReplay {
+ args: ReplayArgs,
+}
+
+pub struct ReplayOutcome {
+ pub total_ops: u64,
+ pub divergences_total: u64,
+ pub divergences_concerning: u64,
+}
+
+/// Raw JSON line from a trace file - intent and outcome share the same shape
+/// with optional fields.
+#[derive(Deserialize)]
+struct RawTraceEntry {
+ seq: u64,
+ phase: String,
+ // Intent fields
+ op: Option<Op>,
+ // Outcome fields
+ result: Option<String>,
+ detail: Option<OpDetail>,
+}
+
+struct ReplayIntent {
+ seq: u64,
+ op: Op,
+}
+
+struct OriginalOutcome {
+ result: String,
+ detail: Option<OpDetail>,
+}
+
+struct Divergence {
+ seq: u64,
+ kind: DivergenceKind,
+ op: Op,
+ original_result: String,
+ replay_result: String,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum DivergenceKind {
+ /// Original succeeded, replay failed - concerning
+ OriginalOkReplayFailed,
+ /// Original succeeded, replay returned expected error - concerning
+ OriginalOkReplayExpected,
+ /// Original got expected error, replay succeeded - benign (no concurrent race in sequential replay)
+ OriginalExpectedReplayOk,
+ /// Original got unexpected error, replay succeeded - interesting
+ OriginalUnexpectedReplayOk,
+ /// Both failed with different classifications
+ BothFailed,
+}
+
+impl DivergenceKind {
+ fn is_concerning(self) -> bool {
+ matches!(
+ self,
+ Self::OriginalOkReplayFailed | Self::OriginalOkReplayExpected
+ )
+ }
+}
+
+impl fmt::Display for DivergenceKind {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::OriginalOkReplayFailed => write!(f, "original_ok→replay_failed"),
+ Self::OriginalOkReplayExpected => write!(f, "original_ok→replay_expected_error"),
+ Self::OriginalExpectedReplayOk => write!(f, "original_expected→replay_ok"),
+ Self::OriginalUnexpectedReplayOk => write!(f, "original_unexpected→replay_ok"),
+ Self::BothFailed => write!(f, "both_failed"),
+ }
+ }
+}
+
+impl LabReplay {
+ pub fn new(args: ReplayArgs) -> Self {
+ Self { args }
+ }
+
+ pub async fn run(self) -> Result<ReplayOutcome, LabError> {
+ let (intents, outcomes) = self.load_traces()?;
+ if intents.is_empty() {
+ return Err(LabError::TraceParse {
+ path: self.args.trace_dir.display().to_string(),
+ line: 0,
+ message: "no intent entries found in trace directory".into(),
+ });
+ }
+
+ info!("=== iggy-chaos replay ===");
+ info!("Trace dir: {:?}", self.args.trace_dir);
+ info!("Intents loaded: {}", intents.len());
+ info!(
+ "Server: {} ({})",
+ self.args.server_address, self.args.transport
+ );
+
+ let client = create_client(&self.args.server_address, self.args.transport).await?;
+ let prefixes = NamespacePrefixes::from_base(&self.args.prefix);
+
+ if self.args.force_cleanup {
+ info!("Cleaning up prefixed resources...");
+ cleanup_all(&client, &prefixes).await;
+ }
+
+ let mut payload_gen = PayloadGenerator::new(0, self.args.message_size);
+ let mut divergences: Vec<Divergence> = Vec::new();
+ let total_ops = intents.len() as u64;
+
+ for intent in &intents {
+ let replay_outcome = match intent
+ .op
+ .execute(&client, &mut payload_gen, &prefixes)
+ .await
+ {
+ Ok(detail) => OpOutcome::success_from(detail),
+ Err(ref e) => {
+ let class = match e {
+ LabError::Iggy(ie) => ErrorClass::coarse(ie),
+ LabError::Timeout { .. } => ErrorClass::Transient,
+ _ => ErrorClass::ServerBug,
+ };
+ OpOutcome::error(class, e.to_string(), intent.op.context())
+ }
+ };
+ let replay_tag = replay_outcome.result_tag();
+
+ if let Some(original) = outcomes.get(&intent.seq) {
+ if replay_tag == original.result {
+ continue;
+ }
+
+ let kind = classify_divergence(&original.result, replay_tag);
+ let divergence = Divergence {
+ seq: intent.seq,
+ kind,
+ op: intent.op.clone(),
+ original_result: original.result.clone(),
+ replay_result: replay_tag.to_owned(),
+ };
+
+ if kind.is_concerning() {
+ warn!(
+ seq = intent.seq,
+ kind = %kind,
+ op = intent.op.kind_tag(),
+ original = %original.result,
+ replay = replay_tag,
+ detail = ?original.detail,
+ "Concerning divergence"
+ );
+ } else {
+ info!(
+ seq = intent.seq,
+ kind = %kind,
+ op = intent.op.kind_tag(),
+ "Benign divergence"
+ );
+ }
+
+ let is_concerning = kind.is_concerning();
+ divergences.push(divergence);
+
+ if self.args.fail_fast && is_concerning {
+ error!(
+ "Fail-fast: stopping on concerning divergence at seq={}",
+ intent.seq
+ );
+ break;
+ }
+ }
+ }
+
+ let divergences_concerning = divergences
+ .iter()
+ .filter(|d| d.kind.is_concerning())
+ .count() as u64;
+ let divergences_total = divergences.len() as u64;
+
+ info!("--- Replay Summary ---");
+ info!("Total ops replayed: {total_ops}");
+ info!("Total divergences: {divergences_total}");
+ info!("Concerning divergences: {divergences_concerning}");
+
+ if divergences_concerning > 0 {
+ error!("REPLAY FAILED - {divergences_concerning} concerning divergence(s):");
+ for d in divergences.iter().filter(|d| d.kind.is_concerning()) {
+ error!(
+ " seq={} op={} {} (original={}, replay={})",
+ d.seq,
+ d.op.kind_tag(),
+ d.kind,
+ d.original_result,
+ d.replay_result,
+ );
+ }
+ } else if divergences_total > 0 {
+ info!("REPLAY OK - {divergences_total} benign divergence(s)");
+ } else {
+ info!("REPLAY OK - exact match");
+ }
+
+ // Cleanup after replay
+ if self.args.force_cleanup {
+ info!("Post-replay cleanup...");
+ cleanup_all(&client, &prefixes).await;
+ }
+
+ Ok(ReplayOutcome {
+ total_ops,
+ divergences_total,
+ divergences_concerning,
+ })
+ }
+
+ fn load_traces(&self) -> Result<(Vec<ReplayIntent>, HashMap<u64, OriginalOutcome>), LabError> {
+ let trace_dir = &self.args.trace_dir;
+
+ // Try single trace.jsonl first (new format), fall back to per-worker files
+ let trace_files = {
+ let single = trace_dir.join("trace.jsonl");
+ if single.exists() {
+ vec![single]
+ } else {
+ let mut files = Vec::new();
+ let entries = std::fs::read_dir(trace_dir).map_err(|e| LabError::Io {
+ context: format!("reading trace dir {}", trace_dir.display()),
+ source: e,
+ })?;
+ for entry in entries {
+ let entry = entry.map_err(|e| LabError::Io {
+ context: format!("iterating trace dir {}", trace_dir.display()),
+ source: e,
+ })?;
+ let path = entry.path();
+ let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
+ if name.starts_with("trace-worker-") && name.ends_with(".jsonl") {
+ files.push(path);
+ }
+ }
+ files
+ }
+ };
+
+ let mut intents: Vec<ReplayIntent> = Vec::new();
+ let mut outcomes: HashMap<u64, OriginalOutcome> = HashMap::new();
+
+ for path in &trace_files {
+ let file = File::open(path).map_err(|e| LabError::Io {
+ context: format!("opening trace file {}", path.display()),
+ source: e,
+ })?;
+ let reader = BufReader::new(file);
+
+ for (line_num, line) in reader.lines().enumerate() {
+ let line = line.map_err(|e| LabError::Io {
+ context: format!("reading line {} of {}", line_num + 1, path.display()),
+ source: e,
+ })?;
+ if line.is_empty() {
+ continue;
+ }
+ let raw: RawTraceEntry =
+ serde_json::from_str(&line).map_err(|e| LabError::TraceParse {
+ path: path.display().to_string(),
+ line: line_num + 1,
+ message: e.to_string(),
+ })?;
+
+ match raw.phase.as_str() {
+ "intent" => {
+ if let Some(op) = raw.op {
+ intents.push(ReplayIntent { seq: raw.seq, op });
+ }
+ }
+ "outcome" => {
+ if let Some(result) = raw.result {
+ outcomes.insert(
+ raw.seq,
+ OriginalOutcome {
+ result,
+ detail: raw.detail,
+ },
+ );
+ }
+ }
+ _ => {}
+ }
+ }
+ }
+
+ intents.sort_by_key(|i| i.seq);
+ Ok((intents, outcomes))
+ }
+}
+
+fn is_error_tag(tag: &str) -> bool {
+ matches!(tag, "expected_error" | "server_bug" | "transient")
+}
+
+fn classify_divergence(original_result: &str, replay_result: &str) -> DivergenceKind {
+ match (original_result, replay_result) {
+ ("ok", t) if is_error_tag(t) && t != "expected_error" => {
+ DivergenceKind::OriginalOkReplayFailed
+ }
+ ("ok", "expected_error") => DivergenceKind::OriginalOkReplayExpected,
+ ("expected_error", "ok") => DivergenceKind::OriginalExpectedReplayOk,
+ (t, "ok") if is_error_tag(t) && t != "expected_error" => {
+ DivergenceKind::OriginalUnexpectedReplayOk
+ }
+ _ => DivergenceKind::BothFailed,
+ }
+}
diff --git a/core/chaos/src/report.rs b/core/chaos/src/report.rs
new file mode 100644
index 000000000..6c80ff976
--- /dev/null
+++ b/core/chaos/src/report.rs
@@ -0,0 +1,139 @@
+/* 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 crate::invariants::InvariantViolation;
+use serde::Serialize;
+use std::fs;
+use std::io::{self, BufWriter, Write};
+use std::path::Path;
+use std::time::Duration;
+
+pub struct ArtifactBundle<'a> {
+ pub run_id: &'a str,
+ pub seed: u64,
+ pub scenario_name: &'a str,
+ pub server_address: &'a str,
+ pub transport: String,
+ pub workers: u32,
+ pub duration: Duration,
+ pub total_ops: u64,
+ pub passed: bool,
+ pub violations: &'a [InvariantViolation],
+ pub ops_target: Option<u64>,
+ pub start_time_utc: &'a str,
+ pub message_size: u32,
+ pub messages_per_batch: u32,
+ pub prefix: &'a str,
+ pub no_fail_fast: bool,
+ pub force_cleanup: bool,
+ pub no_cleanup: bool,
+ pub skip_post_run_verify: bool,
+ pub trace_drops: u64,
+}
+
+#[derive(Serialize)]
+struct ConfigJson<'a> {
+ run_id: &'a str,
+ seed: u64,
+ scenario: &'a str,
+ server_address: &'a str,
+ transport: &'a str,
+ workers: u32,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ ops_target: Option<u64>,
+ start_time_utc: &'a str,
+ message_size: u32,
+ messages_per_batch: u32,
+ prefix: &'a str,
+ no_fail_fast: bool,
+ force_cleanup: bool,
+ no_cleanup: bool,
+ skip_post_run_verify: bool,
+}
+
+#[derive(Serialize)]
+struct SummaryJson<'a> {
+ run_id: &'a str,
+ passed: bool,
+ total_ops: u64,
+ duration_secs: f64,
+ duration_human: &'a str,
+ violation_count: usize,
+ scenario: &'a str,
+ trace_drops: u64,
+}
+
+impl<'a> ArtifactBundle<'a> {
+ pub fn write(&self, output_dir: &Path) -> io::Result<()> {
+ fs::create_dir_all(output_dir)?;
+
+ let config = ConfigJson {
+ run_id: self.run_id,
+ seed: self.seed,
+ scenario: self.scenario_name,
+ server_address: self.server_address,
+ transport: &self.transport,
+ workers: self.workers,
+ ops_target: self.ops_target,
+ start_time_utc: self.start_time_utc,
+ message_size: self.message_size,
+ messages_per_batch: self.messages_per_batch,
+ prefix: self.prefix,
+ no_fail_fast: self.no_fail_fast,
+ force_cleanup: self.force_cleanup,
+ no_cleanup: self.no_cleanup,
+ skip_post_run_verify: self.skip_post_run_verify,
+ };
+ let config_path = output_dir.join("config.json");
+ let file = fs::File::create(config_path)?;
+ let mut config_writer = BufWriter::new(file);
+ serde_json::to_writer_pretty(&mut config_writer, &config).map_err(io::Error::other)?;
+ config_writer.flush()?;
+
+ let duration_human = format!("{:.1?}", self.duration);
+ let summary = SummaryJson {
+ run_id: self.run_id,
+ passed: self.passed,
+ total_ops: self.total_ops,
+ duration_secs: self.duration.as_secs_f64(),
+ duration_human: &duration_human,
+ violation_count: self.violations.len(),
+ scenario: self.scenario_name,
+ trace_drops: self.trace_drops,
+ };
+ let summary_path = output_dir.join("summary.json");
+ let file = fs::File::create(summary_path)?;
+ let mut summary_writer = BufWriter::new(file);
+ serde_json::to_writer_pretty(&mut summary_writer, &summary).map_err(io::Error::other)?;
+ summary_writer.flush()?;
+
+ // violations.jsonl
+ if !self.violations.is_empty() {
+ let violations_path = output_dir.join("violations.jsonl");
+ let file = fs::File::create(violations_path)?;
+ let mut writer = BufWriter::new(file);
+ for v in self.violations {
+ serde_json::to_writer(&mut writer, v).map_err(io::Error::other)?;
+ writer.write_all(b"\n")?;
+ }
+ writer.flush()?;
+ }
+
+ Ok(())
+ }
+}
diff --git a/core/chaos/src/runner.rs b/core/chaos/src/runner.rs
new file mode 100644
index 000000000..1494f14fb
--- /dev/null
+++ b/core/chaos/src/runner.rs
@@ -0,0 +1,420 @@
+/* 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 crate::args::RunArgs;
+use crate::client::create_client;
+use crate::error::LabError;
+use crate::in_flight::InFlightOps;
+use crate::invariants::{self, InvariantViolation};
+use crate::report::ArtifactBundle;
+use crate::scenarios::{self, CleanupPolicy, NamespacePrefixes, VerifyContext};
+use crate::shadow::{ResourceOrigin, ShadowState};
+use crate::trace;
+use crate::watermark::WatermarkState;
+use crate::worker::{WorkerConfig, WorkerInit, WorkerResult, run_worker};
+use iggy::prelude::*;
+use rand::SeedableRng;
+use rand::rngs::StdRng;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+use tokio::task::JoinSet;
+use tracing::{error, info, warn};
+
+pub struct LabRunner {
+ args: RunArgs,
+}
+
+#[allow(dead_code)]
+pub struct RunOutcome {
+ pub passed: bool,
+ pub run_id: String,
+ pub ops_total: u64,
+ pub violations: Vec<InvariantViolation>,
+ pub duration: Duration,
+}
+
+impl LabRunner {
+ pub fn new(args: RunArgs) -> Self {
+ Self { args }
+ }
+
+ pub async fn run(self) -> Result<RunOutcome, LabError> {
+ let seed = self.args.seed.unwrap_or_else(rand::random);
+ let scenario = scenarios::create_scenario(self.args.scenario);
+ let prefixes = NamespacePrefixes::from_base(&self.args.prefix);
+
+ let start_wall = chrono::Utc::now();
+ let start_instant = Instant::now();
+ let run_id = format!(
+ "{}_{seed}_{}",
+ self.args.scenario,
+ start_wall.format("%Y%m%dT%H%M%S")
+ );
+ let start_time_utc = start_wall.to_rfc3339();
+
+ let explicit_output = self.args.output_dir.is_some();
+ let bundle_dir = match self.args.output_dir {
+ Some(ref dir) => dir.clone(),
+ None => PathBuf::from("chaos_logs").join(&run_id),
+ };
+ std::fs::create_dir_all(&bundle_dir).map_err(|e| LabError::Io {
+ context: format!("creating bundle dir {}", bundle_dir.display()),
+ source: e,
+ })?;
+
+ info!("=== iggy-chaos ===");
+ info!("Run ID: {run_id}");
+ info!("Scenario: {}", scenario.name());
+ info!("Seed: {seed}");
+ info!(
+ "Server: {} ({})",
+ self.args.server_address, self.args.transport
+ );
+ info!("Workers: {}", self.args.workers);
+ info!("Duration: {}", self.args.duration);
+ if let Some(ops) = self.args.ops {
+ info!("Ops target: {ops}");
+ }
+ info!("Prefix: {:?}", self.args.prefix);
+ info!("Bundle: {}", bundle_dir.display());
+
+ // Pre-flight check
+ let admin_client = create_client(&self.args.server_address, self.args.transport).await?;
+ self.preflight_check(&admin_client, &prefixes, self.args.force_cleanup)
+ .await?;
+
+ // Setup phase - delegated to the scenario
+ scenario.setup(&admin_client, &prefixes).await?;
+
+ let shared_state = Arc::new(tokio::sync::RwLock::new(ShadowState::new(prefixes.clone())));
+ let watermarks = Arc::new(tokio::sync::Mutex::new(WatermarkState::new()));
+ let in_flight = Arc::new(InFlightOps::new());
+ let stop = Arc::new(AtomicBool::new(false));
+ let seq_counter = Arc::new(AtomicU64::new(0));
+ let ops_counter = Arc::new(AtomicU64::new(0));
+
+ // Populate initial shadow state from setup
+ {
+ let mut state = shared_state.write().await;
+ let streams = admin_client.get_streams().await?;
+ for stream in streams.iter().filter(|s| prefixes.matches(&s.name)) {
+ state.apply_create_stream(stream.name.clone(), ResourceOrigin::Setup);
+ let stream_id = Identifier::numeric(stream.id).unwrap();
+ if let Ok(Some(details)) = admin_client.get_stream(&stream_id).await {
+ for topic in &details.topics {
+ state.apply_create_topic(
+ &stream.name,
+ topic.name.clone(),
+ topic.partitions_count,
+ ResourceOrigin::Setup,
+ );
+ }
+ }
+ }
+ }
+
+ // Spawn workers per lane
+ let lanes = scenario.lanes();
+ let mut join_set = JoinSet::new();
+ let mut worker_id: u32 = 0;
+
+ let (trace_writer, sender_factory) = trace::create_trace_channel(start_instant);
+ let trace_path = bundle_dir.join("trace.jsonl");
+ let trace_handle = tokio::task::spawn_blocking(move || trace_writer.run(&trace_path));
+
+ for lane in &lanes {
+ let lane_workers = lane.worker_count.unwrap_or(self.args.workers);
+ info!("Lane '{}': {} workers", lane.label, lane_workers);
+ for _ in 0..lane_workers {
+ let worker_rng = StdRng::seed_from_u64(derive_worker_seed(seed, worker_id));
+ let client = create_client(&self.args.server_address, self.args.transport).await?;
+ let config = WorkerConfig {
+ msg_size: self.args.message_size,
+ messages_per_batch: self.args.messages_per_batch,
+ prefixes: prefixes.clone(),
+ namespace: lane.namespace,
+ fail_fast: !self.args.no_fail_fast,
+ ops_target: self.args.ops,
+ scope: lane.scope,
+ rate: lane.rate,
+ allow_destructive: lane.allow_destructive,
+ consumer_id: worker_id + 1,
+ lane: lane.label,
+ };
+ let init = WorkerInit {
+ id: worker_id,
+ client,
+ rng: worker_rng,
+ shared_state: Arc::clone(&shared_state),
+ watermarks: Arc::clone(&watermarks),
+ in_flight: Arc::clone(&in_flight),
+ trace_sender: Some(sender_factory.sender(worker_id, lane.label)),
+ seq_counter: Arc::clone(&seq_counter),
+ ops_counter: Arc::clone(&ops_counter),
+ start: start_instant,
+ };
+ let weights = lane.op_weights.clone();
+ let stop_clone = Arc::clone(&stop);
+ join_set.spawn(async move { run_worker(init, &weights, config, stop_clone).await });
+ worker_id += 1;
+ }
+ }
+
+ let run_start = Instant::now();
+
+ if worker_id == 0 {
+ info!("No workers to run, skipping chaos phase.");
+ } else {
+ let duration: Duration = self.args.duration.get_duration();
+ let ops_target = self.args.ops;
+ tokio::select! {
+ _ = tokio::time::sleep(duration) => {
+ info!("Duration elapsed, stopping workers...");
+ }
+ _ = shutdown_signal() => {
+ info!("Received shutdown signal, stopping workers...");
+ }
+ _ = wait_for_ops_target(&ops_counter, ops_target) => {
+ info!("Op count target reached, stopping workers...");
+ }
+ }
+ stop.store(true, Ordering::Relaxed);
+ }
+
+ // Collect results
+ let mut total_ops = 0u64;
+ let mut violations = Vec::new();
+ while let Some(result) = join_set.join_next().await {
+ match result {
+ Ok(WorkerResult::Ok { ops_executed }) => {
+ total_ops += ops_executed;
+ }
+ Ok(WorkerResult::ServerBug(v)) => {
+ violations.push(v);
+ }
+ Err(e) => {
+ error!("Worker panicked: {e}");
+ }
+ }
+ }
+
+ // Close the trace channel and wait for the writer to flush
+ let trace_drops = sender_factory.dropped();
+ drop(sender_factory);
+ if let Err(e) = trace_handle.await {
+ error!("Trace writer task panicked: {e}");
+ }
+
+ let run_duration = run_start.elapsed();
+ info!("All workers stopped. Total ops: {total_ops}, duration: {run_duration:.1?}");
+
+ // Post-run verification - delegated to the scenario
+ if !self.args.skip_post_run_verify {
+ info!("Running post-run verification...");
+ let state = shared_state.read().await;
+ let wm = watermarks.lock().await;
+ let ctx = VerifyContext {
+ client: &admin_client,
+ shadow: &state,
+ prefixes: &prefixes,
+ server_address: &self.args.server_address,
+ transport: self.args.transport,
+ };
+ let post_violations = scenario.verify(&ctx).await;
+ if !post_violations.is_empty() {
+ warn!("Post-run violations: {}", post_violations.len());
+ }
+ violations.extend(post_violations);
+
+ // Watermark-based verification (purge + phantom read detection)
+ let wm_violations = invariants::watermark_verify(&admin_client, &wm, &state).await;
+ if !wm_violations.is_empty() {
+ warn!("Watermark violations: {}", wm_violations.len());
+ }
+ violations.extend(wm_violations);
+ }
+
+ let passed = violations.is_empty();
+
+ // Write artifacts
+ {
+ let bundle = ArtifactBundle {
+ run_id: &run_id,
+ seed,
+ scenario_name: scenario.name(),
+ server_address: &self.args.server_address,
+ transport: self.args.transport.to_string(),
+ workers: self.args.workers,
+ duration: run_duration,
+ total_ops,
+ passed,
+ violations: &violations,
+ ops_target: self.args.ops,
+ start_time_utc: &start_time_utc,
+ message_size: self.args.message_size,
+ messages_per_batch: self.args.messages_per_batch,
+ prefix: &self.args.prefix,
+ no_fail_fast: self.args.no_fail_fast,
+ force_cleanup: self.args.force_cleanup,
+ no_cleanup: self.args.no_cleanup,
+ skip_post_run_verify: self.args.skip_post_run_verify,
+ trace_drops,
+ };
+ if let Err(e) = bundle.write(&bundle_dir) {
+ error!("Failed to write artifacts: {e}");
+ }
+ }
+
+ // Generate Perfetto trace from the JSONL data
+ match crate::perfetto::convert(&bundle_dir) {
+ Ok(stats) => {
+ info!(
+ "Perfetto trace: {} events → {}",
+ stats.events,
+ bundle_dir.join("trace.perfetto.json").display(),
+ );
+ if stats.unmatched_outcomes > 0 {
+ warn!(
+ "Perfetto: {} unmatched outcome entries (missing intents)",
+ stats.unmatched_outcomes,
+ );
+ }
+ }
+ Err(e) => {
+ warn!("Failed to generate Perfetto trace: {e}");
+ }
+ }
+
+ // Cleanup - policy-driven by the scenario
+ if !self.args.no_cleanup {
+ match scenario.cleanup_policy() {
+ CleanupPolicy::DeletePrefix => {
+ info!("Cleaning up lab resources...");
+ crate::client::cleanup_all(&admin_client, &prefixes).await;
+ }
+ CleanupPolicy::NoOp => {
+ info!("Cleanup skipped (scenario policy: NoOp).");
+ }
+ }
+ }
+
+ if passed {
+ info!("PASSED - {total_ops} ops, 0 violations");
+ if !explicit_output {
+ let _ = std::fs::remove_dir_all(&bundle_dir);
+ }
+ } else {
+ error!("FAILED - {total_ops} ops, {} violations", violations.len());
+ for v in &violations {
+ error!(" [{}] {}: {}", v.kind, v.description, v.context);
+ }
+ if !explicit_output {
+ error!("Bundle preserved at: {}", bundle_dir.display());
+ }
+ }
+
+ Ok(RunOutcome {
+ passed,
+ run_id,
+ ops_total: total_ops,
+ violations,
+ duration: run_duration,
+ })
+ }
+
+ async fn preflight_check(
+ &self,
+ client: &IggyClient,
+ prefixes: &NamespacePrefixes,
+ force: bool,
+ ) -> Result<(), LabError> {
+ let streams = client.get_streams().await?;
+ let stale: Vec<_> = streams
+ .iter()
+ .filter(|s| prefixes.matches(&s.name))
+ .collect();
+
+ if stale.is_empty() {
+ return Ok(());
+ }
+
+ if force {
+ info!(
+ "Force-cleaning {} stale resources with base prefix '{}'",
+ stale.len(),
+ prefixes.base,
+ );
+ for stream in &stale {
+ let id = Identifier::numeric(stream.id).unwrap();
+ if let Err(e) = client.delete_stream(&id).await {
+ warn!("Failed to delete stale stream '{}': {e}", stream.name);
+ }
+ }
+ Ok(())
+ } else {
+ error!(
+ "Found {} stale resources with base prefix '{}'. Use --force-cleanup to remove them.",
+ stale.len(),
+ prefixes.base,
+ );
+ Err(LabError::PrefixViolation(format!(
+ "found {} stale resources with base prefix '{}'",
+ stale.len(),
+ prefixes.base,
+ )))
+ }
+ }
+}
+
+async fn wait_for_ops_target(counter: &AtomicU64, target: Option<u64>) {
+ let Some(target) = target else {
+ std::future::pending::<()>().await;
+ return;
+ };
+ let mut interval = tokio::time::interval(Duration::from_millis(50));
+ loop {
+ interval.tick().await;
+ if counter.load(Ordering::Relaxed) >= target {
+ return;
+ }
+ }
+}
+
+/// Derive a per-worker seed that avoids the collision problem of `seed + worker_id`
+/// (where seed=0 worker=1 collides with seed=1 worker=0).
+///
+/// Uses ahash with fixed seeds for cross-version determinism (DefaultHasher
+/// delegates to SipHash whose output is not guaranteed stable across rustc releases).
+fn derive_worker_seed(base_seed: u64, worker_id: u32) -> u64 {
+ const FIXED: ahash::RandomState = ahash::RandomState::with_seeds(0, 0, 0, 0);
+ FIXED.hash_one((base_seed, worker_id))
+}
+
+/// Resolves when either SIGINT (Ctrl+C) or SIGTERM is received.
+async fn shutdown_signal() {
+ use tokio::signal::unix::{SignalKind, signal};
+
+ let mut sigterm = signal(SignalKind::terminate()).expect("Failed to register SIGTERM handler");
+
+ tokio::select! {
+ _ = tokio::signal::ctrl_c() => {}
+ _ = sigterm.recv() => {}
+ }
+}
diff --git a/core/chaos/src/safe_name.rs b/core/chaos/src/safe_name.rs
new file mode 100644
index 000000000..75cef378d
--- /dev/null
+++ b/core/chaos/src/safe_name.rs
@@ -0,0 +1,114 @@
+/* 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 rand::distr::Alphanumeric;
+use rand::{Rng, RngExt};
+use serde::{Deserialize, Serialize};
+use std::fmt;
+use std::ops::Deref;
+
+const RANDOM_SUFFIX_LEN: usize = 8;
+
+/// Resource name that is guaranteed to start with a configurable prefix.
+///
+/// This prevents chaos tests from accidentally operating on non-lab resources.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct SafeResourceName(String);
+
+impl SafeResourceName {
+ pub fn new(prefix: &str, name: &str) -> Self {
+ let full = if name.starts_with(prefix) {
+ name.to_owned()
+ } else {
+ format!("{prefix}{name}")
+ };
+ Self(full)
+ }
+
+ pub fn random(prefix: &str, rng: &mut impl Rng) -> Self {
+ let suffix: String = (0..RANDOM_SUFFIX_LEN)
+ .map(|_| rng.sample(Alphanumeric) as char)
+ .collect();
+ Self(format!("{prefix}{suffix}"))
+ }
+
+ pub fn has_prefix(&self, prefix: &str) -> bool {
+ self.0.starts_with(prefix)
+ }
+
+ pub fn into_inner(self) -> String {
+ self.0
+ }
+}
+
+impl Deref for SafeResourceName {
+ type Target = str;
+
+ fn deref(&self) -> &str {
+ &self.0
+ }
+}
+
+impl fmt::Display for SafeResourceName {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.write_str(&self.0)
+ }
+}
+
+impl From<SafeResourceName> for String {
+ fn from(name: SafeResourceName) -> Self {
+ name.0
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use rand::SeedableRng;
+ use rand::rngs::StdRng;
+
+ #[test]
+ fn new_prepends_prefix_when_missing() {
+ let name = SafeResourceName::new("lab-", "stream1");
+ assert_eq!(&*name, "lab-stream1");
+ }
+
+ #[test]
+ fn new_preserves_existing_prefix() {
+ let name = SafeResourceName::new("lab-", "lab-stream1");
+ assert_eq!(&*name, "lab-stream1");
+ }
+
+ #[test]
+ fn random_produces_prefixed_name() {
+ let mut rng = StdRng::seed_from_u64(42);
+ let name = SafeResourceName::random("lab-", &mut rng);
+ assert!(name.starts_with("lab-"));
+ assert_eq!(name.len(), 4 + RANDOM_SUFFIX_LEN);
+ }
+
+ #[test]
+ fn random_is_deterministic_with_same_seed() {
+ let mut rng1 = StdRng::seed_from_u64(99);
+ let mut rng2 = StdRng::seed_from_u64(99);
+ let a = SafeResourceName::random("lab-", &mut rng1);
+ let b = SafeResourceName::random("lab-", &mut rng2);
+ assert_eq!(a, b);
+ }
+}
diff --git a/core/chaos/src/scenarios/concurrent_crud.rs b/core/chaos/src/scenarios/concurrent_crud.rs
new file mode 100644
index 000000000..5549d9804
--- /dev/null
+++ b/core/chaos/src/scenarios/concurrent_crud.rs
@@ -0,0 +1,93 @@
+/* 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 super::{
+ LaneSpec, NamespacePrefixes, OriginScope, Scenario, Setup, TargetNamespace, TopicDefaults, lane,
+};
+use crate::ops::OpKind;
+use async_trait::async_trait;
+use iggy::prelude::*;
+
+const SETUP_STREAMS: u32 = 5;
+const TOPICS_PER_STREAM: u32 = 2;
+const PARTITIONS_PER_TOPIC: u32 = 3;
+
+pub struct ConcurrentCrud;
+
+#[async_trait]
+impl Scenario for ConcurrentCrud {
+ fn name(&self) -> &'static str {
+ "concurrent-crud"
+ }
+
+ fn describe(&self) -> &'static str {
+ "Concurrent stream and topic CRUD operations.\n\n\
+ Phase 1: Setup - creates 5 streams with 2 topics each.\n\
+ Phase 2: Chaos - N workers run weighted create/delete/purge ops.\n\
+ Phase 3: Verify - post-run invariant checks.\n\n\
+ Focuses on detecting race conditions in metadata operations \
+ (create-while-deleting, purge-while-creating, etc.)."
+ }
+
+ async fn setup(
+ &self,
+ client: &IggyClient,
+ prefixes: &NamespacePrefixes,
+ ) -> Result<(), IggyError> {
+ tracing::info!(
+ "Setup: creating {SETUP_STREAMS} streams with {TOPICS_PER_STREAM} topics each..."
+ );
+ let setup = Setup::new(client, prefixes);
+ let defaults = TopicDefaults {
+ partitions: PARTITIONS_PER_TOPIC,
+ ..Default::default()
+ };
+ for i in 0..SETUP_STREAMS {
+ let stream = setup.create_stream(&format!("setup-s{i}")).await?;
+ for j in 0..TOPICS_PER_STREAM {
+ setup
+ .create_topic(&stream, &format!("setup-t{i}-{j}"), &defaults)
+ .await?;
+ }
+ }
+ Ok(())
+ }
+
+ fn lanes(&self) -> Vec<LaneSpec> {
+ vec![
+ lane("control-plane")
+ .ops(&[
+ (OpKind::CreateStream, 20.0),
+ (OpKind::DeleteStream, 15.0),
+ (OpKind::PurgeStream, 10.0),
+ (OpKind::CreateTopic, 20.0),
+ (OpKind::DeleteTopic, 15.0),
+ (OpKind::PurgeTopic, 10.0),
+ ])
+ .namespace(TargetNamespace::Churn)
+ .destructive()
+ .build(),
+ lane("data-plane")
+ .ops(&[(OpKind::SendMessages, 60.0), (OpKind::PollMessages, 40.0)])
+ .workers(2)
+ .scope(OriginScope::SetupOnly)
+ .namespace(TargetNamespace::Stable)
+ .build(),
+ ]
+ }
+}
diff --git a/core/chaos/src/scenarios/consumer_group_churn.rs b/core/chaos/src/scenarios/consumer_group_churn.rs
new file mode 100644
index 000000000..3e42794d4
--- /dev/null
+++ b/core/chaos/src/scenarios/consumer_group_churn.rs
@@ -0,0 +1,194 @@
+/* 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 super::{
+ LaneSpec, NamespacePrefixes, OriginScope, RateModel, Scenario, Setup, TargetNamespace,
+ TopicDefaults, VerifyContext, lane,
+};
+use crate::invariants::{self, ConsumerGroupHealthCheck, InvariantViolation};
+use crate::ops::OpKind;
+use crate::safe_name::SafeResourceName;
+use async_trait::async_trait;
+use iggy::prelude::*;
+use std::time::Duration;
+
+const STREAMS: u32 = 2;
+const TOPICS_PER_STREAM: u32 = 1;
+const PARTITIONS_PER_TOPIC: u32 = 6;
+const GROUPS_PER_TOPIC: u32 = 1;
+const FRESH_VERIFY_MEMBERS: u32 = 3;
+const REBALANCE_TIMEOUT: Duration = Duration::from_secs(10);
+const SENTINEL_PAYLOAD: &str = "cg-churn-sentinel";
+
+pub struct ConsumerGroupChurn;
+
+struct SetupResource {
+ stream_name: SafeResourceName,
+ topic_name: SafeResourceName,
+ group_name: SafeResourceName,
+}
+
+impl ConsumerGroupChurn {
+ fn setup_resources(prefixes: &NamespacePrefixes) -> Vec<SetupResource> {
+ let mut resources =
+ Vec::with_capacity((STREAMS * TOPICS_PER_STREAM * GROUPS_PER_TOPIC) as usize);
+ for i in 0..STREAMS {
+ let stream_name =
+ SafeResourceName::new(&prefixes.stable, &format!("cg-churn-stream-{i}"));
+ for j in 0..TOPICS_PER_STREAM {
+ let topic_name =
+ SafeResourceName::new(&prefixes.base, &format!("cg-churn-topic-{j}"));
+ for k in 0..GROUPS_PER_TOPIC {
+ let group_name =
+ SafeResourceName::new(&prefixes.base, &format!("cg-churn-group-{k}"));
+ resources.push(SetupResource {
+ stream_name: stream_name.clone(),
+ topic_name: topic_name.clone(),
+ group_name,
+ });
+ }
+ }
+ }
+ resources
+ }
+}
+
+#[async_trait]
+impl Scenario for ConsumerGroupChurn {
+ fn name(&self) -> &'static str {
+ "consumer-group-churn"
+ }
+
+ fn describe(&self) -> &'static str {
+ "Consumer group rebalancing under concurrent membership churn.\n\n\
+ Setup: creates 2 streams, each with 1 topic (6 partitions) and 1 consumer group.\n\
+ Sends a sentinel message to partition 1 of each topic.\n\n\
+ Lanes:\n\
+ - stable-members (global workers): poll/offset/join/get - workers that stay joined \
+ and experience partition reassignment as churn happens beneath them.\n\
+ - producers (2 workers): continuous message production to keep partitions active.\n\
+ - membership-churn (2 workers, 5 ops/sec): join/leave at 50/50 - continuous \
+ rebalancing pressure on the groups.\n\
+ - observers (1 worker): read-only monitoring via get_consumer_group/topic/stream.\n\n\
+ Verify: joins fresh members, confirms rebalance completes with unique partition \
+ assignments, polls concurrently, asserts exactly one member receives sentinel.\n\n\
+ Requires: server with heartbeat.enabled=true and a short heartbeat.interval (e.g. 2s)."
+ }
+
+ async fn setup(
+ &self,
+ client: &IggyClient,
+ prefixes: &NamespacePrefixes,
+ ) -> Result<(), IggyError> {
+ let setup = Setup::new(client, prefixes);
+ let defaults = TopicDefaults {
+ partitions: PARTITIONS_PER_TOPIC,
+ ..Default::default()
+ };
+ for i in 0..STREAMS {
+ let stream = setup.create_stream(&format!("cg-churn-stream-{i}")).await?;
+ for j in 0..TOPICS_PER_STREAM {
+ let topic = setup
+ .create_topic(&stream, &format!("cg-churn-topic-{j}"), &defaults)
+ .await?;
+ for k in 0..GROUPS_PER_TOPIC {
+ setup
+ .create_consumer_group(&stream, &topic, &format!("cg-churn-group-{k}"))
+ .await?;
+ }
+ setup.send_str(&stream, &topic, 1, SENTINEL_PAYLOAD).await?;
+ }
+ }
+ Ok(())
+ }
+
+ fn lanes(&self) -> Vec<LaneSpec> {
+ vec![
+ lane("stable-members")
+ .ops(&[
+ (OpKind::PollGroupMessages, 75.0),
+ (OpKind::StoreConsumerOffset, 15.0),
+ (OpKind::JoinConsumerGroup, 5.0),
+ (OpKind::GetConsumerGroup, 5.0),
+ ])
+ .scope(OriginScope::SetupOnly)
+ .namespace(TargetNamespace::Stable)
+ .build(),
+ lane("producers")
+ .op(OpKind::SendMessages, 100.0)
+ .workers(2)
+ .scope(OriginScope::SetupOnly)
+ .namespace(TargetNamespace::Stable)
+ .build(),
+ lane("membership-churn")
+ .ops(&[
+ (OpKind::JoinConsumerGroup, 50.0),
+ (OpKind::LeaveConsumerGroup, 50.0),
+ ])
+ .workers(2)
+ .scope(OriginScope::SetupOnly)
+ .namespace(TargetNamespace::Stable)
+ .rate(RateModel::TargetOpsPerSec(5))
+ .build(),
+ lane("observers")
+ .ops(&[
+ (OpKind::GetConsumerGroup, 50.0),
+ (OpKind::GetTopicDetails, 30.0),
+ (OpKind::GetStreamDetails, 20.0),
+ ])
+ .workers(1)
+ .scope(OriginScope::SetupOnly)
+ .namespace(TargetNamespace::Stable)
+ .build(),
+ ]
+ }
+
+ async fn verify(&self, ctx: &VerifyContext<'_>) -> Vec<InvariantViolation> {
+ let mut violations =
+ invariants::post_run_verify(ctx.client, ctx.shadow, ctx.prefixes).await;
+ violations.extend(
+ invariants::cross_client_consistency(ctx.server_address, ctx.transport, ctx.prefixes)
+ .await,
+ );
+
+ let resources = Self::setup_resources(ctx.prefixes);
+ for res in &resources {
+ let stream_id = Identifier::from_str_value(&res.stream_name).unwrap();
+ let topic_id = Identifier::from_str_value(&res.topic_name).unwrap();
+ let group_id = Identifier::from_str_value(&res.group_name).unwrap();
+ let label = format!("{}/{}/{}", res.stream_name, res.topic_name, res.group_name);
+
+ let health_check = ConsumerGroupHealthCheck {
+ label: &label,
+ server_address: ctx.server_address,
+ transport: ctx.transport,
+ admin_client: ctx.client,
+ stream_id: &stream_id,
+ topic_id: &topic_id,
+ group_id: &group_id,
+ fresh_member_count: FRESH_VERIFY_MEMBERS,
+ rebalance_timeout: REBALANCE_TIMEOUT,
+ sentinel_payload: SENTINEL_PAYLOAD.as_bytes(),
+ };
+
+ violations.extend(health_check.run().await);
+ }
+
+ violations
+ }
+}
diff --git a/core/chaos/src/scenarios/kit.rs b/core/chaos/src/scenarios/kit.rs
new file mode 100644
index 000000000..d40744e90
--- /dev/null
+++ b/core/chaos/src/scenarios/kit.rs
@@ -0,0 +1,117 @@
+/* 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 super::NamespacePrefixes;
+use crate::safe_name::SafeResourceName;
+use iggy::prelude::*;
+use std::str::FromStr;
+
+pub struct TopicDefaults {
+ pub partitions: u32,
+ pub compression: CompressionAlgorithm,
+ pub expiry: IggyExpiry,
+ pub max_size: MaxTopicSize,
+}
+
+impl Default for TopicDefaults {
+ fn default() -> Self {
+ Self {
+ partitions: 3,
+ compression: CompressionAlgorithm::None,
+ expiry: IggyExpiry::NeverExpire,
+ max_size: MaxTopicSize::Unlimited,
+ }
+ }
+}
+
+/// Convenience wrapper for the repeated create-stream / create-topic /
+/// create-consumer-group / send-messages boilerplate in scenario `setup()`.
+pub struct Setup<'a> {
+ client: &'a IggyClient,
+ prefixes: &'a NamespacePrefixes,
+}
+
+impl<'a> Setup<'a> {
+ pub fn new(client: &'a IggyClient, prefixes: &'a NamespacePrefixes) -> Self {
+ Self { client, prefixes }
+ }
+
+ pub async fn create_stream(&self, suffix: &str) -> Result<SafeResourceName, IggyError> {
+ let name = SafeResourceName::new(&self.prefixes.stable, suffix);
+ self.client.create_stream(&name).await?;
+ Ok(name)
+ }
+
+ pub async fn create_topic(
+ &self,
+ stream: &SafeResourceName,
+ suffix: &str,
+ defaults: &TopicDefaults,
+ ) -> Result<SafeResourceName, IggyError> {
+ let stream_id = Identifier::from_str_value(stream).unwrap();
+ let name = SafeResourceName::new(&self.prefixes.base, suffix);
+ self.client
+ .create_topic(
+ &stream_id,
+ &name,
+ defaults.partitions,
+ defaults.compression,
+ None,
+ defaults.expiry,
+ defaults.max_size,
+ )
+ .await?;
+ Ok(name)
+ }
+
+ pub async fn create_consumer_group(
+ &self,
+ stream: &SafeResourceName,
+ topic: &SafeResourceName,
+ suffix: &str,
+ ) -> Result<SafeResourceName, IggyError> {
+ let stream_id = Identifier::from_str_value(stream).unwrap();
+ let topic_id = Identifier::from_str_value(topic).unwrap();
+ let name = SafeResourceName::new(&self.prefixes.base, suffix);
+ self.client
+ .create_consumer_group(&stream_id, &topic_id, &name)
+ .await?;
+ Ok(name)
+ }
+
+ pub async fn send_str(
+ &self,
+ stream: &SafeResourceName,
+ topic: &SafeResourceName,
+ partition: u32,
+ payload: &str,
+ ) -> Result<(), IggyError> {
+ let stream_id = Identifier::from_str_value(stream).unwrap();
+ let topic_id = Identifier::from_str_value(topic).unwrap();
+ let msg = IggyMessage::from_str(payload).unwrap();
+ let mut messages = vec![msg];
+ self.client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(partition),
+ &mut messages,
+ )
+ .await
+ }
+}
diff --git a/core/chaos/src/scenarios/lane_builder.rs b/core/chaos/src/scenarios/lane_builder.rs
new file mode 100644
index 000000000..2e8e6b1f0
--- /dev/null
+++ b/core/chaos/src/scenarios/lane_builder.rs
@@ -0,0 +1,91 @@
+/* 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 super::{LaneSpec, OriginScope, RateModel, TargetNamespace};
+use crate::ops::OpKind;
+
+pub fn lane(label: &'static str) -> LaneBuilder {
+ LaneBuilder {
+ label,
+ ops: Vec::new(),
+ workers: None,
+ scope: OriginScope::Any,
+ namespace: TargetNamespace::Both,
+ rate: RateModel::Unbounded,
+ destructive: false,
+ }
+}
+
+pub struct LaneBuilder {
+ label: &'static str,
+ ops: Vec<(OpKind, f64)>,
+ workers: Option<u32>,
+ scope: OriginScope,
+ namespace: TargetNamespace,
+ rate: RateModel,
+ destructive: bool,
+}
+
+impl LaneBuilder {
+ pub fn ops(mut self, slice: &[(OpKind, f64)]) -> Self {
+ self.ops = slice.to_vec();
+ self
+ }
+
+ pub fn op(mut self, kind: OpKind, weight: f64) -> Self {
+ self.ops.push((kind, weight));
+ self
+ }
+
+ pub fn workers(mut self, n: u32) -> Self {
+ self.workers = Some(n);
+ self
+ }
+
+ pub fn scope(mut self, s: OriginScope) -> Self {
+ self.scope = s;
+ self
+ }
+
+ pub fn namespace(mut self, ns: TargetNamespace) -> Self {
+ self.namespace = ns;
+ self
+ }
+
+ pub fn rate(mut self, r: RateModel) -> Self {
+ self.rate = r;
+ self
+ }
+
+ pub fn destructive(mut self) -> Self {
+ self.destructive = true;
+ self
+ }
+
+ pub fn build(self) -> LaneSpec {
+ LaneSpec {
+ label: self.label,
+ op_weights: self.ops,
+ worker_count: self.workers,
+ scope: self.scope,
+ namespace: self.namespace,
+ rate: self.rate,
+ allow_destructive: self.destructive,
+ }
+ }
+}
diff --git a/core/chaos/src/scenarios/mixed_workload.rs b/core/chaos/src/scenarios/mixed_workload.rs
new file mode 100644
index 000000000..b123875d7
--- /dev/null
+++ b/core/chaos/src/scenarios/mixed_workload.rs
@@ -0,0 +1,76 @@
+/* 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 super::{LaneSpec, Scenario, TargetNamespace, lane};
+use crate::ops::OpKind;
+use async_trait::async_trait;
+
+pub struct MixedWorkload;
+
+#[async_trait]
+impl Scenario for MixedWorkload {
+ fn name(&self) -> &'static str {
+ "mixed-workload"
+ }
+
+ fn describe(&self) -> &'static str {
+ "Full chaos: all operation types interleaved.\n\n\
+ All workers independently generate a weighted mix of every operation type: \
+ create/delete streams and topics, send/poll messages, purge, partition \
+ management, consumer groups, and consumer offsets.\n\n\
+ Default weights: SendMessages 40%, PollMessages 25%, \
+ CreateStream/Topic 10%, DeleteStream/Topic 8%, \
+ Purge/Segments 7%, ConsumerGroup 5%, Offsets 5%."
+ }
+
+ fn lanes(&self) -> Vec<LaneSpec> {
+ vec![
+ lane("data-plane")
+ .ops(&[(OpKind::SendMessages, 60.0), (OpKind::PollMessages, 40.0)])
+ .build(),
+ lane("control-plane")
+ .ops(&[
+ (OpKind::CreateStream, 10.0),
+ (OpKind::DeleteStream, 8.0),
+ (OpKind::PurgeStream, 4.0),
+ (OpKind::CreateTopic, 10.0),
+ (OpKind::DeleteTopic, 8.0),
+ (OpKind::PurgeTopic, 4.0),
+ (OpKind::DeleteSegments, 6.0),
+ (OpKind::CreatePartitions, 4.0),
+ (OpKind::DeletePartitions, 2.0),
+ (OpKind::CreateConsumerGroup, 6.0),
+ (OpKind::DeleteConsumerGroup, 4.0),
+ (OpKind::StoreConsumerOffset, 4.0),
+ ])
+ .workers(4)
+ .namespace(TargetNamespace::Churn)
+ .destructive()
+ .build(),
+ lane("admin")
+ .ops(&[
+ (OpKind::GetStreams, 40.0),
+ (OpKind::GetStreamDetails, 30.0),
+ (OpKind::GetTopicDetails, 20.0),
+ (OpKind::GetStats, 10.0),
+ ])
+ .workers(2)
+ .build(),
+ ]
+ }
+}
diff --git a/core/chaos/src/scenarios/mod.rs b/core/chaos/src/scenarios/mod.rs
new file mode 100644
index 000000000..1d10a601d
--- /dev/null
+++ b/core/chaos/src/scenarios/mod.rs
@@ -0,0 +1,208 @@
+/* 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.
+ */
+
+pub mod concurrent_crud;
+pub mod consumer_group_churn;
+pub mod kit;
+pub mod lane_builder;
+pub mod mixed_workload;
+pub mod segment_rotation;
+pub mod stale_consumers;
+
+pub use kit::{Setup, TopicDefaults};
+pub use lane_builder::lane;
+
+use crate::args::{ScenarioName, Transport};
+use crate::invariants::{self, InvariantViolation};
+use crate::ops::OpKind;
+use crate::shadow::ShadowState;
+use async_trait::async_trait;
+use iggy::prelude::*;
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+#[allow(dead_code)]
+pub enum OriginScope {
+ #[default]
+ Any,
+ /// Only target resources created during setup (skip creative ops).
+ SetupOnly,
+ /// Only target dynamically-created resources.
+ DynamicOnly,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Namespace {
+ Stable,
+ Churn,
+}
+
+/// Controls which namespace a lane's operations target.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum TargetNamespace {
+ Stable,
+ Churn,
+ #[default]
+ Both,
+}
+
+/// Derived stable/churn prefixes from a single base prefix.
+#[derive(Debug, Clone)]
+pub struct NamespacePrefixes {
+ pub base: String,
+ pub stable: String,
+ pub churn: String,
+}
+
+impl NamespacePrefixes {
+ pub fn from_base(base: &str) -> Self {
+ Self {
+ base: base.to_owned(),
+ stable: format!("{base}s-"),
+ churn: format!("{base}c-"),
+ }
+ }
+
+ pub fn matches(&self, name: &str) -> bool {
+ name.starts_with(&self.stable) || name.starts_with(&self.churn)
+ }
+
+ pub fn namespace_of(&self, stream_name: &str) -> Option<Namespace> {
+ if stream_name.starts_with(&self.stable) {
+ Some(Namespace::Stable)
+ } else if stream_name.starts_with(&self.churn) {
+ Some(Namespace::Churn)
+ } else {
+ None
+ }
+ }
+
+ /// Prefix for creating new streams in the given lane namespace.
+ /// `Both` creates in churn - stable resources are only created during setup.
+ pub fn create_prefix(&self, ns: TargetNamespace) -> &str {
+ match ns {
+ TargetNamespace::Stable => &self.stable,
+ TargetNamespace::Churn | TargetNamespace::Both => &self.churn,
+ }
+ }
+}
+
+#[derive(Debug, Clone, Copy, Default)]
+pub enum RateModel {
+ #[default]
+ Unbounded,
+ /// Target N ops/sec per worker; sleep between ops to approximate.
+ TargetOpsPerSec(u32),
+}
+
+/// One lane of worker execution with its own op weight distribution.
+pub struct LaneSpec {
+ pub label: &'static str,
+ pub op_weights: Vec<(OpKind, f64)>,
+ /// `None` = use the global `--workers` CLI value.
+ pub worker_count: Option<u32>,
+ pub scope: OriginScope,
+ pub namespace: TargetNamespace,
+ pub rate: RateModel,
+ pub allow_destructive: bool,
+}
+
+#[derive(Debug, Clone, Copy, Default)]
+#[allow(dead_code)]
+pub enum CleanupPolicy {
+ #[default]
+ DeletePrefix,
+ NoOp,
+}
+
+/// Everything the default `verify()` implementation needs.
+pub struct VerifyContext<'a> {
+ pub client: &'a IggyClient,
+ pub shadow: &'a ShadowState,
+ pub prefixes: &'a NamespacePrefixes,
+ pub server_address: &'a str,
+ pub transport: Transport,
+}
+
+#[async_trait]
+pub trait Scenario: Send + Sync {
+ fn name(&self) -> &'static str;
+ fn describe(&self) -> &'static str;
+
+ /// Create pre-existing resources. Default: no-op.
+ async fn setup(
+ &self,
+ _client: &IggyClient,
+ _prefixes: &NamespacePrefixes,
+ ) -> Result<(), IggyError> {
+ Ok(())
+ }
+
+ /// Declare worker lanes (replaces op_weights).
+ fn lanes(&self) -> Vec<LaneSpec>;
+
+ /// Post-run verification. Default: generic shadow + cross-client checks.
+ async fn verify(&self, ctx: &VerifyContext<'_>) -> Vec<InvariantViolation> {
+ let mut violations =
+ invariants::post_run_verify(ctx.client, ctx.shadow, ctx.prefixes).await;
+ violations.extend(
+ invariants::cross_client_consistency(ctx.server_address, ctx.transport, ctx.prefixes)
+ .await,
+ );
+ violations
+ }
+
+ /// Cleanup strategy. Default: delete all prefixed streams.
+ fn cleanup_policy(&self) -> CleanupPolicy {
+ CleanupPolicy::DeletePrefix
+ }
+}
+
+pub fn create_scenario(name: ScenarioName) -> Box<dyn Scenario> {
+ match name {
+ ScenarioName::ConcurrentCrud => Box::new(concurrent_crud::ConcurrentCrud),
+ ScenarioName::SegmentRotation => Box::new(segment_rotation::SegmentRotation),
+ ScenarioName::MixedWorkload => Box::new(mixed_workload::MixedWorkload),
+ ScenarioName::StaleConsumers => Box::new(stale_consumers::StaleConsumers),
+ ScenarioName::ConsumerGroupChurn => Box::new(consumer_group_churn::ConsumerGroupChurn),
+ }
+}
+
+pub fn list_scenarios() -> Vec<(&'static str, &'static str)> {
+ vec![
+ (
+ concurrent_crud::ConcurrentCrud.name(),
+ concurrent_crud::ConcurrentCrud.describe(),
+ ),
+ (
+ segment_rotation::SegmentRotation.name(),
+ segment_rotation::SegmentRotation.describe(),
+ ),
+ (
+ mixed_workload::MixedWorkload.name(),
+ mixed_workload::MixedWorkload.describe(),
+ ),
+ (
+ stale_consumers::StaleConsumers.name(),
+ stale_consumers::StaleConsumers.describe(),
+ ),
+ (
+ consumer_group_churn::ConsumerGroupChurn.name(),
+ consumer_group_churn::ConsumerGroupChurn.describe(),
+ ),
+ ]
+}
diff --git a/core/chaos/src/scenarios/segment_rotation.rs b/core/chaos/src/scenarios/segment_rotation.rs
new file mode 100644
index 000000000..3d15a0370
--- /dev/null
+++ b/core/chaos/src/scenarios/segment_rotation.rs
@@ -0,0 +1,80 @@
+/* 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 super::{
+ LaneSpec, NamespacePrefixes, OriginScope, RateModel, Scenario, Setup, TargetNamespace,
+ TopicDefaults, lane,
+};
+use crate::ops::OpKind;
+use async_trait::async_trait;
+use iggy::prelude::*;
+
+pub struct SegmentRotation;
+
+#[async_trait]
+impl Scenario for SegmentRotation {
+ fn name(&self) -> &'static str {
+ "segment-rotation"
+ }
+
+ fn describe(&self) -> &'static str {
+ "Segment rotation stress test.\n\n\
+ Phase 1: Setup - creates 1 stream with 1 topic and 1 partition.\n\
+ Phase 2: Chaos - all workers send small messages to the same partition \
+ while concurrently polling.\n\
+ Phase 3: Verify - offset monotonicity, no gaps, watermark violations.\n\n\
+ Targets segment rotation logic by producing high message throughput \
+ to a single partition."
+ }
+
+ async fn setup(
+ &self,
+ client: &IggyClient,
+ prefixes: &NamespacePrefixes,
+ ) -> Result<(), IggyError> {
+ tracing::info!("Setup: creating 1 stream with 1 topic and 1 partition...");
+ let setup = Setup::new(client, prefixes);
+ let defaults = TopicDefaults {
+ partitions: 1,
+ ..Default::default()
+ };
+ let stream = setup.create_stream("seg-rot-stream").await?;
+ setup
+ .create_topic(&stream, "seg-rot-topic", &defaults)
+ .await?;
+ Ok(())
+ }
+
+ fn lanes(&self) -> Vec<LaneSpec> {
+ vec![
+ lane("data-plane")
+ .ops(&[(OpKind::SendMessages, 55.0), (OpKind::PollMessages, 45.0)])
+ .scope(OriginScope::SetupOnly)
+ .namespace(TargetNamespace::Stable)
+ .build(),
+ lane("segment-cleanup")
+ .op(OpKind::DeleteSegments, 100.0)
+ .workers(1)
+ .scope(OriginScope::SetupOnly)
+ .namespace(TargetNamespace::Stable)
+ .rate(RateModel::TargetOpsPerSec(2))
+ .destructive()
+ .build(),
+ ]
+ }
+}
diff --git a/core/chaos/src/scenarios/stale_consumers.rs b/core/chaos/src/scenarios/stale_consumers.rs
new file mode 100644
index 000000000..7d95f5c78
--- /dev/null
+++ b/core/chaos/src/scenarios/stale_consumers.rs
@@ -0,0 +1,270 @@
+/* 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 super::{LaneSpec, NamespacePrefixes, Scenario, Setup, TopicDefaults, VerifyContext};
+use crate::client::create_stale_client;
+use crate::eventually::{await_members_count, await_rebalance_complete};
+use crate::invariants::{ConsumerGroupHealthCheck, InvariantViolation};
+use crate::safe_name::SafeResourceName;
+use crate::violation::ViolationCategory;
+use async_trait::async_trait;
+use iggy::prelude::*;
+use std::time::Duration;
+use tracing::info;
+
+const PARTITION_COUNT: u32 = 3;
+const STALE_MEMBER_COUNT: u32 = 3;
+const FRESH_MEMBER_COUNT: u32 = 3;
+const SENTINEL_PAYLOAD: &str = "stale-consumers-sentinel";
+/// Generous timeout for heartbeat eviction (server interval * 1.2 + propagation).
+const EVICTION_TIMEOUT: Duration = Duration::from_secs(30);
+const REBALANCE_TIMEOUT: Duration = Duration::from_secs(10);
+
+const STREAM_SUFFIX: &str = "stale-cg-stream";
+const TOPIC_SUFFIX: &str = "stale-cg-topic";
+const GROUP_SUFFIX: &str = "stale-cg-group";
+
+pub struct StaleConsumers;
+
+impl StaleConsumers {
+ fn stream_name(prefixes: &NamespacePrefixes) -> SafeResourceName {
+ SafeResourceName::new(&prefixes.stable, STREAM_SUFFIX)
+ }
+
+ fn topic_name(prefixes: &NamespacePrefixes) -> SafeResourceName {
+ SafeResourceName::new(&prefixes.base, TOPIC_SUFFIX)
+ }
+
+ fn group_name(prefixes: &NamespacePrefixes) -> SafeResourceName {
+ SafeResourceName::new(&prefixes.base, GROUP_SUFFIX)
+ }
+}
+
+#[async_trait]
+impl Scenario for StaleConsumers {
+ fn name(&self) -> &'static str {
+ "stale-consumers"
+ }
+
+ fn describe(&self) -> &'static str {
+ "Consumer group rebalancing after heartbeat eviction.\n\n\
+ Phase 1: Setup - creates 1 stream, 1 topic (3 partitions), 1 consumer group,\n\
+ and produces a single sentinel message to partition 1.\n\
+ Phase 2: No chaos lanes (skip with --duration 1s or --ops 0).\n\
+ Phase 3: Verify - joins stale members (1h heartbeat), lets them poll\n\
+ without committing offsets (exercising the fetched-but-uncommitted path),\n\
+ waits for server eviction, joins fresh members, polls concurrently, and asserts:\n\
+ - stale members were evicted (members_count drops to 0)\n\
+ - rebalance assigns all partitions with no duplicates\n\
+ - exactly one fresh member receives the sentinel message\n\n\
+ Requires: server with heartbeat.enabled=true and a short heartbeat.interval (e.g. 2s)."
+ }
+
+ async fn setup(
+ &self,
+ client: &IggyClient,
+ prefixes: &NamespacePrefixes,
+ ) -> Result<(), IggyError> {
+ info!(
+ "Setup: creating stream, topic ({PARTITION_COUNT} partitions), consumer group, sentinel message"
+ );
+ let setup = Setup::new(client, prefixes);
+ let defaults = TopicDefaults {
+ partitions: PARTITION_COUNT,
+ ..Default::default()
+ };
+ let stream = setup.create_stream(STREAM_SUFFIX).await?;
+ let topic = setup.create_topic(&stream, TOPIC_SUFFIX, &defaults).await?;
+ setup
+ .create_consumer_group(&stream, &topic, GROUP_SUFFIX)
+ .await?;
+ setup.send_str(&stream, &topic, 1, SENTINEL_PAYLOAD).await?;
+ Ok(())
+ }
+
+ fn lanes(&self) -> Vec<LaneSpec> {
+ vec![]
+ }
+
+ async fn verify(&self, ctx: &VerifyContext<'_>) -> Vec<InvariantViolation> {
+ match run_verification(ctx).await {
+ Ok(violations) => violations,
+ Err(v) => vec![v],
+ }
+ }
+}
+
+async fn run_verification(
+ ctx: &VerifyContext<'_>,
+) -> Result<Vec<InvariantViolation>, InvariantViolation> {
+ let stream_name = StaleConsumers::stream_name(ctx.prefixes);
+ let topic_name = StaleConsumers::topic_name(ctx.prefixes);
+ let group_name = StaleConsumers::group_name(ctx.prefixes);
+
+ let stream_id = Identifier::from_str_value(&stream_name).unwrap();
+ let topic_id = Identifier::from_str_value(&topic_name).unwrap();
+ let group_id = Identifier::from_str_value(&group_name).unwrap();
+
+ // --- Phase 1: Join stale members ---
+ info!("Verify: joining {STALE_MEMBER_COUNT} stale members (1h heartbeat)");
+ let mut stale_clients = Vec::with_capacity(STALE_MEMBER_COUNT as usize);
+ for i in 0..STALE_MEMBER_COUNT {
+ let client =
+ create_stale_client(ctx.server_address)
+ .await
+ .map_err(|e| InvariantViolation {
+ category: ViolationCategory::Infrastructure,
+ kind: "stale_client_create_failed",
+ description: format!("Failed to create stale client #{i}: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ })?;
+ client
+ .join_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .map_err(|e| InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "stale_join_failed",
+ description: format!("Stale client #{i} failed to join group: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ })?;
+ stale_clients.push(client);
+ }
+
+ await_members_count(
+ ctx.client,
+ &stream_id,
+ &topic_id,
+ &group_id,
+ STALE_MEMBER_COUNT,
+ EVICTION_TIMEOUT,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "members_never_joined",
+ description: format!("Expected {STALE_MEMBER_COUNT} members after join, timed out: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ })?;
+ info!("Verify: {STALE_MEMBER_COUNT} stale members joined");
+
+ await_rebalance_complete(
+ ctx.client,
+ &stream_id,
+ &topic_id,
+ &group_id,
+ REBALANCE_TIMEOUT,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "stale_rebalance_incomplete",
+ description: format!(
+ "Rebalance for stale members did not complete within {REBALANCE_TIMEOUT:?}: {e}"
+ ),
+ context: String::new(),
+ trace_anchor: None,
+ })?;
+ info!("Verify: stale members have partitions assigned, polling without commit");
+
+ // --- Phase 2: Stale members poll WITHOUT committing offsets ---
+ let mut stale_saw_sentinel = false;
+ for (i, client) in stale_clients.iter().enumerate() {
+ let consumer = Consumer::group(group_id.clone());
+ let polled = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ None,
+ &consumer,
+ &PollingStrategy::next(),
+ 10,
+ false,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "stale_poll_failed",
+ description: format!("Stale member #{i} poll failed: {e}"),
+ context: String::new(),
+ trace_anchor: None,
+ })?;
+ if contains_sentinel(&polled) {
+ stale_saw_sentinel = true;
+ info!("Verify: stale member #{i} observed sentinel (not committed)");
+ }
+ }
+ if !stale_saw_sentinel {
+ return Err(InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "stale_never_claimed_sentinel",
+ description: "No stale member observed the sentinel during uncommitted poll".into(),
+ context: format!(
+ "All {PARTITION_COUNT} partitions assigned across {STALE_MEMBER_COUNT} members; \
+ sentinel on partition 1 should be reachable"
+ ),
+ trace_anchor: None,
+ });
+ }
+ info!("Verify: stale poll phase complete, waiting for eviction");
+
+ // --- Phase 3: Wait for heartbeat eviction ---
+ await_members_count(
+ ctx.client,
+ &stream_id,
+ &topic_id,
+ &group_id,
+ 0,
+ EVICTION_TIMEOUT,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ category: ViolationCategory::ConsumerGroup,
+ kind: "members_never_evicted",
+ description: format!("Stale members were not evicted within {EVICTION_TIMEOUT:?}: {e}"),
+ context: "Ensure server has heartbeat.enabled=true with a short interval (e.g. 2s)".into(),
+ trace_anchor: None,
+ })?;
+ info!("Verify: all stale members evicted");
+ drop(stale_clients);
+
+ // --- Phase 4: Fresh member health check (shared logic) ---
+ let health_check = ConsumerGroupHealthCheck {
+ label: "stale-consumers",
+ server_address: ctx.server_address,
+ transport: ctx.transport,
+ admin_client: ctx.client,
+ stream_id: &stream_id,
+ topic_id: &topic_id,
+ group_id: &group_id,
+ fresh_member_count: FRESH_MEMBER_COUNT,
+ rebalance_timeout: REBALANCE_TIMEOUT,
+ sentinel_payload: SENTINEL_PAYLOAD.as_bytes(),
+ };
+
+ Ok(health_check.run().await)
+}
+
+fn contains_sentinel(polled: &PolledMessages) -> bool {
+ polled
+ .messages
+ .iter()
+ .any(|m| m.payload == SENTINEL_PAYLOAD)
+}
diff --git a/core/chaos/src/shadow.rs b/core/chaos/src/shadow.rs
new file mode 100644
index 000000000..c7b577fbb
--- /dev/null
+++ b/core/chaos/src/shadow.rs
@@ -0,0 +1,276 @@
+/* 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 crate::scenarios::{Namespace, NamespacePrefixes, OriginScope, TargetNamespace};
+use rand::{Rng, RngExt};
+use std::collections::{HashMap, HashSet, VecDeque};
+use std::time::{Duration, Instant};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ResourceOrigin {
+ Setup,
+ Dynamic,
+}
+
+const TOMBSTONE_RETENTION: Duration = Duration::from_secs(30);
+
+pub struct ShadowState {
+ pub streams: HashMap<String, ShadowStream>,
+ prefixes: NamespacePrefixes,
+ /// Recently deleted resource names with deletion timestamp, for error classification.
+ recently_deleted: VecDeque<(String, Instant)>,
+ /// Recently purged (stream, topic) pairs - offsets reset after purge.
+ recently_purged: VecDeque<(String, Option<String>, Instant)>,
+}
+
+pub struct ShadowStream {
+ pub topics: HashMap<String, ShadowTopic>,
+ pub origin: ResourceOrigin,
+ pub namespace: Namespace,
+}
+
+pub struct ShadowTopic {
+ pub partitions: u32,
+ pub consumer_groups: HashSet<String>,
+ pub origin: ResourceOrigin,
+}
+
+impl ShadowState {
+ pub fn new(prefixes: NamespacePrefixes) -> Self {
+ Self {
+ streams: HashMap::new(),
+ prefixes,
+ recently_deleted: VecDeque::new(),
+ recently_purged: VecDeque::new(),
+ }
+ }
+
+ pub fn stream_exists(&self, name: &str) -> bool {
+ self.streams.contains_key(name)
+ }
+
+ pub fn topic_exists(&self, stream: &str, topic: &str) -> bool {
+ self.streams
+ .get(stream)
+ .is_some_and(|s| s.topics.contains_key(topic))
+ }
+
+ pub fn get_topic(&self, stream: &str, topic: &str) -> Option<&ShadowTopic> {
+ self.streams.get(stream).and_then(|s| s.topics.get(topic))
+ }
+
+ pub fn random_stream(
+ &self,
+ rng: &mut impl Rng,
+ scope: OriginScope,
+ namespace: TargetNamespace,
+ ) -> Option<&str> {
+ let candidates: Vec<_> = self
+ .streams
+ .iter()
+ .filter(|(_, s)| {
+ matches_scope(s.origin, scope) && matches_namespace(s.namespace, namespace)
+ })
+ .collect();
+ if candidates.is_empty() {
+ return None;
+ }
+ let idx = rng.random_range(0..candidates.len());
+ Some(candidates[idx].0.as_str())
+ }
+
+ /// Returns (stream_name, topic_name) for a random topic that exists.
+ pub fn random_topic(
+ &self,
+ rng: &mut impl Rng,
+ scope: OriginScope,
+ namespace: TargetNamespace,
+ ) -> Option<(&str, &str)> {
+ let streams_with_topics: Vec<_> = self
+ .streams
+ .iter()
+ .filter(|(_, s)| {
+ matches_scope(s.origin, scope) && matches_namespace(s.namespace, namespace)
+ })
+ .flat_map(|(sname, s)| {
+ s.topics
+ .iter()
+ .filter(move |(_, t)| matches_scope(t.origin, scope))
+ .map(move |(tname, _)| (sname.as_str(), tname.as_str()))
+ })
+ .collect();
+ if streams_with_topics.is_empty() {
+ return None;
+ }
+ Some(streams_with_topics[rng.random_range(0..streams_with_topics.len())])
+ }
+
+ /// Returns (stream_name, topic_name, partition_count) for a random topic.
+ pub fn random_topic_with_partitions(
+ &self,
+ rng: &mut impl Rng,
+ scope: OriginScope,
+ namespace: TargetNamespace,
+ ) -> Option<(&str, &str, u32)> {
+ let (s, t) = self.random_topic(rng, scope, namespace)?;
+ let partitions = self.streams[s].topics[t].partitions;
+ Some((s, t, partitions))
+ }
+
+ pub fn apply_create_stream(&mut self, name: String, origin: ResourceOrigin) {
+ let namespace = self
+ .prefixes
+ .namespace_of(&name)
+ .unwrap_or(Namespace::Churn);
+ self.streams.insert(
+ name,
+ ShadowStream {
+ topics: HashMap::new(),
+ origin,
+ namespace,
+ },
+ );
+ }
+
+ pub fn apply_delete_stream(&mut self, name: &str) {
+ if self.streams.remove(name).is_some() {
+ self.recently_deleted
+ .push_back((name.to_owned(), Instant::now()));
+ }
+ }
+
+ pub fn apply_create_topic(
+ &mut self,
+ stream: &str,
+ name: String,
+ partitions: u32,
+ origin: ResourceOrigin,
+ ) {
+ if let Some(s) = self.streams.get_mut(stream) {
+ s.topics.insert(
+ name,
+ ShadowTopic {
+ partitions,
+ consumer_groups: HashSet::new(),
+ origin,
+ },
+ );
+ }
+ }
+
+ pub fn apply_delete_topic(&mut self, stream: &str, topic: &str) {
+ if let Some(s) = self.streams.get_mut(stream)
+ && s.topics.remove(topic).is_some()
+ {
+ let key = format!("{stream}/{topic}");
+ self.recently_deleted.push_back((key, Instant::now()));
+ }
+ }
+
+ pub fn apply_create_partitions(&mut self, stream: &str, topic: &str, count: u32) {
+ if let Some(t) = self
+ .streams
+ .get_mut(stream)
+ .and_then(|s| s.topics.get_mut(topic))
+ {
+ t.partitions += count;
+ }
+ }
+
+ pub fn apply_delete_partitions(&mut self, stream: &str, topic: &str, count: u32) {
+ if let Some(t) = self
+ .streams
+ .get_mut(stream)
+ .and_then(|s| s.topics.get_mut(topic))
+ {
+ t.partitions = t.partitions.saturating_sub(count);
+ }
+ }
+
+ pub fn apply_create_consumer_group(&mut self, stream: &str, topic: &str, name: String) {
+ if let Some(t) = self
+ .streams
+ .get_mut(stream)
+ .and_then(|s| s.topics.get_mut(topic))
+ {
+ t.consumer_groups.insert(name);
+ }
+ }
+
+ pub fn apply_delete_consumer_group(&mut self, stream: &str, topic: &str, name: &str) {
+ if let Some(t) = self
+ .streams
+ .get_mut(stream)
+ .and_then(|s| s.topics.get_mut(topic))
+ {
+ t.consumer_groups.remove(name);
+ }
+ }
+
+ pub fn was_recently_deleted(&self, name: &str) -> bool {
+ self.recently_deleted.iter().any(|(n, _)| n == name)
+ }
+
+ /// Record that a stream or topic was purged (offsets reset).
+ /// `topic` is `None` for stream-level purge (all topics affected).
+ pub fn record_purge(&mut self, stream: String, topic: Option<String>) {
+ self.recently_purged
+ .push_back((stream, topic, Instant::now()));
+ }
+
+ /// Check whether offsets for this partition may have been reset by a recent purge.
+ pub fn was_recently_purged(&self, stream: &str, topic: &str) -> bool {
+ self.recently_purged
+ .iter()
+ .any(|(s, t, _)| s == stream && (t.is_none() || t.as_deref() == Some(topic)))
+ }
+
+ pub fn prune_tombstones(&mut self) {
+ let cutoff = Instant::now() - TOMBSTONE_RETENTION;
+ while self
+ .recently_deleted
+ .front()
+ .is_some_and(|(_, ts)| *ts < cutoff)
+ {
+ self.recently_deleted.pop_front();
+ }
+ while self
+ .recently_purged
+ .front()
+ .is_some_and(|(_, _, ts)| *ts < cutoff)
+ {
+ self.recently_purged.pop_front();
+ }
+ }
+}
+
+fn matches_scope(origin: ResourceOrigin, scope: OriginScope) -> bool {
+ match scope {
+ OriginScope::Any => true,
+ OriginScope::SetupOnly => origin == ResourceOrigin::Setup,
+ OriginScope::DynamicOnly => origin == ResourceOrigin::Dynamic,
+ }
+}
+
+fn matches_namespace(ns: Namespace, target_ns: TargetNamespace) -> bool {
+ match target_ns {
+ TargetNamespace::Stable => ns == Namespace::Stable,
+ TargetNamespace::Churn => ns == Namespace::Churn,
+ TargetNamespace::Both => true,
+ }
+}
diff --git a/core/chaos/src/trace.rs b/core/chaos/src/trace.rs
new file mode 100644
index 000000000..ffa51c9d8
--- /dev/null
+++ b/core/chaos/src/trace.rs
@@ -0,0 +1,211 @@
+/* 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.
+ */
+
+//! Channel-based trace writer.
+//!
+//! Workers serialize trace entries and send them through a bounded mpsc channel
+//! to a single background writer task. This avoids per-worker file handles and
+//! blocking tokio threads on I/O flushes.
+
+use crate::ops::{Op, OpDetail, OpOutcome};
+use serde::Serialize;
+use std::fs::File;
+use std::io::{BufWriter, Write};
+use std::path::Path;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+use tokio::sync::mpsc;
+
+const CHANNEL_CAPACITY: usize = 4096;
+const FLUSH_INTERVAL_ENTRIES: usize = 64;
+const FLUSH_INTERVAL: Duration = Duration::from_millis(100);
+
+/// Serialized JSON line ready for writing. Pre-serialized on the sender side
+/// to keep the writer's work minimal.
+struct TraceMessage {
+ json_line: Vec<u8>,
+}
+
+/// Per-worker sender. Cheap to clone (shares the channel).
+#[derive(Clone)]
+pub struct TraceSender {
+ tx: mpsc::Sender<TraceMessage>,
+ worker_id: u32,
+ lane: &'static str,
+ start: Instant,
+ dropped: Arc<AtomicU64>,
+}
+
+/// Creates per-worker `TraceSender` instances that share the same channel.
+pub struct TraceSenderFactory {
+ tx: mpsc::Sender<TraceMessage>,
+ start: Instant,
+ dropped: Arc<AtomicU64>,
+}
+
+/// Background writer that drains the channel and writes to a single trace file.
+pub struct TraceWriter {
+ rx: mpsc::Receiver<TraceMessage>,
+}
+
+#[derive(Serialize)]
+struct TraceEntry<'a> {
+ seq: u64,
+ t_us: u64,
+ worker: u32,
+ lane: &'static str,
+ phase: &'static str,
+ #[serde(flatten)]
+ data: TraceData<'a>,
+}
+
+#[derive(Serialize)]
+#[serde(untagged)]
+enum TraceData<'a> {
+ Intent {
+ op: &'a Op,
+ },
+ Outcome {
+ result: &'static str,
+ latency_us: u64,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ detail: Option<&'a OpDetail>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ error: Option<&'a str>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ error_context: Option<&'a str>,
+ },
+}
+
+/// Create a trace channel pair. The factory stays with the runner to create
+/// per-worker senders. The writer runs in a background task.
+pub fn create_trace_channel(start: Instant) -> (TraceWriter, TraceSenderFactory) {
+ let (tx, rx) = mpsc::channel(CHANNEL_CAPACITY);
+ let dropped = Arc::new(AtomicU64::new(0));
+ let writer = TraceWriter { rx };
+ let factory = TraceSenderFactory { tx, start, dropped };
+ (writer, factory)
+}
+
+impl TraceSenderFactory {
+ pub fn sender(&self, worker_id: u32, lane: &'static str) -> TraceSender {
+ TraceSender {
+ tx: self.tx.clone(),
+ worker_id,
+ lane,
+ start: self.start,
+ dropped: Arc::clone(&self.dropped),
+ }
+ }
+
+ pub fn dropped(&self) -> u64 {
+ self.dropped.load(Ordering::Relaxed)
+ }
+}
+
+impl TraceSender {
+ pub fn write_intent(&self, seq: u64, op: &Op) {
+ let entry = TraceEntry {
+ seq,
+ t_us: self.start.elapsed().as_micros() as u64,
+ worker: self.worker_id,
+ lane: self.lane,
+ phase: "intent",
+ data: TraceData::Intent { op },
+ };
+ self.send(entry);
+ }
+
+ pub fn write_outcome(&self, seq: u64, outcome: &OpOutcome, latency: Duration) {
+ let (detail, error, error_context) = match outcome {
+ OpOutcome::Success { detail } => (detail.as_ref(), None, None),
+ OpOutcome::Error { error, context, .. } => {
+ (None, Some(error.as_str()), Some(context.as_str()))
+ }
+ };
+ let entry = TraceEntry {
+ seq,
+ t_us: self.start.elapsed().as_micros() as u64,
+ worker: self.worker_id,
+ lane: self.lane,
+ phase: "outcome",
+ data: TraceData::Outcome {
+ result: outcome.result_tag(),
+ latency_us: latency.as_micros() as u64,
+ detail,
+ error,
+ error_context,
+ },
+ };
+ self.send(entry);
+ }
+
+ fn send(&self, entry: TraceEntry<'_>) {
+ let mut buf = match serde_json::to_vec(&entry) {
+ Ok(b) => b,
+ Err(_) => return,
+ };
+ buf.push(b'\n');
+
+ if self.tx.try_send(TraceMessage { json_line: buf }).is_err() {
+ self.dropped.fetch_add(1, Ordering::Relaxed);
+ }
+ }
+}
+
+impl TraceWriter {
+ /// Run the writer loop. Blocks until all senders are dropped (channel closed).
+ /// Call this in `tokio::task::spawn_blocking` or a dedicated thread.
+ pub fn run(mut self, path: &Path) -> std::io::Result<()> {
+ let file = File::create(path)?;
+ let mut writer = BufWriter::new(file);
+ let mut pending = 0usize;
+ let mut last_flush = Instant::now();
+
+ while let Some(msg) = self.rx.blocking_recv() {
+ let _ = writer.write_all(&msg.json_line);
+ pending += 1;
+
+ // Drain any immediately available entries before flushing
+ while pending < FLUSH_INTERVAL_ENTRIES {
+ match self.rx.try_recv() {
+ Ok(msg) => {
+ let _ = writer.write_all(&msg.json_line);
+ pending += 1;
+ }
+ Err(_) => break,
+ }
+ }
+
+ if pending >= FLUSH_INTERVAL_ENTRIES || last_flush.elapsed() >= FLUSH_INTERVAL {
+ let _ = writer.flush();
+ pending = 0;
+ last_flush = Instant::now();
+ }
+ }
+
+ // Final flush + sync
+ let _ = writer.flush();
+ if let Ok(file) = writer.into_inner() {
+ let _ = file.sync_all();
+ }
+
+ Ok(())
+ }
+}
diff --git a/core/chaos/src/violation.rs b/core/chaos/src/violation.rs
new file mode 100644
index 000000000..cde8f1b1a
--- /dev/null
+++ b/core/chaos/src/violation.rs
@@ -0,0 +1,34 @@
+/* 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 serde::Serialize;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ViolationCategory {
+ /// Runtime invariant broken (offset regression, server bug, phantom read)
+ Invariant,
+ /// Shadow state diverges from server state for stable resources
+ ShadowMismatch,
+ /// Two independent clients see inconsistent state
+ CrossClient,
+ /// Consumer group assignment or delivery invariant broken
+ ConsumerGroup,
+ /// Infrastructure failure during verification (connect/fetch/task panics)
+ Infrastructure,
+}
diff --git a/core/chaos/src/watermark.rs b/core/chaos/src/watermark.rs
new file mode 100644
index 000000000..b4e6070d1
--- /dev/null
+++ b/core/chaos/src/watermark.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.
+ */
+
+//! Per-partition send watermarks and purge history.
+//!
+//! Lives in a separate `Mutex` from `ShadowState` to avoid adding write-lock
+//! contention to the hot-path `SendMessages` ops. Workers update watermarks
+//! after successful sends - no shadow write-lock needed.
+
+use ahash::AHashMap;
+
+/// stream → topic → partition → messages_sent
+type SendMap = AHashMap<String, AHashMap<String, AHashMap<u32, u64>>>;
+
+pub struct WatermarkState {
+ /// Messages sent per partition since last reset (purge/delete).
+ send: SendMap,
+ /// Permanent purge record: (stream, topic) where topic=None means stream-level purge.
+ purge_history: Vec<(String, Option<String>)>,
+}
+
+impl WatermarkState {
+ pub fn new() -> Self {
+ Self {
+ send: AHashMap::new(),
+ purge_history: Vec::new(),
+ }
+ }
+
+ pub fn record_send(&mut self, stream: &str, topic: &str, partition: u32, count: u32) {
+ *self
+ .send
+ .entry(stream.to_owned())
+ .or_default()
+ .entry(topic.to_owned())
+ .or_default()
+ .entry(partition)
+ .or_default() += count as u64;
+ }
+
+ pub fn send_watermark(&self, stream: &str, topic: &str, partition: u32) -> u64 {
+ self.send
+ .get(stream)
+ .and_then(|t| t.get(topic))
+ .and_then(|p| p.get(&partition))
+ .copied()
+ .unwrap_or(0)
+ }
+
+ /// Reset all partition watermarks for a stream (on delete or stream-level purge).
+ pub fn reset_for_stream(&mut self, stream: &str) {
+ self.send.remove(stream);
+ }
+
+ /// Reset all partition watermarks for a specific topic.
+ pub fn reset_for_topic(&mut self, stream: &str, topic: &str) {
+ if let Some(topics) = self.send.get_mut(stream) {
+ topics.remove(topic);
+ }
+ }
+
+ pub fn record_purge(&mut self, stream: String, topic: Option<String>) {
+ self.purge_history.push((stream, topic));
+ }
+
+ pub fn purge_history(&self) -> &[(String, Option<String>)] {
+ &self.purge_history
+ }
+}
diff --git a/core/chaos/src/worker.rs b/core/chaos/src/worker.rs
new file mode 100644
index 000000000..1aa5adbf2
--- /dev/null
+++ b/core/chaos/src/worker.rs
@@ -0,0 +1,394 @@
+/* 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 crate::error::LabError;
+use crate::in_flight::InFlightOps;
+use crate::invariants::{self, InvariantViolation, PartitionKey, TraceAnchor};
+use crate::ops::{ErrorClass, Op, OpKind, OpOutcome};
+use crate::payload::PayloadGenerator;
+use crate::scenarios::{NamespacePrefixes, OriginScope, RateModel, TargetNamespace};
+use crate::shadow::ShadowState;
+use crate::trace::TraceSender;
+use crate::violation::ViolationCategory;
+use crate::watermark::WatermarkState;
+use iggy::prelude::IggyClient;
+use rand::distr::weighted::WeightedIndex;
+use rand::prelude::Distribution;
+use rand::rngs::StdRng;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+use tokio::sync::{Mutex, RwLock};
+use tracing::{debug, error, warn};
+
+const OP_EXECUTE_TIMEOUT: Duration = Duration::from_secs(30);
+
+pub struct WorkerConfig {
+ pub msg_size: u32,
+ pub messages_per_batch: u32,
+ pub prefixes: NamespacePrefixes,
+ pub namespace: TargetNamespace,
+ pub fail_fast: bool,
+ pub ops_target: Option<u64>,
+ pub scope: OriginScope,
+ pub rate: RateModel,
+ pub allow_destructive: bool,
+ pub consumer_id: u32,
+ pub lane: &'static str,
+}
+
+pub struct WorkerInit {
+ pub id: u32,
+ pub client: IggyClient,
+ pub rng: StdRng,
+ pub shared_state: Arc<RwLock<ShadowState>>,
+ pub watermarks: Arc<Mutex<WatermarkState>>,
+ pub in_flight: Arc<InFlightOps>,
+ pub trace_sender: Option<TraceSender>,
+ pub seq_counter: Arc<AtomicU64>,
+ pub ops_counter: Arc<AtomicU64>,
+ pub start: Instant,
+}
+
+struct Worker {
+ id: u32,
+ client: IggyClient,
+ rng: StdRng,
+ payload_gen: PayloadGenerator,
+ shared_state: Arc<RwLock<ShadowState>>,
+ watermarks: Arc<Mutex<WatermarkState>>,
+ in_flight: Arc<InFlightOps>,
+ trace_sender: Option<TraceSender>,
+ op_kinds: Vec<OpKind>,
+ weighted_dist: WeightedIndex<f64>,
+ config: WorkerConfig,
+ seq_counter: Arc<AtomicU64>,
+ ops_counter: Arc<AtomicU64>,
+ start: Instant,
+}
+
+pub enum WorkerResult {
+ Ok { ops_executed: u64 },
+ ServerBug(InvariantViolation),
+}
+
+impl Worker {
+ fn try_new(
+ init: WorkerInit,
+ op_weights: &[(OpKind, f64)],
+ config: WorkerConfig,
+ ) -> Option<Self> {
+ let effective: Vec<_> = op_weights
+ .iter()
+ .filter(|(kind, _)| {
+ let scope_ok = config.scope != OriginScope::SetupOnly || !kind.is_creative();
+ let destr_ok = config.allow_destructive || !kind.is_destructive();
+ scope_ok && destr_ok
+ })
+ .cloned()
+ .collect();
+ let op_kinds: Vec<OpKind> = effective.iter().map(|(k, _)| *k).collect();
+ let weights: Vec<f64> = effective.iter().map(|(_, w)| *w).collect();
+ let weighted_dist = WeightedIndex::new(&weights).ok()?;
+
+ let payload_gen = PayloadGenerator::new(init.id, config.msg_size);
+
+ Some(Self {
+ id: init.id,
+ client: init.client,
+ rng: init.rng,
+ payload_gen,
+ shared_state: init.shared_state,
+ watermarks: init.watermarks,
+ in_flight: init.in_flight,
+ trace_sender: init.trace_sender,
+ op_kinds,
+ weighted_dist,
+ config,
+ seq_counter: init.seq_counter,
+ ops_counter: init.ops_counter,
+ start: init.start,
+ })
+ }
+
+ async fn run(mut self, stop: Arc<AtomicBool>) -> WorkerResult {
+ let mut last_offsets: HashMap<PartitionKey, u64> = HashMap::new();
+ let mut ops_executed: u64 = 0;
+
+ let op_interval = match self.config.rate {
+ RateModel::TargetOpsPerSec(target) => {
+ Some(Duration::from_secs_f64(1.0 / target as f64))
+ }
+ _ => None,
+ };
+ let mut next_deadline = tokio::time::Instant::now();
+
+ while !stop.load(Ordering::Relaxed) && !self.ops_limit_reached() {
+ let kind = self.pick_op_kind();
+
+ let op = {
+ let state = self.shared_state.read().await;
+ Op::generate(
+ kind,
+ &state,
+ &mut self.rng,
+ &self.config.prefixes,
+ self.config.namespace,
+ self.config.messages_per_batch,
+ self.config.scope,
+ self.config.consumer_id,
+ )
+ };
+
+ let Some(op) = op else {
+ continue;
+ };
+
+ let seq = self.seq_counter.fetch_add(1, Ordering::Relaxed);
+ self.trace_intent(seq, &op);
+
+ let flight_key = op.in_flight_key();
+ if let Some(ref key) = flight_key {
+ self.in_flight.register(key);
+ }
+
+ let start = Instant::now();
+ let exec_result = tokio::select! {
+ result = op.execute(
+ &self.client,
+ &mut self.payload_gen,
+ &self.config.prefixes,
+ ) => result,
+ _ = tokio::time::sleep(OP_EXECUTE_TIMEOUT) => {
+ warn!(
+ "Worker {}: op {} timed out after {OP_EXECUTE_TIMEOUT:?}",
+ self.id, op.kind_tag()
+ );
+ Err(LabError::Timeout {
+ context: format!("op {} execution", op.kind_tag()),
+ })
+ }
+ };
+ let latency = start.elapsed();
+
+ // Lock strategy: read-lock for error classification (no mutation),
+ // write-lock for successful shadow updates and convergence after
+ // ExpectedConcurrent errors.
+ let outcome = match exec_result {
+ Ok(detail) => {
+ let outcome = OpOutcome::success_from(detail);
+ if op.mutates_shadow() {
+ let mut state = self.shared_state.write().await;
+ op.update_shadow(&mut state, &outcome);
+ }
+ outcome
+ }
+ Err(ref e) => {
+ let class = {
+ let state = self.shared_state.read().await;
+ match e {
+ LabError::Iggy(ie) => op.classify_error(ie, &state, &self.in_flight),
+ LabError::Timeout { .. } => ErrorClass::Transient,
+ LabError::PrefixViolation(_) | LabError::InvalidIdentifier(_) => {
+ ErrorClass::ServerBug
+ }
+ LabError::Io { .. } | LabError::TraceParse { .. } => {
+ ErrorClass::ServerBug
+ }
+ }
+ };
+
+ if class == ErrorClass::ExpectedConcurrent
+ && op.mutates_shadow()
+ && let LabError::Iggy(ie) = e
+ {
+ let mut state = self.shared_state.write().await;
+ op.converge_shadow(ie, &mut state);
+ }
+
+ OpOutcome::error(class, e.to_string(), op.context())
+ }
+ };
+
+ if let Some(ref key) = flight_key {
+ self.in_flight.deregister(key);
+ }
+
+ if outcome.is_success() {
+ self.update_watermarks(&op).await;
+ }
+
+ self.trace_outcome(seq, &outcome, latency);
+
+ debug!(
+ worker = self.id,
+ op = op.kind_tag(),
+ result = outcome.result_tag(),
+ latency_us = latency.as_micros() as u64,
+ );
+
+ // Invariant checks (after shadow update so purge records are visible)
+ {
+ let state = self.shared_state.read().await;
+ if let Err(violation) =
+ invariants::check_offset_monotonicity(&mut last_offsets, &op, &outcome, &state)
+ {
+ let violation = violation.with_anchor(self.trace_anchor(seq));
+ error!(worker = self.id, violation = ?violation, "Offset monotonicity violated");
+ if self.config.fail_fast {
+ return WorkerResult::ServerBug(violation);
+ }
+ }
+ }
+
+ if let Some(result) = self.check_fail_fast(&outcome, &op, seq) {
+ return result;
+ }
+
+ ops_executed += 1;
+ self.ops_counter.fetch_add(1, Ordering::Relaxed);
+
+ if let Some(interval) = op_interval {
+ next_deadline += interval;
+ tokio::time::sleep_until(next_deadline).await;
+ }
+ }
+
+ WorkerResult::Ok { ops_executed }
+ }
+
+ fn trace_intent(&self, seq: u64, op: &Op) {
+ if let Some(ts) = &self.trace_sender {
+ ts.write_intent(seq, op);
+ }
+ }
+
+ fn trace_outcome(&self, seq: u64, outcome: &OpOutcome, latency: Duration) {
+ if let Some(ts) = &self.trace_sender {
+ ts.write_outcome(seq, outcome, latency);
+ }
+ }
+
+ /// Check error classification and return early on server bugs (fail-fast mode)
+ /// or log transient/expected errors.
+ fn check_fail_fast(&self, outcome: &OpOutcome, op: &Op, seq: u64) -> Option<WorkerResult> {
+ match outcome.error_class() {
+ Some(ErrorClass::ServerBug) => {
+ let error = match outcome {
+ OpOutcome::Error { error, .. } => error.as_str(),
+ _ => unreachable!(),
+ };
+ error!(worker = self.id, "Server bug: {error}");
+ if self.config.fail_fast {
+ return Some(WorkerResult::ServerBug(
+ InvariantViolation {
+ category: ViolationCategory::Invariant,
+ kind: "server_bug",
+ description: error.to_owned(),
+ context: format!("op={:?}", op),
+ trace_anchor: None,
+ }
+ .with_anchor(self.trace_anchor(seq)),
+ ));
+ }
+ }
+ Some(ErrorClass::Transient) => {
+ let error = match outcome {
+ OpOutcome::Error { error, .. } => error.as_str(),
+ _ => unreachable!(),
+ };
+ warn!(worker = self.id, "Transient error: {error}");
+ }
+ Some(ErrorClass::ExpectedConcurrent) => {
+ debug!(
+ worker = self.id,
+ op = op.kind_tag(),
+ "Expected concurrent error"
+ );
+ }
+ None => {}
+ }
+ None
+ }
+
+ async fn update_watermarks(&self, op: &Op) {
+ let mut wm = self.watermarks.lock().await;
+ match op {
+ Op::SendMessages {
+ stream,
+ topic,
+ partition,
+ count,
+ } => {
+ wm.record_send(stream, topic, *partition, *count);
+ }
+ Op::PurgeStream { name } => {
+ wm.record_purge(name.clone(), None);
+ wm.reset_for_stream(name);
+ }
+ Op::PurgeTopic { stream, topic } => {
+ wm.record_purge(stream.clone(), Some(topic.clone()));
+ wm.reset_for_topic(stream, topic);
+ }
+ Op::DeleteStream { name } => {
+ wm.reset_for_stream(name);
+ }
+ Op::DeleteTopic { stream, topic } => {
+ wm.reset_for_topic(stream, topic);
+ }
+ _ => {}
+ }
+ }
+
+ fn ops_limit_reached(&self) -> bool {
+ self.config
+ .ops_target
+ .is_some_and(|target| self.ops_counter.load(Ordering::Relaxed) >= target)
+ }
+
+ fn pick_op_kind(&mut self) -> OpKind {
+ let idx = self.weighted_dist.sample(&mut self.rng);
+ self.op_kinds[idx]
+ }
+
+ fn trace_anchor(&self, seq: u64) -> TraceAnchor {
+ TraceAnchor {
+ seq,
+ t_us: self.start.elapsed().as_micros() as u64,
+ worker: self.id,
+ lane: self.config.lane,
+ }
+ }
+}
+
+pub async fn run_worker(
+ init: WorkerInit,
+ op_weights: &[(OpKind, f64)],
+ config: WorkerConfig,
+ stop: Arc<AtomicBool>,
+) -> WorkerResult {
+ let id = init.id;
+ match Worker::try_new(init, op_weights, config) {
+ Some(w) => w.run(stop).await,
+ None => {
+ warn!("Worker {id} has no eligible ops after scope/destructive filtering, exiting");
+ WorkerResult::Ok { ops_executed: 0 }
+ }
+ }
+}