From 42a4d20687f7940f87e092e69f5fe9d54b70221a Mon Sep 17 00:00:00 2001 From: andrussal Date: Wed, 22 Jul 2026 08:32:31 +0200 Subject: [PATCH] feat(testing): add verb-level scenario DSL --- examples/queue/examples/Cargo.toml | 4 + examples/queue/examples/src/bin/dsl_demo.rs | 26 +++ examples/queue/examples/tests/dsl_chaos.rs | 21 +++ examples/queue/testing/workloads/src/dsl.rs | 131 ++++++++++++++ examples/queue/testing/workloads/src/lib.rs | 5 + .../queue/testing/workloads/src/produce.rs | 138 ++++++++++++++- testing-framework/core/src/workloads/chaos.rs | 160 +++++++++++++++++- testing-framework/core/src/workloads/mod.rs | 2 +- 8 files changed, 469 insertions(+), 18 deletions(-) create mode 100644 examples/queue/examples/src/bin/dsl_demo.rs create mode 100644 examples/queue/examples/tests/dsl_chaos.rs create mode 100644 examples/queue/testing/workloads/src/dsl.rs diff --git a/examples/queue/examples/Cargo.toml b/examples/queue/examples/Cargo.toml index 875cddb..6ffc463 100644 --- a/examples/queue/examples/Cargo.toml +++ b/examples/queue/examples/Cargo.toml @@ -24,6 +24,10 @@ path = "src/bin/compose_convergence.rs" name = "queue_compose_roundtrip" path = "src/bin/compose_roundtrip.rs" +[[bin]] +name = "queue_dsl_demo" +path = "src/bin/dsl_demo.rs" + [dependencies] anyhow = "1.0" async-trait = { workspace = true } diff --git a/examples/queue/examples/src/bin/dsl_demo.rs b/examples/queue/examples/src/bin/dsl_demo.rs new file mode 100644 index 0000000..ad2afb4 --- /dev/null +++ b/examples/queue/examples/src/bin/dsl_demo.rs @@ -0,0 +1,26 @@ +use queue_runtime_workloads::{ + QueueDslExt as _, QueueRunExt as _, QueueScenario, RestartChaosBuilderExt as _, +}; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + QueueScenario::nodes(5) + .produce(400) + .rate_per_sec(40) + .done() + .restart_nodes_randomly() + .every_secs(5, 15) + .excluding_nodes(["node-0"]) + .done() + .expect_converged(400) + .within_secs(60) + .run_secs(120) + .await + .map_err(|error| anyhow::anyhow!(error))?; + + Ok(()) +} diff --git a/examples/queue/examples/tests/dsl_chaos.rs b/examples/queue/examples/tests/dsl_chaos.rs new file mode 100644 index 0000000..46c6a58 --- /dev/null +++ b/examples/queue/examples/tests/dsl_chaos.rs @@ -0,0 +1,21 @@ +use queue_runtime_workloads::{ + QueueDslExt as _, QueueRunExt as _, QueueScenario, RestartChaosBuilderExt as _, +}; +use testing_framework_core::scenario::DynError; + +#[tokio::test] +async fn dsl_restart_scenario_converges() -> Result<(), DynError> { + QueueScenario::nodes(3) + .produce(100) + .rate_per_sec(50) + .done() + .restart_nodes_randomly() + .every_secs(4, 8) + .cooldown_secs(10) + .excluding_nodes(["node-0"]) + .done() + .expect_converged(100) + .within_secs(30) + .run_secs(30) + .await +} diff --git a/examples/queue/testing/workloads/src/dsl.rs b/examples/queue/testing/workloads/src/dsl.rs new file mode 100644 index 0000000..3b12eeb --- /dev/null +++ b/examples/queue/testing/workloads/src/dsl.rs @@ -0,0 +1,131 @@ +use std::time::Duration; + +use queue_runtime_ext::{QueueEnv, QueueLocalDeployer, QueueScenarioBuilder, QueueTopology}; +use testing_framework_core::scenario::{ + Deployer, DynError, + internal::{CoreBuilderAccess, NodeControlScenarioBuilder}, +}; + +use crate::{QueueConverges, QueueProduceWorkload}; + +/// Entry point for the queue verb DSL. +pub struct QueueScenario; + +impl QueueScenario { + #[must_use] + pub fn nodes(count: usize) -> QueueScenarioBuilder { + QueueScenarioBuilder::with_deployment(QueueTopology::new(count)) + } +} + +/// Queue domain verbs available on every scenario builder over [`QueueEnv`]. +/// +/// Verbs only expand: each sub-builder lowers to `with_workload` / +/// `with_expectation` calls with the corresponding noun object. +pub trait QueueDslExt: CoreBuilderAccess + Sized { + /// Enqueue `operations` payloads through the first node. + #[must_use] + fn produce(self, operations: usize) -> QueueProduceBuilder { + QueueProduceBuilder { + builder: self, + workload: QueueProduceWorkload::new().operations(operations), + } + } + + /// Expect all nodes to agree on a queue of at least `min_queue_len`. + #[must_use] + fn expect_converged(self, min_queue_len: usize) -> QueueConvergedBuilder { + QueueConvergedBuilder { + builder: self, + expectation: QueueConverges::new(min_queue_len), + } + } +} + +impl> QueueDslExt for B {} + +pub struct QueueProduceBuilder> { + builder: B, + workload: QueueProduceWorkload, +} + +impl> QueueProduceBuilder { + #[must_use] + pub fn rate_per_sec(mut self, value: usize) -> Self { + self.workload = self.workload.rate_per_sec(value); + self + } + + #[must_use] + pub fn payload_prefix(mut self, value: impl Into) -> Self { + self.workload = self.workload.payload_prefix(value); + self + } + + #[must_use] + pub fn done(self) -> B { + let Self { builder, workload } = self; + builder.map_core_builder(|inner| inner.with_workload(workload)) + } +} + +pub struct QueueConvergedBuilder> { + builder: B, + expectation: QueueConverges, +} + +impl> QueueConvergedBuilder { + #[must_use] + pub fn within_secs(self, secs: u64) -> B { + self.within(Duration::from_secs(secs)) + } + + #[must_use] + pub fn within(self, timeout: Duration) -> B { + let Self { + builder, + expectation, + } = self; + builder.map_core_builder(|inner| inner.with_expectation(expectation.timeout(timeout))) + } +} + +/// Finisher: set the run duration, build the scenario, and run it against the +/// local process deployer. +pub trait QueueRunExt: Sized { + fn run_secs(self, secs: u64) -> impl Future> + Send; +} + +impl QueueRunExt for QueueScenarioBuilder { + async fn run_secs(self, secs: u64) -> Result<(), DynError> { + let scenario = self + .with_run_duration(Duration::from_secs(secs)) + .build() + .map_err(DynError::from)?; + run_local(scenario).await + } +} + +impl QueueRunExt for NodeControlScenarioBuilder { + async fn run_secs(self, secs: u64) -> Result<(), DynError> { + let scenario = self + .with_run_duration(Duration::from_secs(secs)) + .build() + .map_err(DynError::from)?; + run_local(scenario).await + } +} + +async fn run_local( + mut scenario: testing_framework_core::scenario::Scenario, +) -> Result<(), DynError> +where + Caps: Send + Sync, + QueueLocalDeployer: Deployer, + >::Error: Into, +{ + let deployer = QueueLocalDeployer::default(); + let runner = deployer.deploy(&scenario).await.map_err(Into::into)?; + runner.run(&mut scenario).await?; + Ok(()) +} diff --git a/examples/queue/testing/workloads/src/lib.rs b/examples/queue/testing/workloads/src/lib.rs index 91b2670..88f1cc8 100644 --- a/examples/queue/testing/workloads/src/lib.rs +++ b/examples/queue/testing/workloads/src/lib.rs @@ -1,10 +1,15 @@ mod drained; +mod dsl; mod expectations; mod produce; mod roundtrip; pub use drained::QueueDrained; +pub use dsl::{ + QueueConvergedBuilder, QueueDslExt, QueueProduceBuilder, QueueRunExt, QueueScenario, +}; pub use expectations::QueueConverges; pub use produce::QueueProduceWorkload; pub use queue_runtime_ext::{QueueBuilderExt, QueueEnv, QueueScenarioBuilder, QueueTopology}; pub use roundtrip::QueueRoundTripWorkload; +pub use testing_framework_core::workloads::RestartChaosBuilderExt; diff --git a/examples/queue/testing/workloads/src/produce.rs b/examples/queue/testing/workloads/src/produce.rs index c2e3794..0a26224 100644 --- a/examples/queue/testing/workloads/src/produce.rs +++ b/examples/queue/testing/workloads/src/produce.rs @@ -1,10 +1,16 @@ use std::time::Duration; use async_trait::async_trait; +use queue_node::QueueHttpClient; use queue_runtime_ext::QueueEnv; use serde::{Deserialize, Serialize}; use testing_framework_core::scenario::{DynError, RunContext, Workload}; -use tracing::info; +use tracing::{info, warn}; + +const REQUEST_RETRY_INTERVAL: Duration = Duration::from_millis(250); +const REQUEST_RETRY_WINDOW: Duration = Duration::from_secs(30); +const ENSURE_PRODUCED_WINDOW: Duration = Duration::from_secs(60); +const ENSURE_STABILITY_DELAY: Duration = Duration::from_secs(1); #[derive(Clone)] pub struct QueueProduceWorkload { @@ -25,6 +31,11 @@ struct EnqueueResponse { queue_len: usize, } +#[derive(Deserialize)] +struct ProducerStateResponse { + queue_len: usize, +} + impl QueueProduceWorkload { #[must_use] pub fn new() -> Self { @@ -81,13 +92,7 @@ impl Workload for QueueProduceWorkload { for idx in 0..self.operations { let payload = format!("{}-{idx}", self.payload_prefix); - let response: EnqueueResponse = producer - .post("/queue/enqueue", &EnqueueRequest { payload }) - .await?; - - if !response.accepted { - return Err(format!("node rejected enqueue at operation {idx}").into()); - } + let response = enqueue_with_retry(producer, payload, idx).await?; if (idx + 1) % 25 == 0 { info!( @@ -103,7 +108,122 @@ impl Workload for QueueProduceWorkload { } } - Ok(()) + self.ensure_produced(producer, interval).await + } +} + +impl QueueProduceWorkload { + /// Top up the queue until the produced count is durably visible. + /// + /// A node restart wipes its in-memory queue; ops accepted but not yet + /// pulled by a peer (or enqueued before the restarted node re-adopted the + /// cluster state) are lost. Re-reads the producer state and enqueues the + /// deficit until the target sticks across a sync interval. + async fn ensure_produced( + &self, + producer: &QueueHttpClient, + interval: Option, + ) -> Result<(), DynError> { + let deadline = tokio::time::Instant::now() + ENSURE_PRODUCED_WINDOW; + let mut extra_index = 0_usize; + + loop { + let observed = producer_queue_len(producer).await?; + + if observed >= self.operations { + tokio::time::sleep(ENSURE_STABILITY_DELAY).await; + if producer_queue_len(producer).await? >= self.operations { + if extra_index > 0 { + info!( + topped_up = extra_index, + target = self.operations, + "queue produce recovered lost operations" + ); + } + return Ok(()); + } + continue; + } + + if tokio::time::Instant::now() >= deadline { + return Err(format!( + "queue produce could not reach {} durable operations (observed {observed})", + self.operations + ) + .into()); + } + + warn!( + observed, + target = self.operations, + "queue produce detected lost operations; topping up" + ); + + for _ in observed..self.operations { + let payload = format!("{}-extra-{extra_index}", self.payload_prefix); + enqueue_with_retry(producer, payload, self.operations + extra_index).await?; + extra_index += 1; + + if let Some(delay) = interval { + tokio::time::sleep(delay).await; + } + } + } + } +} + +async fn enqueue_with_retry( + producer: &QueueHttpClient, + payload: String, + operation: usize, +) -> Result { + let deadline = tokio::time::Instant::now() + REQUEST_RETRY_WINDOW; + + loop { + match producer + .post( + "/queue/enqueue", + &EnqueueRequest { + payload: payload.clone(), + }, + ) + .await + { + Ok(response) => { + let response: EnqueueResponse = response; + if !response.accepted { + return Err(format!("node rejected enqueue at operation {operation}").into()); + } + return Ok(response); + } + Err(error) => { + if tokio::time::Instant::now() >= deadline { + return Err(format!( + "queue enqueue kept failing at operation {operation}: {error}" + ) + .into()); + } + tokio::time::sleep(REQUEST_RETRY_INTERVAL).await; + } + } + } +} + +async fn producer_queue_len(producer: &QueueHttpClient) -> Result { + let deadline = tokio::time::Instant::now() + REQUEST_RETRY_WINDOW; + + loop { + match producer.get::("/queue/state").await { + Ok(state) => return Ok(state.queue_len), + Err(error) => { + if tokio::time::Instant::now() >= deadline { + return Err( + format!("queue state kept failing during produce top-up: {error}").into(), + ); + } + tokio::time::sleep(REQUEST_RETRY_INTERVAL).await; + } + } } } diff --git a/testing-framework/core/src/workloads/chaos.rs b/testing-framework/core/src/workloads/chaos.rs index 6c6be9b..504c2cf 100644 --- a/testing-framework/core/src/workloads/chaos.rs +++ b/testing-framework/core/src/workloads/chaos.rs @@ -1,12 +1,17 @@ -use std::{collections::HashMap, mem::swap, time::Duration}; +use std::{ + collections::{HashMap, HashSet}, + mem::swap, + time::Duration, +}; use async_trait::async_trait; use rand::{Rng as _, seq::SliceRandom as _, thread_rng}; -use tokio::time::{Instant, sleep}; +use tokio::time::{Instant, sleep, sleep_until, timeout_at}; use crate::{ scenario::{ - Application, DynError, NodeControlCapability, RunContext, Workload, internal::CoreBuilder, + Application, DynError, NodeControlCapability, RunContext, ScenarioBuilder, Workload, + internal::{CoreBuilder, CoreBuilderAccess, NodeControlScenarioBuilder}, }, topology::DeploymentDescriptor, }; @@ -40,6 +45,122 @@ impl ChaosBuilderExt for CoreBuilder RestartBuilder; +} + +impl RestartChaosBuilderExt for ScenarioBuilder { + type Target = NodeControlScenarioBuilder; + + fn restart_nodes_randomly(self) -> RestartBuilder { + RestartBuilder::new(self.with_node_control()) + } +} + +impl RestartChaosBuilderExt for NodeControlScenarioBuilder { + type Target = Self; + + fn restart_nodes_randomly(self) -> RestartBuilder { + RestartBuilder::new(self) + } +} + +impl RestartChaosBuilderExt for CoreBuilder { + type Target = CoreBuilder; + + fn restart_nodes_randomly(self) -> RestartBuilder { + RestartBuilder::new(self.with_node_control()) + } +} + +impl RestartChaosBuilderExt for CoreBuilder { + type Target = Self; + + fn restart_nodes_randomly(self) -> RestartBuilder { + RestartBuilder::new(self) + } +} + +pub struct RestartBuilder { + builder: B, + min_delay: Duration, + max_delay: Duration, + target_cooldown: Duration, + excluded_nodes: HashSet, +} + +impl RestartBuilder { + fn new(builder: B) -> Self { + Self { + builder, + min_delay: DEFAULT_CHAOS_MIN_DELAY, + max_delay: DEFAULT_CHAOS_MAX_DELAY, + target_cooldown: DEFAULT_CHAOS_TARGET_COOLDOWN, + excluded_nodes: HashSet::new(), + } + } + + #[must_use] + pub fn every_secs(self, min: u64, max: u64) -> Self { + self.every(Duration::from_secs(min), Duration::from_secs(max)) + } + + #[must_use] + pub const fn every(mut self, min: Duration, max: Duration) -> Self { + self.min_delay = min; + self.max_delay = max; + self + } + + #[must_use] + pub fn cooldown_secs(self, secs: u64) -> Self { + self.cooldown(Duration::from_secs(secs)) + } + + #[must_use] + pub const fn cooldown(mut self, cooldown: Duration) -> Self { + self.target_cooldown = cooldown; + self + } + + #[must_use] + pub fn excluding_nodes(mut self, nodes: impl IntoIterator>) -> Self { + self.excluded_nodes + .extend(nodes.into_iter().map(Into::into)); + self + } + + #[must_use] + pub fn done(self) -> B { + let Self { + builder, + mut min_delay, + mut max_delay, + mut target_cooldown, + excluded_nodes, + } = self; + + if min_delay > max_delay { + swap(&mut min_delay, &mut max_delay); + } + + if target_cooldown < min_delay { + target_cooldown = min_delay; + } + + builder.map_core_builder(|inner| { + inner.with_workload( + RandomRestartWorkload::new(min_delay, max_delay, target_cooldown) + .excluding_nodes(excluded_nodes), + ) + }) + } +} + pub struct ChaosBuilder { builder: CoreBuilder, } @@ -116,18 +237,27 @@ pub struct RandomRestartWorkload { min_delay: Duration, max_delay: Duration, target_cooldown: Duration, + excluded_nodes: HashSet, } impl RandomRestartWorkload { #[must_use] - pub const fn new(min_delay: Duration, max_delay: Duration, target_cooldown: Duration) -> Self { + pub fn new(min_delay: Duration, max_delay: Duration, target_cooldown: Duration) -> Self { Self { min_delay, max_delay, target_cooldown, + excluded_nodes: HashSet::new(), } } + #[must_use] + pub fn excluding_nodes(mut self, nodes: impl IntoIterator>) -> Self { + self.excluded_nodes + .extend(nodes.into_iter().map(Into::into)); + self + } + fn random_delay(&self) -> Duration { if self.max_delay <= self.min_delay { return self.min_delay; @@ -165,7 +295,12 @@ impl RandomRestartWorkload { return Vec::new(); } - (0..node_count).map(node_target).collect() + (0..node_count) + .map(node_target) + .filter(|target| match target { + Target::Node(name) => !self.excluded_nodes.contains(name), + }) + .collect() } async fn pick_target( @@ -253,9 +388,16 @@ impl Workload for RandomRestartWorkload { let mut cooldowns = self.initialize_cooldowns(&targets); - loop { - sleep(self.random_delay()).await; - let target = self.pick_target(&targets, &cooldowns).await?; + let deadline = Instant::now() + ctx.run_duration(); + while Instant::now() < deadline { + sleep_until((Instant::now() + self.random_delay()).min(deadline)).await; + if Instant::now() >= deadline { + break; + } + let target = match timeout_at(deadline, self.pick_target(&targets, &cooldowns)).await { + Ok(target) => target?, + Err(_) => break, + }; match target { Target::Node(ref name) => handle @@ -266,6 +408,8 @@ impl Workload for RandomRestartWorkload { cooldowns.insert(target, Instant::now() + self.target_cooldown); } + + Ok(()) } } diff --git a/testing-framework/core/src/workloads/mod.rs b/testing-framework/core/src/workloads/mod.rs index c1813d2..420ffb4 100644 --- a/testing-framework/core/src/workloads/mod.rs +++ b/testing-framework/core/src/workloads/mod.rs @@ -1,3 +1,3 @@ mod chaos; -pub use chaos::{ChaosBuilderExt, RandomRestartWorkload}; +pub use chaos::{ChaosBuilderExt, RandomRestartWorkload, RestartBuilder, RestartChaosBuilderExt};