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//! After each round the harness reads every lifecycle key from **every** replica
7//! (`workload::collect_lifecycle`), so the oracle can check that the replicas agree
8//! on which keys are deleted (#203).
9//!
10//! [`workload`]: crate::workload
11//! [`oracle`]: crate::oracle
12
13use crate::dispatch::Dispatcher;
14use crate::oracle::{self, SoakStats, Violation};
15use crate::replica::ReplicaSet;
16use crate::target::S3Target;
17use crate::workload::{self, WorkloadConfig};
18use gonzalo_core::Store;
19use gonzalo_store_server::ServerStore;
20use std::sync::Arc;
21use std::time::Duration;
22
23/// The result of a soak run: the collected stats and any invariant violations.
24pub struct SoakOutcome {
25    pub stats: SoakStats,
26    pub violations: Vec<Violation>,
27}
28
29impl SoakOutcome {
30    pub fn passed(&self) -> bool {
31        self.violations.is_empty()
32    }
33}
34
35/// Build a dispatcher (the "Service") over the replicas' HTTP endpoints.
36pub fn dispatcher_over(base_urls: &[String]) -> Result<Arc<Dispatcher>, String> {
37    let mut replicas: Vec<Arc<dyn Store>> = Vec::with_capacity(base_urls.len());
38    for url in base_urls {
39        let store = ServerStore::http(url).map_err(|e| format!("connect {url}: {e}"))?;
40        replicas.push(Arc::new(store));
41    }
42    Ok(Arc::new(Dispatcher::new(replicas)))
43}
44
45/// Run `rounds` soak rounds against `replicas` freshly-spawned `gonzalod`
46/// processes over `target`. Each round runs the workload while performing one
47/// kill+recover cycle on a (rotating) replica, then checks the oracle. Each round
48/// uses a distinct collection so op-ids never collide across rounds.
49///
50/// `rounds == 1` is exactly the bounded per-PR gate; the deep soak passes N > 1.
51pub async fn run_rounds(
52    target: &S3Target,
53    base_cfg: WorkloadConfig,
54    replicas: usize,
55    rounds: usize,
56) -> Result<Vec<SoakOutcome>, String> {
57    let mut set = ReplicaSet::spawn(replicas, target).await?;
58    let dispatcher = dispatcher_over(&set.base_urls())?;
59    let mut outcomes = Vec::with_capacity(rounds);
60
61    for r in 0..rounds {
62        let mut cfg = base_cfg.clone();
63        cfg.collection = format!("{}-r{r}", base_cfg.collection);
64        let round_cfg = cfg.clone();
65
66        let workload_task = {
67            let d = dispatcher.clone();
68            tokio::spawn(async move { workload::run(d, cfg).await })
69        };
70
71        if replicas > 1 {
72            // Warm up, kill a rotating replica mid-load (a pod death), hold,
73            // then recover it — the failover path a k8s Service masks.
74            let victim = 1 + (r % (replicas - 1));
75            tokio::time::sleep(Duration::from_millis(200)).await;
76            set.kill(victim);
77            tokio::time::sleep(Duration::from_millis(450)).await;
78            set.respawn(victim).await?;
79        }
80
81        let mut stats = workload_task
82            .await
83            .map_err(|e| format!("workload task panicked: {e}"))?;
84        // Settled: the writers have stopped and the victim answered /readyz again,
85        // so every replica is live. The replicas front one shared bucket, so there
86        // is no replication lag to wait out. Read every replica directly.
87        stats.lifecycle = workload::collect_lifecycle(dispatcher.replicas(), &round_cfg).await;
88        let violations = oracle::check(&stats);
89        outcomes.push(SoakOutcome { stats, violations });
90    }
91
92    Ok(outcomes)
93}
gonzalo_soak/harness.rs

gonzalo_soak/
harness.rs
