gonzalo_soak/
harness.rs

1//! Ties the pieces together: spawn `gonzalod` replicas over the S3 backend, build a
2//! [`Dispatcher`] over them, run the [`workload`] while injecting replica-kill
3//! chaos, and check the [`oracle`]. Used by the bounded gate (`tests/ha_soak.rs`)
4//! and, with a longer chaos loop, by the `gonzalo-soak` binary.
5//!
6//! [`workload`]: crate::workload
7//! [`oracle`]: crate::oracle
8
9use crate::dispatch::Dispatcher;
10use crate::oracle::{self, SoakStats, Violation};
11use crate::replica::ReplicaSet;
12use crate::target::S3Target;
13use crate::workload::{self, WorkloadConfig};
14use gonzalo_core::Store;
15use gonzalo_store_server::ServerStore;
16use std::sync::Arc;
17use std::time::Duration;
18
19/// The result of a soak run: the collected stats and any invariant violations.
20pub struct SoakOutcome {
21    pub stats: SoakStats,
22    pub violations: Vec<Violation>,
23}
24
25impl SoakOutcome {
26    pub fn passed(&self) -> bool {
27        self.violations.is_empty()
28    }
29}
30
31/// Build a dispatcher (the "Service") over the replicas' HTTP endpoints.
32pub fn dispatcher_over(base_urls: &[String]) -> Result<Arc<Dispatcher>, String> {
33    let mut replicas: Vec<Arc<dyn Store>> = Vec::with_capacity(base_urls.len());
34    for url in base_urls {
35        let store = ServerStore::http(url).map_err(|e| format!("connect {url}: {e}"))?;
36        replicas.push(Arc::new(store));
37    }
38    Ok(Arc::new(Dispatcher::new(replicas)))
39}
40
41/// Run `rounds` soak rounds against `replicas` freshly-spawned `gonzalod`
42/// processes over `target`. Each round runs the workload while performing one
43/// kill+recover cycle on a (rotating) replica, then checks the oracle. Each round
44/// uses a distinct collection so op-ids never collide across rounds.
45///
46/// `rounds == 1` is exactly the bounded per-PR gate; the deep soak passes N > 1.
47pub async fn run_rounds(
48    target: &S3Target,
49    base_cfg: WorkloadConfig,
50    replicas: usize,
51    rounds: usize,
52) -> Result<Vec<SoakOutcome>, String> {
53    let mut set = ReplicaSet::spawn(replicas, target).await?;
54    let dispatcher = dispatcher_over(&set.base_urls())?;
55    let mut outcomes = Vec::with_capacity(rounds);
56
57    for r in 0..rounds {
58        let mut cfg = base_cfg.clone();
59        cfg.collection = format!("{}-r{r}", base_cfg.collection);
60
61        let workload_task = {
62            let d = dispatcher.clone();
63            tokio::spawn(async move { workload::run(d, cfg).await })
64        };
65
66        if replicas > 1 {
67            // Warm up, kill a rotating replica mid-load (a pod death), hold,
68            // then recover it — the failover path a k8s Service masks.
69            let victim = 1 + (r % (replicas - 1));
70            tokio::time::sleep(Duration::from_millis(200)).await;
71            set.kill(victim);
72            tokio::time::sleep(Duration::from_millis(450)).await;
73            set.respawn(victim).await?;
74        }
75
76        let stats = workload_task
77            .await
78            .map_err(|e| format!("workload task panicked: {e}"))?;
79        let violations = oracle::check(&stats);
80        outcomes.push(SoakOutcome { stats, violations });
81    }
82
83    Ok(outcomes)
84}
gonzalo_soak/harness.rs

gonzalo_soak/
harness.rs
