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/02/14 11:13:32 UTC
(iggy) branch lab updated (3e8290a14 -> 886f4c5c0)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch lab
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard 3e8290a14 lab
new 886f4c5c0 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 (3e8290a14)
\
N -- N -- N refs/heads/lab (886f4c5c0)
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:
Cargo.lock | 3 +-
core/lab/Cargo.toml | 3 +-
core/lab/src/args.rs | 17 +-
core/lab/src/client.rs | 30 ++
.../continuous/tcp_server.rs => lab/src/error.rs} | 40 +-
core/lab/src/eventually.rs | 130 +++++++
core/lab/src/invariants.rs | 94 ++---
core/lab/src/main.rs | 7 +-
core/lab/src/ops/classify.rs | 134 ++++++-
core/lab/src/ops/execute.rs | 322 ++++++++++------
core/lab/src/ops/generate.rs | 144 +++++--
core/lab/src/ops/mod.rs | 263 +++++++++++--
core/lab/src/ops/precondition.rs | 28 ++
core/lab/src/ops/shadow_update.rs | 21 +-
core/lab/src/replay.rs | 94 +++--
core/lab/src/report.rs | 47 ++-
core/lab/src/runner.rs | 302 +++++++--------
core/lab/src/scenarios/concurrent_crud.rs | 75 +++-
core/lab/src/scenarios/mixed_workload.rs | 65 +++-
core/lab/src/scenarios/mod.rs | 146 ++++++-
core/lab/src/scenarios/segment_rotation.rs | 53 ++-
core/lab/src/scenarios/stale_consumers.rs | 421 +++++++++++++++++++++
core/lab/src/shadow.rs | 124 +++---
core/lab/src/trace.rs | 32 +-
core/lab/src/worker.rs | 204 ++++++----
25 files changed, 2159 insertions(+), 640 deletions(-)
copy core/{server/src/shard/tasks/continuous/tcp_server.rs => lab/src/error.rs} (58%)
create mode 100644 core/lab/src/eventually.rs
create mode 100644 core/lab/src/scenarios/stale_consumers.rs
(iggy) 01/01: lab
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch lab
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 886f4c5c02f26aa2970502fd40e5e2530809edd1
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Fri Feb 13 21:13:59 2026 +0100
lab
---
Cargo.lock | 19 ++
Cargo.toml | 1 +
DEPENDENCIES.md | 1 +
core/lab/Cargo.toml | 44 +++
core/lab/src/args.rs | 182 +++++++++++++
core/lab/src/client.rs | 96 +++++++
core/lab/src/error.rs | 47 ++++
core/lab/src/eventually.rs | 130 +++++++++
core/lab/src/invariants.rs | 273 +++++++++++++++++++
core/lab/src/main.rs | 97 +++++++
core/lab/src/ops/classify.rs | 221 +++++++++++++++
core/lab/src/ops/execute.rs | 355 ++++++++++++++++++++++++
core/lab/src/ops/generate.rs | 281 +++++++++++++++++++
core/lab/src/ops/mod.rs | 408 ++++++++++++++++++++++++++++
core/lab/src/ops/precondition.rs | 126 +++++++++
core/lab/src/ops/shadow_update.rs | 102 +++++++
core/lab/src/replay.rs | 336 +++++++++++++++++++++++
core/lab/src/report.rs | 136 ++++++++++
core/lab/src/runner.rs | 366 +++++++++++++++++++++++++
core/lab/src/safe_name.rs | 114 ++++++++
core/lab/src/scenarios/concurrent_crud.rs | 105 +++++++
core/lab/src/scenarios/mixed_workload.rs | 90 ++++++
core/lab/src/scenarios/mod.rs | 197 ++++++++++++++
core/lab/src/scenarios/segment_rotation.rs | 89 ++++++
core/lab/src/scenarios/stale_consumers.rs | 421 +++++++++++++++++++++++++++++
core/lab/src/shadow.rs | 276 +++++++++++++++++++
core/lab/src/trace.rs | 129 +++++++++
core/lab/src/worker.rs | 296 ++++++++++++++++++++
28 files changed, 4938 insertions(+)
diff --git a/Cargo.lock b/Cargo.lock
index d7ffd97b4..6450dd97a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4621,6 +4621,25 @@ dependencies = [
"tracing-subscriber",
]
+[[package]]
+name = "iggy-lab"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "chrono",
+ "clap",
+ "iggy",
+ "iggy_common",
+ "rand 0.10.0",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tokio",
+ "tracing",
+ "tracing-subscriber",
+]
+
[[package]]
name = "iggy-mcp"
version = "0.2.3-edge.1"
diff --git a/Cargo.toml b/Cargo.toml
index e306352c1..0d4dbc4fe 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -45,6 +45,7 @@ members = [
"core/harness_derive",
"core/integration",
"core/journal",
+ "core/lab",
"core/message_bus",
"core/metadata",
"core/partitions",
diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md
index a9e14e6d4..94d445ae9 100644
--- a/DEPENDENCIES.md
+++ b/DEPENDENCIES.md
@@ -393,6 +393,7 @@ iggy-bench: 0.3.3-edge.1, "Apache-2.0",
iggy-bench-dashboard-server: 0.5.1-edge.1, "Apache-2.0",
iggy-cli: 0.10.3-edge.1, "Apache-2.0",
iggy-connectors: 0.2.4-edge.1, "Apache-2.0",
+iggy-lab: 0.1.0, "Apache-2.0",
iggy-mcp: 0.2.3-edge.1, "Apache-2.0",
iggy_binary_protocol: 0.8.3-edge.1, "Apache-2.0",
iggy_common: 0.8.3-edge.1, "Apache-2.0",
diff --git a/core/lab/Cargo.toml b/core/lab/Cargo.toml
new file mode 100644
index 000000000..78f921ca0
--- /dev/null
+++ b/core/lab/Cargo.toml
@@ -0,0 +1,44 @@
+# 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-lab"
+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-lab"
+path = "src/main.rs"
+
+[dependencies]
+async-trait = { workspace = true }
+bytes = { workspace = true }
+chrono = { workspace = true }
+clap = { 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/lab/src/args.rs b/core/lab/src/args.rs
new file mode 100644
index 000000000..301d72865
--- /dev/null
+++ b/core/lab/src/args.rs
@@ -0,0 +1,182 @@
+/* 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-lab",
+ 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,
+ },
+}
+
+#[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
+ #[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,
+}
+
+/// 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,
+}
+
+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"),
+ }
+ }
+}
diff --git a/core/lab/src/client.rs b/core/lab/src/client.rs
new file mode 100644
index 000000000..b856347f1
--- /dev/null
+++ b/core/lab/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/lab/src/error.rs b/core/lab/src/error.rs
new file mode 100644
index 000000000..ef0ddb670
--- /dev/null
+++ b/core/lab/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/lab/src/eventually.rs b/core/lab/src/eventually.rs
new file mode 100644
index 000000000..2d24fff47
--- /dev/null
+++ b/core/lab/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/lab/src/invariants.rs b/core/lab/src/invariants.rs
new file mode 100644
index 000000000..d9e5c6c89
--- /dev/null
+++ b/core/lab/src/invariants.rs
@@ -0,0 +1,273 @@
+/* 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, OpOutcome};
+use crate::scenarios::{Namespace, NamespacePrefixes};
+use crate::shadow::ShadowState;
+use iggy::prelude::*;
+use serde::Serialize;
+use std::collections::HashMap;
+
+pub type PartitionKey = (String, String, u32);
+
+#[derive(Debug, Clone, Serialize)]
+pub struct InvariantViolation {
+ pub kind: &'static str,
+ pub description: String,
+ pub context: String,
+}
+
+/// 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(detail),
+ } => match parse_poll_offset(detail) {
+ Some(o) => o,
+ None => return Ok(()),
+ },
+ _ => 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 {
+ kind: "offset_regression",
+ description: format!("Offset went backwards: {last} -> {offset}"),
+ context: format!("stream={stream}, topic={topic}, partition={partition}"),
+ });
+ }
+ 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 {
+ kind: "post_run_fetch_failed",
+ description: format!("Failed to get streams: {e}"),
+ context: String::new(),
+ });
+ 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 {
+ kind: "missing_stable_stream",
+ description: format!("Shadow has stable stream '{name}' but server does not"),
+ context: String::new(),
+ });
+ }
+ }
+
+ // 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 {
+ kind: "extra_stable_stream",
+ description: format!("Server has stable stream '{name}' not in shadow"),
+ context: String::new(),
+ });
+ }
+ // 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 {
+ kind: "stream_details_fetch_failed",
+ description: format!("Failed to get details for stream '{name}'"),
+ context: String::new(),
+ });
+ 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 {
+ kind: "missing_stable_topic",
+ description: format!(
+ "Shadow has topic '{topic_name}' in stable stream '{name}' but server does not"
+ ),
+ context: String::new(),
+ });
+ }
+ }
+
+ for &topic_name in server_topics.keys() {
+ if !shadow_stream.topics.contains_key(topic_name) && is_stable {
+ violations.push(InvariantViolation {
+ kind: "extra_stable_topic",
+ description: format!(
+ "Server has topic '{topic_name}' in stable stream '{name}' not in shadow"
+ ),
+ context: String::new(),
+ });
+ }
+ }
+ }
+
+ 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 {
+ kind: "cross_client_connect_failed",
+ description: format!("Client A: {e}"),
+ context: String::new(),
+ });
+ return violations;
+ }
+ };
+ let client_b = match create_client(addr, transport).await {
+ Ok(c) => c,
+ Err(e) => {
+ violations.push(InvariantViolation {
+ kind: "cross_client_connect_failed",
+ description: format!("Client B: {e}"),
+ context: String::new(),
+ });
+ 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 {
+ kind: "cross_client_fetch_failed",
+ description: format!("Failed to get streams: {e}"),
+ context: String::new(),
+ });
+ 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 {
+ 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<_>>()
+ ),
+ });
+ }
+
+ violations
+}
+
+fn parse_poll_offset(detail: &str) -> Option<u64> {
+ // detail format: "received=N, offset=M"
+ detail
+ .split(", ")
+ .find_map(|part| part.strip_prefix("offset="))
+ .and_then(|v| v.parse().ok())
+}
diff --git a/core/lab/src/main.rs b/core/lab/src/main.rs
new file mode 100644
index 000000000..bb8f1255d
--- /dev/null
+++ b/core/lab/src/main.rs
@@ -0,0 +1,97 @@
+/* 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 invariants;
+mod ops;
+mod replay;
+mod report;
+mod runner;
+mod safe_name;
+mod scenarios;
+mod shadow;
+mod trace;
+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::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/lab/src/ops/classify.rs b/core/lab/src/ops/classify.rs
new file mode 100644
index 000000000..4cffe1645
--- /dev/null
+++ b/core/lab/src/ops/classify.rs
@@ -0,0 +1,221 @@
+/* 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::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(_, _) => 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 knowledge to distinguish
+ /// between concurrent races (benign), server bugs, and transient failures.
+ pub fn classify_error(&self, err: &IggyError, state: &ShadowState) -> 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(_, _) => ErrorClass::Transient,
+
+ IggyError::StreamNameNotFound(name) => {
+ if state.was_recently_deleted(name) || !state.stream_exists(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))
+ {
+ 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))
+ {
+ 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),
+ }
+ }
+}
diff --git a/core/lab/src/ops/execute.rs b/core/lab/src/ops/execute.rs
new file mode 100644
index 000000000..af4b76db5
--- /dev/null
+++ b/core/lab/src/ops/execute.rs
@@ -0,0 +1,355 @@
+/* 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;
+use crate::error::LabError;
+use crate::scenarios::NamespacePrefixes;
+use bytes::Bytes;
+use iggy::prelude::*;
+use iggy_common::IggyError;
+use rand::{Rng, RngExt};
+
+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,
+ msg_size: u32,
+ rng: &mut (impl Rng + ?Sized),
+ prefixes: &NamespacePrefixes,
+ ) -> Result<Option<String>, 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(format!("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(format!("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 mut payload = vec![0u8; msg_size as usize];
+ rng.fill(&mut payload[..]);
+ messages.push(
+ IggyMessage::builder()
+ .payload(Bytes::from(payload))
+ .build()
+ .unwrap(),
+ );
+ }
+ client
+ .send_messages(&stream_id, &topic_id, &partitioning, &mut messages)
+ .await?;
+ Ok(Some(format!("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?;
+ Ok(Some(format!(
+ "received={}, offset={}",
+ polled.messages.len(),
+ 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(format!("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(format!(
+ "members={}, partitions={}",
+ details.members_count, 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?;
+ Ok(Some(format!(
+ "received={}, partition_id={}, offset={}",
+ polled.messages.len(),
+ polled.partition_id,
+ 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(format!("count={}", streams.len())))
+ }
+
+ Op::GetStreamDetails { stream } => {
+ let stream_id = ident(stream)?;
+ match client.get_stream(&stream_id).await? {
+ Some(details) => Ok(Some(format!("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(format!("partitions={}", details.partitions_count))),
+ None => Err(IggyError::TopicNameNotFound(stream.clone(), topic.clone()).into()),
+ }
+ }
+
+ Op::GetStats => {
+ let stats = client.get_stats().await?;
+ Ok(Some(format!(
+ "streams={} topics={} partitions={}",
+ stats.streams_count, stats.topics_count, stats.partitions_count
+ )))
+ }
+ }
+ }
+}
diff --git a/core/lab/src/ops/generate.rs b/core/lab/src/ops/generate.rs
new file mode 100644
index 000000000..b26948c6c
--- /dev/null
+++ b/core/lab/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::{LaneNamespace, NamespacePrefixes, NamespaceScope};
+use crate::shadow::ShadowState;
+use rand::{Rng, RngExt};
+
+const DEFAULT_PARTITIONS: u32 = 3;
+
+fn pick_random_group(
+ state: &ShadowState,
+ rng: &mut impl Rng,
+ scope: NamespaceScope,
+ namespace: LaneNamespace,
+) -> 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: LaneNamespace,
+ messages_per_batch: u32,
+ scope: NamespaceScope,
+ consumer_id: u32,
+ ) -> Option<Self> {
+ let effective_namespace = if kind.is_resource_destructive() {
+ LaneNamespace::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(1..=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(1..=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(1..=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(1..=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/lab/src/ops/mod.rs b/core/lab/src/ops/mod.rs
new file mode 100644
index 000000000..7d0c200bd
--- /dev/null
+++ b/core/lab/src/ops/mod.rs
@@ -0,0 +1,408 @@
+/* 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 precondition;
+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 = "status")]
+pub enum OpOutcome {
+ Success {
+ detail: Option<String>,
+ },
+ Error {
+ class: ErrorClass,
+ error: String,
+ context: String,
+ },
+}
+
+impl OpOutcome {
+ pub fn success_from(detail: Option<String>) -> 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(),
+ }
+ }
+
+ /// 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/lab/src/ops/precondition.rs b/core/lab/src/ops/precondition.rs
new file mode 100644
index 000000000..b36271d15
--- /dev/null
+++ b/core/lab/src/ops/precondition.rs
@@ -0,0 +1,126 @@
+/* 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;
+use crate::shadow::ShadowState;
+
+#[allow(dead_code)]
+impl Op {
+ /// Check whether this op's preconditions are met according to our shadow state.
+ /// A false return means the op would definitely fail, so the worker should skip it.
+ pub fn precondition_met(&self, state: &ShadowState) -> bool {
+ match self {
+ Op::CreateStream { name } => !state.stream_exists(name),
+ Op::DeleteStream { name } | Op::PurgeStream { name } => state.stream_exists(name),
+
+ Op::CreateTopic { stream, name, .. } => {
+ state.stream_exists(stream) && !state.topic_exists(stream, name)
+ }
+ Op::DeleteTopic { stream, topic } | Op::PurgeTopic { stream, topic } => {
+ state.topic_exists(stream, topic)
+ }
+
+ Op::SendMessages {
+ stream,
+ topic,
+ partition,
+ ..
+ }
+ | Op::PollMessages {
+ stream,
+ topic,
+ partition,
+ ..
+ } => state
+ .get_topic(stream, topic)
+ .is_some_and(|t| *partition >= 1 && *partition <= t.partitions),
+
+ Op::DeleteSegments {
+ stream,
+ topic,
+ partition,
+ ..
+ } => state
+ .get_topic(stream, topic)
+ .is_some_and(|t| *partition >= 1 && *partition <= t.partitions),
+
+ Op::CreatePartitions { stream, topic, .. } => state.topic_exists(stream, topic),
+
+ Op::DeletePartitions {
+ stream,
+ topic,
+ count,
+ } => state
+ .get_topic(stream, topic)
+ .is_some_and(|t| t.partitions > *count),
+
+ Op::CreateConsumerGroup {
+ stream,
+ topic,
+ name,
+ } => state
+ .get_topic(stream, topic)
+ .is_some_and(|t| !t.consumer_groups.contains(name)),
+
+ Op::DeleteConsumerGroup {
+ stream,
+ topic,
+ name,
+ } => state
+ .get_topic(stream, topic)
+ .is_some_and(|t| t.consumer_groups.contains(name)),
+
+ Op::JoinConsumerGroup {
+ stream,
+ topic,
+ group,
+ }
+ | Op::LeaveConsumerGroup {
+ stream,
+ topic,
+ group,
+ }
+ | Op::GetConsumerGroup {
+ stream,
+ topic,
+ group,
+ }
+ | Op::PollGroupMessages {
+ stream,
+ topic,
+ group,
+ ..
+ } => state
+ .get_topic(stream, topic)
+ .is_some_and(|t| t.consumer_groups.contains(group)),
+
+ Op::StoreConsumerOffset {
+ stream,
+ topic,
+ partition,
+ ..
+ } => state
+ .get_topic(stream, topic)
+ .is_some_and(|t| *partition >= 1 && *partition <= t.partitions),
+
+ Op::GetStreams | Op::GetStats => true,
+ Op::GetStreamDetails { stream } => state.stream_exists(stream),
+ Op::GetTopicDetails { stream, topic } => state.topic_exists(stream, topic),
+ }
+ }
+}
diff --git a/core/lab/src/ops/shadow_update.rs b/core/lab/src/ops/shadow_update.rs
new file mode 100644
index 000000000..f4829b534
--- /dev/null
+++ b/core/lab/src/ops/shadow_update.rs
@@ -0,0 +1,102 @@
+/* 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};
+
+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();
+ }
+}
diff --git a/core/lab/src/replay.rs b/core/lab/src/replay.rs
new file mode 100644
index 000000000..fc115b003
--- /dev/null
+++ b/core/lab/src/replay.rs
@@ -0,0 +1,336 @@
+/* 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, OpOutcome};
+use crate::scenarios::NamespacePrefixes;
+use rand::SeedableRng;
+use rand::rngs::StdRng;
+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<String>,
+}
+
+struct ReplayIntent {
+ seq: u64,
+ op: Op,
+}
+
+struct OriginalOutcome {
+ result: String,
+ detail: Option<String>,
+}
+
+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-lab 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;
+ }
+
+ // Deterministic RNG for message payloads (content is irrelevant to correctness)
+ let mut rng = StdRng::seed_from_u64(0);
+ 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, self.args.message_size, &mut rng, &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.as_deref().unwrap_or(""),
+ "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;
+ let entries = std::fs::read_dir(trace_dir).map_err(|e| LabError::Io {
+ context: format!("reading trace dir {}", trace_dir.display()),
+ source: e,
+ })?;
+
+ let mut intents: Vec<ReplayIntent> = Vec::new();
+ let mut outcomes: HashMap<u64, OriginalOutcome> = HashMap::new();
+
+ 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") {
+ continue;
+ }
+
+ 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/lab/src/report.rs b/core/lab/src/report.rs
new file mode 100644
index 000000000..1e1789e12
--- /dev/null
+++ b/core/lab/src/report.rs
@@ -0,0 +1,136 @@
+/* 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,
+}
+
+#[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,
+}
+
+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,
+ };
+ 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/lab/src/runner.rs b/core/lab/src/runner.rs
new file mode 100644
index 000000000..17ee43f01
--- /dev/null
+++ b/core/lab/src/runner.rs
@@ -0,0 +1,366 @@
+/* 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::invariants::InvariantViolation;
+use crate::report::ArtifactBundle;
+use crate::scenarios::{self, CleanupPolicy, NamespacePrefixes, VerifyContext};
+use crate::shadow::{ResourceOrigin, ShadowState};
+use crate::trace::WorkerTraceWriter;
+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("lab-bundles").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-lab ===");
+ 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 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;
+
+ 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(seed.wrapping_add(worker_id as u64));
+ let client = create_client(&self.args.server_address, self.args.transport).await?;
+ let trace_writer = Some(
+ WorkerTraceWriter::new(&bundle_dir, worker_id, start_instant)
+ .expect("Failed to create trace writer"),
+ );
+ 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,
+ destructive: lane.destructive,
+ consumer_id: worker_id + 1,
+ };
+ let init = WorkerInit {
+ id: worker_id,
+ client,
+ rng: worker_rng,
+ shared_state: Arc::clone(&shared_state),
+ trace_writer,
+ seq_counter: Arc::clone(&seq_counter),
+ ops_counter: Arc::clone(&ops_counter),
+ };
+ 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}");
+ }
+ }
+ }
+
+ 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 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);
+ }
+
+ 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,
+ };
+ if let Err(e) = bundle.write(&bundle_dir) {
+ error!("Failed to write artifacts: {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;
+ }
+ }
+}
+
+/// 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/lab/src/safe_name.rs b/core/lab/src/safe_name.rs
new file mode 100644
index 000000000..75cef378d
--- /dev/null
+++ b/core/lab/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/lab/src/scenarios/concurrent_crud.rs b/core/lab/src/scenarios/concurrent_crud.rs
new file mode 100644
index 000000000..bb5ab3cb7
--- /dev/null
+++ b/core/lab/src/scenarios/concurrent_crud.rs
@@ -0,0 +1,105 @@
+/* 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::{LaneNamespace, LaneSpec, NamespacePrefixes, NamespaceScope, RateModel, Scenario};
+use crate::ops::OpKind;
+use crate::safe_name::SafeResourceName;
+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..."
+ );
+ for i in 0..SETUP_STREAMS {
+ let stream_name = SafeResourceName::new(&prefixes.stable, &format!("setup-s{i}"));
+ client.create_stream(&stream_name).await?;
+ let stream_id = Identifier::from_str_value(&stream_name).unwrap();
+ for j in 0..TOPICS_PER_STREAM {
+ let topic_name = SafeResourceName::new(&prefixes.base, &format!("setup-t{i}-{j}"));
+ client
+ .create_topic(
+ &stream_id,
+ &topic_name,
+ PARTITIONS_PER_TOPIC,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::Unlimited,
+ )
+ .await?;
+ }
+ }
+ Ok(())
+ }
+
+ fn lanes(&self) -> Vec<LaneSpec> {
+ vec![
+ LaneSpec {
+ label: "control-plane",
+ op_weights: vec![
+ (OpKind::CreateStream, 20.0),
+ (OpKind::DeleteStream, 15.0),
+ (OpKind::PurgeStream, 10.0),
+ (OpKind::CreateTopic, 20.0),
+ (OpKind::DeleteTopic, 15.0),
+ (OpKind::PurgeTopic, 10.0),
+ ],
+ worker_count: None,
+ scope: NamespaceScope::Any,
+ namespace: LaneNamespace::Churn,
+ rate: RateModel::Unbounded,
+ destructive: true,
+ },
+ LaneSpec {
+ label: "data-plane",
+ op_weights: vec![(OpKind::SendMessages, 60.0), (OpKind::PollMessages, 40.0)],
+ worker_count: Some(2),
+ scope: NamespaceScope::SetupOnly,
+ namespace: LaneNamespace::Stable,
+ rate: RateModel::Unbounded,
+ destructive: false,
+ },
+ ]
+ }
+}
diff --git a/core/lab/src/scenarios/mixed_workload.rs b/core/lab/src/scenarios/mixed_workload.rs
new file mode 100644
index 000000000..d18934ee6
--- /dev/null
+++ b/core/lab/src/scenarios/mixed_workload.rs
@@ -0,0 +1,90 @@
+/* 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::{LaneNamespace, LaneSpec, NamespaceScope, RateModel, Scenario};
+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![
+ LaneSpec {
+ label: "data-plane",
+ op_weights: vec![(OpKind::SendMessages, 60.0), (OpKind::PollMessages, 40.0)],
+ worker_count: None,
+ scope: NamespaceScope::Any,
+ namespace: LaneNamespace::Both,
+ rate: RateModel::Unbounded,
+ destructive: false,
+ },
+ LaneSpec {
+ label: "control-plane",
+ op_weights: vec![
+ (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),
+ ],
+ worker_count: Some(4),
+ scope: NamespaceScope::Any,
+ namespace: LaneNamespace::Churn,
+ rate: RateModel::Unbounded,
+ destructive: true,
+ },
+ LaneSpec {
+ label: "admin",
+ op_weights: vec![
+ (OpKind::GetStreams, 40.0),
+ (OpKind::GetStreamDetails, 30.0),
+ (OpKind::GetTopicDetails, 20.0),
+ (OpKind::GetStats, 10.0),
+ ],
+ worker_count: Some(2),
+ scope: NamespaceScope::Any,
+ namespace: LaneNamespace::Both,
+ rate: RateModel::Unbounded,
+ destructive: false,
+ },
+ ]
+ }
+}
diff --git a/core/lab/src/scenarios/mod.rs b/core/lab/src/scenarios/mod.rs
new file mode 100644
index 000000000..97e7abb38
--- /dev/null
+++ b/core/lab/src/scenarios/mod.rs
@@ -0,0 +1,197 @@
+/* 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 mixed_workload;
+pub mod segment_rotation;
+pub mod stale_consumers;
+
+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 NamespaceScope {
+ #[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 LaneNamespace {
+ 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: LaneNamespace) -> &str {
+ match ns {
+ LaneNamespace::Stable => &self.stable,
+ LaneNamespace::Churn | LaneNamespace::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: NamespaceScope,
+ pub namespace: LaneNamespace,
+ pub rate: RateModel,
+ pub 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),
+ }
+}
+
+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(),
+ ),
+ ]
+}
diff --git a/core/lab/src/scenarios/segment_rotation.rs b/core/lab/src/scenarios/segment_rotation.rs
new file mode 100644
index 000000000..f7b7c82bf
--- /dev/null
+++ b/core/lab/src/scenarios/segment_rotation.rs
@@ -0,0 +1,89 @@
+/* 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::{LaneNamespace, LaneSpec, NamespacePrefixes, NamespaceScope, RateModel, Scenario};
+use crate::ops::OpKind;
+use crate::safe_name::SafeResourceName;
+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 stream_name = SafeResourceName::new(&prefixes.stable, "seg-rot-stream");
+ client.create_stream(&stream_name).await?;
+ let stream_id = Identifier::from_str_value(&stream_name).unwrap();
+ let topic_name = SafeResourceName::new(&prefixes.base, "seg-rot-topic");
+ client
+ .create_topic(
+ &stream_id,
+ &topic_name,
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::Unlimited,
+ )
+ .await?;
+ Ok(())
+ }
+
+ fn lanes(&self) -> Vec<LaneSpec> {
+ vec![
+ LaneSpec {
+ label: "data-plane",
+ op_weights: vec![(OpKind::SendMessages, 55.0), (OpKind::PollMessages, 45.0)],
+ worker_count: None,
+ scope: NamespaceScope::SetupOnly,
+ namespace: LaneNamespace::Stable,
+ rate: RateModel::Unbounded,
+ destructive: false,
+ },
+ LaneSpec {
+ label: "segment-cleanup",
+ op_weights: vec![(OpKind::DeleteSegments, 100.0)],
+ worker_count: Some(1),
+ scope: NamespaceScope::SetupOnly,
+ namespace: LaneNamespace::Stable,
+ rate: RateModel::TargetOpsPerSec(2),
+ destructive: true,
+ },
+ ]
+ }
+}
diff --git a/core/lab/src/scenarios/stale_consumers.rs b/core/lab/src/scenarios/stale_consumers.rs
new file mode 100644
index 000000000..14000fe5b
--- /dev/null
+++ b/core/lab/src/scenarios/stale_consumers.rs
@@ -0,0 +1,421 @@
+/* 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, VerifyContext};
+use crate::client::create_stale_client;
+use crate::eventually::{await_members_count, await_rebalance_complete};
+use crate::invariants::InvariantViolation;
+use crate::safe_name::SafeResourceName;
+use async_trait::async_trait;
+use iggy::prelude::*;
+use std::collections::HashMap;
+use std::str::FromStr;
+use std::time::Duration;
+use tokio::task::JoinSet;
+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> {
+ let stream_name = Self::stream_name(prefixes);
+ let topic_name = Self::topic_name(prefixes);
+ let group_name = Self::group_name(prefixes);
+
+ info!(
+ "Setup: creating stream, topic ({PARTITION_COUNT} partitions), consumer group, sentinel message"
+ );
+
+ client.create_stream(&stream_name).await?;
+ let stream_id = Identifier::from_str_value(&stream_name).unwrap();
+
+ client
+ .create_topic(
+ &stream_id,
+ &topic_name,
+ PARTITION_COUNT,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::Unlimited,
+ )
+ .await?;
+
+ let topic_id = Identifier::from_str_value(&topic_name).unwrap();
+ client
+ .create_consumer_group(&stream_id, &topic_id, &group_name)
+ .await?;
+
+ let sentinel = IggyMessage::from_str(SENTINEL_PAYLOAD).unwrap();
+ let mut messages = vec![sentinel];
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(1),
+ &mut messages,
+ )
+ .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();
+
+ let mut violations = Vec::new();
+
+ // --- Step 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 {
+ kind: "stale_client_create_failed",
+ description: format!("Failed to create stale client #{i}: {e}"),
+ context: String::new(),
+ })?;
+ client
+ .join_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .map_err(|e| InvariantViolation {
+ kind: "stale_join_failed",
+ description: format!("Stale client #{i} failed to join group: {e}"),
+ context: String::new(),
+ })?;
+ stale_clients.push(client);
+ }
+
+ // --- Step 2: Confirm stale members joined ---
+ await_members_count(
+ ctx.client,
+ &stream_id,
+ &topic_id,
+ &group_id,
+ STALE_MEMBER_COUNT,
+ EVICTION_TIMEOUT,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ kind: "members_never_joined",
+ description: format!("Expected {STALE_MEMBER_COUNT} members after join, timed out: {e}"),
+ context: String::new(),
+ })?;
+ info!("Verify: {STALE_MEMBER_COUNT} stale members joined");
+
+ // --- Step 2b: Wait for stale members to get partitions assigned ---
+ await_rebalance_complete(
+ ctx.client,
+ &stream_id,
+ &topic_id,
+ &group_id,
+ REBALANCE_TIMEOUT,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ kind: "stale_rebalance_incomplete",
+ description: format!(
+ "Rebalance for stale members did not complete within {REBALANCE_TIMEOUT:?}: {e}"
+ ),
+ context: String::new(),
+ })?;
+ info!("Verify: stale members have partitions assigned, polling without commit");
+
+ // --- Step 2c: Stale members poll WITHOUT committing offsets ---
+ // Exercises the "fetched but uncommitted" path — offsets must remain
+ // available for fresh members after eviction.
+ // With PARTITION_COUNT partitions across STALE_MEMBER_COUNT members (1:1),
+ // every partition is assigned, so the sentinel on partition 1 is reachable.
+ 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 {
+ kind: "stale_poll_failed",
+ description: format!("Stale member #{i} poll failed: {e}"),
+ context: String::new(),
+ })?;
+ if contains_sentinel(&polled) {
+ stale_saw_sentinel = true;
+ info!("Verify: stale member #{i} observed sentinel (not committed)");
+ }
+ }
+ if !stale_saw_sentinel {
+ return Err(InvariantViolation {
+ 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"
+ ),
+ });
+ }
+ info!("Verify: stale poll phase complete, waiting for eviction");
+
+ // --- Step 3: Wait for heartbeat eviction ---
+ // Keep stale clients alive (connected) — the server must evict them based on
+ // missed heartbeats, not connection close.
+ await_members_count(
+ ctx.client,
+ &stream_id,
+ &topic_id,
+ &group_id,
+ 0,
+ EVICTION_TIMEOUT,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ 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(),
+ })?;
+ info!("Verify: all stale members evicted");
+ drop(stale_clients);
+
+ // --- Step 4: Join fresh members ---
+ // Fresh clients always use TCP because heartbeat eviction is a binary
+ // protocol mechanism — HTTP/REST does not participate in heartbeat keepalive.
+ info!("Verify: joining {FRESH_MEMBER_COUNT} fresh members");
+ let mut fresh_clients = Vec::with_capacity(FRESH_MEMBER_COUNT as usize);
+ for i in 0..FRESH_MEMBER_COUNT {
+ let client = crate::client::create_client(ctx.server_address, crate::args::Transport::Tcp)
+ .await
+ .map_err(|e| InvariantViolation {
+ kind: "fresh_client_create_failed",
+ description: format!("Failed to create fresh client #{i}: {e}"),
+ context: String::new(),
+ })?;
+ client
+ .join_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .map_err(|e| InvariantViolation {
+ kind: "fresh_join_failed",
+ description: format!("Fresh client #{i} failed to join group: {e}"),
+ context: String::new(),
+ })?;
+ fresh_clients.push(client);
+ }
+
+ // --- Step 5a: Confirm all fresh members joined ---
+ await_members_count(
+ ctx.client,
+ &stream_id,
+ &topic_id,
+ &group_id,
+ FRESH_MEMBER_COUNT,
+ REBALANCE_TIMEOUT,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ kind: "fresh_members_never_joined",
+ description: format!("Expected {FRESH_MEMBER_COUNT} fresh members, timed out: {e}"),
+ context: String::new(),
+ })?;
+
+ // --- Step 5b: Wait for rebalance ---
+ let details = await_rebalance_complete(
+ ctx.client,
+ &stream_id,
+ &topic_id,
+ &group_id,
+ REBALANCE_TIMEOUT,
+ )
+ .await
+ .map_err(|e| InvariantViolation {
+ kind: "rebalance_incomplete",
+ description: format!("Rebalance did not complete within {REBALANCE_TIMEOUT:?}: {e}"),
+ context: String::new(),
+ })?;
+ info!(
+ "Verify: rebalance complete — {} members, {} partitions",
+ details.members_count, details.partitions_count
+ );
+
+ // --- Step 6: Check partition assignments are unique ---
+ 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 {
+ kind: "duplicate_partition_assignment",
+ description: format!("Partition {partition} assigned to multiple members"),
+ context: format!("member_id={} and member_id={prev_owner}", member.id),
+ });
+ } else {
+ partition_owners.insert(partition, member.id);
+ }
+ }
+ }
+
+ // --- Step 7: Poll concurrently from all fresh members ---
+ // auto_commit=false so each poll is independent of group offset mutation —
+ // prevents one fast commit from masking duplicate delivery.
+ info!("Verify: polling concurrently from {FRESH_MEMBER_COUNT} fresh members");
+ let mut poll_tasks = JoinSet::new();
+ for client in fresh_clients {
+ let s_id = stream_id.clone();
+ let t_id = topic_id.clone();
+ let g_id = 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 mut sentinel_receivers = 0u32;
+ while let Some(result) = poll_tasks.join_next().await {
+ match result {
+ Ok(Ok(polled)) => {
+ if contains_sentinel(&polled) {
+ sentinel_receivers += 1;
+ }
+ }
+ Ok(Err(e)) => {
+ violations.push(InvariantViolation {
+ kind: "poll_failed",
+ description: format!("Fresh member poll failed: {e}"),
+ context: String::new(),
+ });
+ }
+ Err(e) => {
+ violations.push(InvariantViolation {
+ kind: "poll_task_panicked",
+ description: format!("Poll task panicked: {e}"),
+ context: String::new(),
+ });
+ }
+ }
+ }
+
+ // --- Step 8: Assert exactly one member received the sentinel ---
+ if sentinel_receivers == 0 {
+ violations.push(InvariantViolation {
+ kind: "sentinel_lost",
+ description: "No fresh member received the sentinel message".into(),
+ context: String::new(),
+ });
+ } else if sentinel_receivers > 1 {
+ violations.push(InvariantViolation {
+ kind: "duplicate_sentinel_delivery",
+ description: format!(
+ "{sentinel_receivers} members received the sentinel (expected exactly 1)"
+ ),
+ context: String::new(),
+ });
+ }
+
+ Ok(violations)
+}
+
+fn contains_sentinel(polled: &PolledMessages) -> bool {
+ polled
+ .messages
+ .iter()
+ .any(|m| m.payload == SENTINEL_PAYLOAD)
+}
diff --git a/core/lab/src/shadow.rs b/core/lab/src/shadow.rs
new file mode 100644
index 000000000..c4c4e1faa
--- /dev/null
+++ b/core/lab/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::{LaneNamespace, Namespace, NamespacePrefixes, NamespaceScope};
+use rand::{Rng, RngExt};
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
+use std::time::{Duration, Instant};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ResourceOrigin {
+ Setup,
+ Dynamic,
+}
+
+const TOMBSTONE_RETENTION: Duration = Duration::from_secs(10);
+
+pub struct ShadowState {
+ pub streams: BTreeMap<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: BTreeMap<String, ShadowTopic>,
+ pub origin: ResourceOrigin,
+ pub namespace: Namespace,
+}
+
+pub struct ShadowTopic {
+ pub partitions: u32,
+ pub consumer_groups: BTreeSet<String>,
+ pub origin: ResourceOrigin,
+}
+
+impl ShadowState {
+ pub fn new(prefixes: NamespacePrefixes) -> Self {
+ Self {
+ streams: BTreeMap::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: NamespaceScope,
+ namespace: LaneNamespace,
+ ) -> 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: NamespaceScope,
+ namespace: LaneNamespace,
+ ) -> 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: NamespaceScope,
+ namespace: LaneNamespace,
+ ) -> 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: BTreeMap::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: BTreeSet::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: NamespaceScope) -> bool {
+ match scope {
+ NamespaceScope::Any => true,
+ NamespaceScope::SetupOnly => origin == ResourceOrigin::Setup,
+ NamespaceScope::DynamicOnly => origin == ResourceOrigin::Dynamic,
+ }
+}
+
+fn matches_namespace(ns: Namespace, lane_ns: LaneNamespace) -> bool {
+ match lane_ns {
+ LaneNamespace::Stable => ns == Namespace::Stable,
+ LaneNamespace::Churn => ns == Namespace::Churn,
+ LaneNamespace::Both => true,
+ }
+}
diff --git a/core/lab/src/trace.rs b/core/lab/src/trace.rs
new file mode 100644
index 000000000..ab42ca681
--- /dev/null
+++ b/core/lab/src/trace.rs
@@ -0,0 +1,129 @@
+/* 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::ops::{Op, OpOutcome};
+use serde::Serialize;
+use std::fs::File;
+use std::io::{self, BufWriter, Write};
+use std::path::Path;
+use std::time::{Duration, Instant};
+
+const FLUSH_INTERVAL_ENTRIES: usize = 64;
+const FLUSH_INTERVAL: Duration = Duration::from_millis(100);
+
+pub struct WorkerTraceWriter {
+ writer: BufWriter<File>,
+ worker_id: u32,
+ start: Instant,
+ pending: usize,
+ last_flush: Instant,
+}
+
+#[derive(Serialize)]
+struct TraceEntry<'a> {
+ seq: u64,
+ t_us: u64,
+ worker: u32,
+ 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 str>,
+ #[serde(skip_serializing_if = "Option::is_none")]
+ error: Option<&'a str>,
+ },
+}
+
+impl WorkerTraceWriter {
+ pub fn new(output_dir: &Path, worker_id: u32, start: Instant) -> io::Result<Self> {
+ let path = output_dir.join(format!("trace-worker-{worker_id}.jsonl"));
+ let file = File::create(path)?;
+ Ok(Self {
+ writer: BufWriter::new(file),
+ worker_id,
+ start,
+ pending: 0,
+ last_flush: Instant::now(),
+ })
+ }
+
+ pub fn write_intent(&mut self, seq: u64, op: &Op) {
+ let entry = TraceEntry {
+ seq,
+ t_us: self.start.elapsed().as_micros() as u64,
+ worker: self.worker_id,
+ phase: "intent",
+ data: TraceData::Intent { op },
+ };
+ self.write_entry(&entry);
+ }
+
+ pub fn write_outcome(&mut self, seq: u64, outcome: &OpOutcome, latency: Duration) {
+ let (detail, error) = match outcome {
+ OpOutcome::Success { detail } => (detail.as_deref(), None),
+ OpOutcome::Error { context, error, .. } => {
+ (Some(context.as_str()), Some(error.as_str()))
+ }
+ };
+ let entry = TraceEntry {
+ seq,
+ t_us: self.start.elapsed().as_micros() as u64,
+ worker: self.worker_id,
+ phase: "outcome",
+ data: TraceData::Outcome {
+ result: outcome.result_tag(),
+ latency_us: latency.as_micros() as u64,
+ detail,
+ error,
+ },
+ };
+ self.write_entry(&entry);
+ }
+
+ fn write_entry<T: Serialize>(&mut self, entry: &T) {
+ // Serialization to a BufWriter; I/O errors here are non-fatal for chaos testing
+ if serde_json::to_writer(&mut self.writer, entry).is_ok() {
+ let _ = self.writer.write_all(b"\n");
+ self.pending += 1;
+ self.maybe_flush();
+ }
+ }
+
+ fn maybe_flush(&mut self) {
+ if self.pending >= FLUSH_INTERVAL_ENTRIES || self.last_flush.elapsed() >= FLUSH_INTERVAL {
+ self.flush();
+ }
+ }
+
+ pub fn flush(&mut self) {
+ let _ = self.writer.flush();
+ self.pending = 0;
+ self.last_flush = Instant::now();
+ }
+}
diff --git a/core/lab/src/worker.rs b/core/lab/src/worker.rs
new file mode 100644
index 000000000..bf0285081
--- /dev/null
+++ b/core/lab/src/worker.rs
@@ -0,0 +1,296 @@
+/* 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::invariants::{self, InvariantViolation, PartitionKey};
+use crate::ops::{ErrorClass, Op, OpKind, OpOutcome};
+use crate::scenarios::{LaneNamespace, NamespacePrefixes, NamespaceScope, RateModel};
+use crate::shadow::ShadowState;
+use crate::trace::WorkerTraceWriter;
+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::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: LaneNamespace,
+ pub fail_fast: bool,
+ pub ops_target: Option<u64>,
+ pub scope: NamespaceScope,
+ pub rate: RateModel,
+ pub destructive: bool,
+ pub consumer_id: u32,
+}
+
+pub struct WorkerInit {
+ pub id: u32,
+ pub client: IggyClient,
+ pub rng: StdRng,
+ pub shared_state: Arc<RwLock<ShadowState>>,
+ pub trace_writer: Option<WorkerTraceWriter>,
+ pub seq_counter: Arc<AtomicU64>,
+ pub ops_counter: Arc<AtomicU64>,
+}
+
+struct Worker {
+ id: u32,
+ client: IggyClient,
+ rng: StdRng,
+ shared_state: Arc<RwLock<ShadowState>>,
+ trace_writer: Option<WorkerTraceWriter>,
+ op_kinds: Vec<OpKind>,
+ weighted_dist: WeightedIndex<f64>,
+ config: WorkerConfig,
+ seq_counter: Arc<AtomicU64>,
+ ops_counter: Arc<AtomicU64>,
+}
+
+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 != NamespaceScope::SetupOnly || !kind.is_creative();
+ let destr_ok = config.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()?;
+
+ Some(Self {
+ id: init.id,
+ client: init.client,
+ rng: init.rng,
+ shared_state: init.shared_state,
+ trace_writer: init.trace_writer,
+ op_kinds,
+ weighted_dist,
+ config,
+ seq_counter: init.seq_counter,
+ ops_counter: init.ops_counter,
+ })
+ }
+
+ 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);
+ if let Some(tw) = &mut self.trace_writer {
+ tw.write_intent(seq, &op);
+ }
+
+ let start = Instant::now();
+ let exec_result = tokio::select! {
+ result = op.execute(
+ &self.client,
+ self.config.msg_size,
+ &mut self.rng,
+ &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 only for successful shadow-mutating ops.
+ 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),
+ LabError::Timeout { .. } => ErrorClass::Transient,
+ LabError::PrefixViolation(_) | LabError::InvalidIdentifier(_) => {
+ ErrorClass::ServerBug
+ }
+ LabError::Io { .. } | LabError::TraceParse { .. } => {
+ ErrorClass::ServerBug
+ }
+ }
+ };
+ OpOutcome::error(class, e.to_string(), op.context())
+ }
+ };
+
+ if let Some(tw) = &mut self.trace_writer {
+ tw.write_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)
+ {
+ error!(worker = self.id, violation = ?violation, "Offset monotonicity violated");
+ if self.config.fail_fast {
+ return WorkerResult::ServerBug(violation);
+ }
+ }
+ }
+ 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 WorkerResult::ServerBug(InvariantViolation {
+ kind: "server_bug",
+ description: error.to_owned(),
+ context: format!("op={:?}", op),
+ });
+ }
+ }
+ 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 => {}
+ }
+
+ 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;
+ }
+ }
+
+ if let Some(tw) = &mut self.trace_writer {
+ tw.flush();
+ }
+
+ WorkerResult::Ok { ops_executed }
+ }
+
+ 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]
+ }
+}
+
+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 }
+ }
+ }
+}