diff --git a/awa/tests/persona_soak/config.rs b/awa/tests/persona_soak/config.rs index bae8a9b03..8287a7d78 100644 --- a/awa/tests/persona_soak/config.rs +++ b/awa/tests/persona_soak/config.rs @@ -152,6 +152,18 @@ pub struct SoakConfig { /// for every runtime. The library's own defaults are 30 s and 90 s. pub heartbeat_interval: Duration, pub heartbeat_staleness: Duration, + /// How often the resident memory of the soak process is sampled. + pub memory_sample_every: Duration, + /// Load time excluded from the memory growth gate while caches and pools fill. + pub memory_warmup: Duration, + /// Executions after the warm-up below which the memory growth gate is not evaluated. + pub memory_min_executions: i64, + /// Largest accepted growth of resident memory, in MiB per 100k executions. + pub memory_slope_limit_mib_per_100k: f64, + /// How long completions are held back halfway through the load; zero skips it. + pub completion_stall: Duration, + /// Largest accepted rise of resident memory, in MiB, during and after that stall. + pub completion_stall_memory_limit_mib: f64, /// Queue and claim ring rotation intervals and the prune lock timeout of every /// in-process runtime. Unset keeps the library defaults. pub queue_rotate_interval: Option, @@ -272,6 +284,20 @@ impl SoakConfig { heartbeat_staleness: scaled_staleness(Duration::from_secs( env_parse("AWA_SOAK_STALENESS_SECONDS").unwrap_or(DEFAULT_STALENESS_SECONDS), )), + memory_sample_every: Duration::from_secs( + env_parse("AWA_SOAK_MEMORY_SAMPLE_SECONDS").unwrap_or(5), + ), + memory_warmup: Duration::from_secs( + env_parse("AWA_SOAK_MEMORY_WARMUP_SECONDS").unwrap_or(120), + ), + memory_min_executions: env_parse("AWA_SOAK_MEMORY_MIN_EXECUTIONS").unwrap_or(20_000), + memory_slope_limit_mib_per_100k: env_parse("AWA_SOAK_MEMORY_SLOPE_MIB_PER_100K") + .unwrap_or(64.0), + completion_stall: Duration::from_secs( + env_parse("AWA_SOAK_COMPLETION_STALL_SECONDS").unwrap_or(0), + ), + completion_stall_memory_limit_mib: env_parse("AWA_SOAK_COMPLETION_STALL_MEMORY_MIB") + .unwrap_or(128.0), queue_rotate_interval: env_parse::("AWA_SOAK_QUEUE_ROTATE_MS") .map(Duration::from_millis), claim_rotate_interval: env_parse::("AWA_SOAK_CLAIM_ROTATE_MS") diff --git a/awa/tests/persona_soak/memory.rs b/awa/tests/persona_soak/memory.rs new file mode 100644 index 000000000..0dce1b446 --- /dev/null +++ b/awa/tests/persona_soak/memory.rs @@ -0,0 +1,382 @@ +//! Resident memory of the soak process against handler executions. +//! +//! Every runtime the personas start lives in this process, so its resident set is the +//! worker-side memory under test. A sampler records it alongside the executions so far; +//! after the warm-up, resident memory must not keep climbing with executions. Growth is +//! the least-squares slope of resident memory over executions, so a bounded transient +//! barely moves it while memory retained per execution shows up however long the run. +//! +//! An optional completion stall holds completions back while handlers keep running. The +//! window from the stall to as long again after it is left out of the slope and checked on +//! its own: resident memory may rise by at most a fixed amount, however long the stall. +//! +//! The harness's own ledger also lives here and grows with every job enqueued. When the +//! runtimes fall behind the producers, as on a host short of CPU, that growth lands on few +//! executions and would read as a leak, so the slope is then reported as not evaluated. + +use super::config::Persona; +use super::shared::Shared; +use super::stop::Stop; +use sqlx::Connection; +use std::fmt::Write; +use std::sync::{Arc, Mutex}; + +#[derive(Clone, Copy, Debug, serde::Serialize)] +pub struct MemorySample { + pub at_ms: i64, + pub load: bool, + pub executions: i64, + pub ledger_jobs: i64, + pub rss_kib: i64, + pub hwm_kib: i64, + pub rss_anon_kib: i64, +} + +#[derive(Default)] +pub struct MemoryLog { + samples: Mutex>, + verdict: Mutex>, + stall_ms: Mutex>, + stall_started: Mutex>, +} + +#[derive(Clone, Debug, serde::Serialize)] +pub struct MemoryVerdict { + pub samples: usize, + pub executions_evaluated: i64, + pub enqueued_evaluated: i64, + /// Why the slope was not evaluated, when it was not. + pub not_evaluated: Option, + pub slope_mib_per_100k: Option, + pub limit_mib_per_100k: f64, + pub start_rss_mib: f64, + pub end_rss_mib: f64, + pub peak_hwm_mib: f64, + pub stall_rise_mib: Option, + pub stall_limit_mib: f64, +} + +impl MemoryLog { + pub fn samples(&self) -> Vec { + self.samples.lock().expect("memory samples").clone() + } + + /// Length of the completion stall once it has started, zero otherwise. + pub fn stalled_for(&self) -> std::time::Duration { + match *self.stall_started.lock().expect("stall start") { + Some(hold) => hold, + None => std::time::Duration::ZERO, + } + } + + pub fn verdict(&self) -> Option { + self.verdict.lock().expect("memory verdict").clone() + } + + fn push(&self, sample: MemorySample) { + self.samples.lock().expect("memory samples").push(sample); + } +} + +fn status_kib(status: &str, field: &str) -> i64 { + status + .lines() + .find(|line| line.starts_with(field)) + .and_then(|line| line.split_whitespace().nth(1)) + .and_then(|value| value.parse().ok()) + .unwrap_or(0) +} + +fn executions(shared: &Shared) -> i64 { + Persona::ALL + .into_iter() + .filter(|persona| shared.config.has(*persona)) + .map(|persona| shared.stats(persona).get("executions")) + .sum() +} + +/// Samples until `stop`. `/proc/self/status` exists only on Linux; elsewhere nothing is +/// recorded and the gate reports that it could not evaluate. +pub async fn sample(shared: Arc, stop: Stop) { + loop { + if let Ok(status) = std::fs::read_to_string("/proc/self/status") { + shared.memory.push(MemorySample { + at_ms: shared.now_ms(), + load: !shared.load_stop.is_stopped(), + executions: executions(&shared), + ledger_jobs: shared.ledger.len() as i64, + rss_kib: status_kib(&status, "VmRSS:"), + hwm_kib: status_kib(&status, "VmHWM:"), + rss_anon_kib: status_kib(&status, "RssAnon:"), + }); + } + if stop.sleep(shared.config.memory_sample_every).await { + return; + } + } +} + +/// Halfway through the load, holds the queue-storage completion tables in SHARE mode for +/// `completion_stall`: completions cannot commit while claims and handlers carry on, the +/// shape in which a runtime that buffers finished attempts grows without bound. +pub async fn stall_completions(shared: Arc) { + let hold = shared.config.completion_stall; + if hold.is_zero() { + return; + } + let Some(schema) = shared.info.queue_storage_schema.clone() else { + shared.note( + "memory", + "could not stall completions: no queue-storage schema", + ); + return; + }; + if shared.load_stop.sleep(shared.config.load / 2).await { + return; + } + let stalled = async { + let mut connection = sqlx::PgConnection::connect(&shared.config.database_url).await?; + let mut transaction = connection.begin().await?; + sqlx::query(&format!( + "LOCK TABLE {schema}.lease_claim_closures, {schema}.receipt_completion_batches \ + IN SHARE MODE" + )) + .execute(&mut *transaction) + .await?; + *shared.memory.stall_started.lock().expect("stall start") = Some(hold); + shared.chaos_window(hold + shared.config.chaos_slack); + shared.note( + "memory", + format!("completions stalled for {}s", hold.as_secs()), + ); + let stalled_at = shared.now_ms(); + shared.load_stop.sleep(hold).await; + let released = transaction.rollback().await; + *shared.memory.stall_ms.lock().expect("stall window") = Some((stalled_at, shared.now_ms())); + released + } + .await; + match stalled { + Ok(()) => shared.note("memory", "completions released"), + Err(error) => shared.note("memory", format!("could not stall completions: {error}")), + } +} + +/// Least-squares slope of y over x shared by every segment, each with its own intercept, +/// so a step between segments does not count as growth. +fn slope(segments: &[Vec<(f64, f64)>]) -> Option { + let mut covariance = 0.0; + let mut spread = 0.0; + let mut points = 0; + for segment in segments.iter().filter(|segment| segment.len() >= 2) { + let count = segment.len() as f64; + let mean_x = segment.iter().map(|(x, _)| x).sum::() / count; + let mean_y = segment.iter().map(|(_, y)| y).sum::() / count; + covariance += segment + .iter() + .map(|(x, y)| (x - mean_x) * (y - mean_y)) + .sum::(); + spread += segment + .iter() + .map(|(x, _)| (x - mean_x).powi(2)) + .sum::(); + points += segment.len(); + } + (points >= 3 && spread > 0.0).then(|| covariance / spread) +} + +/// Records the gate's verdict and raises `memory_growth` when resident memory grows +/// faster than the configured limit per 100k executions after the warm-up. +pub fn evaluate(shared: &Shared, load_started_ms: i64) { + let samples = shared.memory.samples(); + let (Some(first), Some(last)) = (samples.first().copied(), samples.last().copied()) else { + shared.note( + "memory", + "no memory samples: /proc/self/status is unavailable", + ); + return; + }; + let warm_from = load_started_ms + shared.config.memory_warmup.as_millis() as i64; + let stall = *shared.memory.stall_ms.lock().expect("stall window"); + let in_stall = |at_ms: i64| { + stall.is_some_and(|(start, end)| at_ms >= start && at_ms <= end + (end - start)) + }; + let steady: Vec<&MemorySample> = samples + .iter() + .filter(|sample| sample.load && sample.at_ms >= warm_from && !in_stall(sample.at_ms)) + .collect(); + let stall_rise_mib = stall.and_then(|(start, _)| { + let before = samples.iter().rev().find(|sample| sample.at_ms <= start)?; + let peak = samples + .iter() + .filter(|sample| in_stall(sample.at_ms)) + .map(|sample| sample.rss_kib) + .max()?; + Some((peak - before.rss_kib) as f64 / 1024.0) + }); + let (before_stall, after_stall): (Vec<&MemorySample>, Vec<&MemorySample>) = steady + .iter() + .partition(|sample| stall.is_none_or(|(start, _)| sample.at_ms < start)); + let segments: Vec> = [&before_stall, &after_stall] + .iter() + .map(|segment| { + segment + .iter() + .map(|sample| (sample.executions as f64, sample.rss_kib as f64)) + .collect() + }) + .collect(); + let executions_evaluated: i64 = [&before_stall, &after_stall] + .iter() + .filter_map(|segment| Some(segment.last()?.executions - segment.first()?.executions)) + .sum(); + let enqueued_evaluated: i64 = [&before_stall, &after_stall] + .iter() + .filter_map(|segment| Some(segment.last()?.ledger_jobs - segment.first()?.ledger_jobs)) + .sum(); + let not_evaluated = if executions_evaluated < shared.config.memory_min_executions { + Some(format!( + "{executions_evaluated} executions after the warm-up, {} needed", + shared.config.memory_min_executions + )) + } else if enqueued_evaluated * 4 > executions_evaluated * 5 { + Some(format!( + "the runtimes fell behind the producers ({enqueued_evaluated} jobs enqueued, \ + {executions_evaluated} executions after the warm-up), so the harness ledger's \ + growth would dominate the slope" + )) + } else { + None + }; + let slope_mib_per_100k = if not_evaluated.is_none() { + slope(&segments).map(|kib_per_execution| kib_per_execution * 100_000.0 / 1024.0) + } else { + None + }; + let verdict = MemoryVerdict { + samples: samples.len(), + executions_evaluated, + enqueued_evaluated, + not_evaluated, + slope_mib_per_100k, + limit_mib_per_100k: shared.config.memory_slope_limit_mib_per_100k, + start_rss_mib: first.rss_kib as f64 / 1024.0, + end_rss_mib: last.rss_kib as f64 / 1024.0, + peak_hwm_mib: samples + .iter() + .map(|sample| sample.hwm_kib) + .max() + .unwrap_or(0) as f64 + / 1024.0, + stall_rise_mib, + stall_limit_mib: shared.config.completion_stall_memory_limit_mib, + }; + match verdict.stall_rise_mib { + // A stall that was asked for but never held would otherwise pass unseen. + None if !shared.config.completion_stall.is_zero() => shared.violation( + "memory_growth_under_stall", + format!( + "AWA_SOAK_COMPLETION_STALL_SECONDS={} was set but no stall was measured; \ + see the memory notes in the timeline", + shared.config.completion_stall.as_secs() + ), + vec![], + ), + Some(rise) if rise > verdict.stall_limit_mib => shared.violation( + "memory_growth_under_stall", + format!( + "resident memory rose {rise:.0} MiB while completions were stalled \ + (limit {:.0}); finished attempts are piling up in memory", + verdict.stall_limit_mib + ), + vec![], + ), + Some(rise) => shared.note( + "memory", + format!( + "resident memory rose {rise:.0} MiB while completions were stalled (limit {:.0})", + verdict.stall_limit_mib + ), + ), + None => {} + } + match verdict.slope_mib_per_100k { + Some(slope) if slope > verdict.limit_mib_per_100k => shared.violation( + "memory_growth", + format!( + "resident memory grew {slope:.1} MiB per 100k executions over {} executions \ + after the warm-up (limit {:.1}); see memory.csv", + verdict.executions_evaluated, verdict.limit_mib_per_100k + ), + vec![], + ), + Some(slope) => shared.note( + "memory", + format!( + "resident memory slope {slope:.1} MiB per 100k executions over {} executions \ + (limit {:.1})", + verdict.executions_evaluated, verdict.limit_mib_per_100k + ), + ), + None => shared.note( + "memory", + format!( + "memory growth not evaluated: {}", + verdict.not_evaluated.as_deref().unwrap_or("no slope") + ), + ), + } + *shared.memory.verdict.lock().expect("memory verdict") = Some(verdict); +} + +pub fn csv(samples: &[MemorySample]) -> String { + let mut out = String::from("at_s,phase,executions,ledger_jobs,rss_kib,hwm_kib,rss_anon_kib\n"); + for sample in samples { + let _ = writeln!( + out, + "{:.1},{},{},{},{},{},{}", + sample.at_ms as f64 / 1000.0, + if sample.load { "load" } else { "drain" }, + sample.executions, + sample.ledger_jobs, + sample.rss_kib, + sample.hwm_kib, + sample.rss_anon_kib + ); + } + out +} + +#[cfg(test)] +mod tests { + use super::slope; + + fn line(from: usize, to: usize, intercept: f64, per_execution: f64) -> Vec<(f64, f64)> { + (from..to) + .map(|step| { + let executions = step as f64 * 1_000.0; + (executions, intercept + executions * per_execution) + }) + .collect() + } + + #[test] + fn slope_recovers_linear_growth() { + let kib_per_execution = slope(&[line(0, 10, 50_000.0, 0.17)]).expect("slope"); + assert!((kib_per_execution - 0.17).abs() < 1e-9); + } + + #[test] + fn slope_ignores_a_step_between_segments() { + let segments = [line(0, 10, 50_000.0, 0.0), line(20, 30, 90_000.0, 0.0)]; + assert_eq!(slope(&segments), Some(0.0)); + let growing = [line(0, 10, 50_000.0, 0.17), line(20, 30, 90_000.0, 0.17)]; + assert!((slope(&growing).expect("slope") - 0.17).abs() < 1e-9); + } + + #[test] + fn slope_needs_spread_in_executions() { + assert_eq!(slope(&[vec![(5.0, 1.0), (5.0, 2.0), (5.0, 3.0)]]), None); + assert_eq!(slope(&[vec![(1.0, 1.0), (2.0, 2.0)]]), None); + } +} diff --git a/awa/tests/persona_soak/mod.rs b/awa/tests/persona_soak/mod.rs index bfb4ccb14..ad6795b84 100644 --- a/awa/tests/persona_soak/mod.rs +++ b/awa/tests/persona_soak/mod.rs @@ -8,6 +8,7 @@ pub mod db; pub mod fleet; pub mod ledger; pub mod maintenance_metrics; +pub mod memory; pub mod oracle; pub mod personas; pub mod pgserver; diff --git a/awa/tests/persona_soak/oracle.rs b/awa/tests/persona_soak/oracle.rs index ed5f53366..1393b5ce1 100644 --- a/awa/tests/persona_soak/oracle.rs +++ b/awa/tests/persona_soak/oracle.rs @@ -358,7 +358,7 @@ impl Oracle { let schema = schema(&self.shared); let mut threshold = self.shared.config.promotion_overdue; if self.shared.chaos_active() { - threshold += self.shared.config.chaos_slack; + threshold += self.shared.chaos_slack(); } let rows: Result, _> = sqlx::query_as(&format!( "SELECT job_id, queue, extract(epoch FROM now() - run_at)::float8 @@ -447,7 +447,7 @@ impl Oracle { let now = self.shared.now_ms(); let mut window = self.shared.config.stall_window; if self.shared.chaos_active() { - window += self.shared.config.chaos_slack; + window += self.shared.chaos_slack(); } let mut candidates = Vec::new(); for (queue_name, backlog, cursor) in rows { diff --git a/awa/tests/persona_soak/report.rs b/awa/tests/persona_soak/report.rs index d309bfc94..84259795b 100644 --- a/awa/tests/persona_soak/report.rs +++ b/awa/tests/persona_soak/report.rs @@ -134,6 +134,40 @@ pub fn render(shared: &Shared, outcome: &Outcome) -> String { } let _ = writeln!(out); + let _ = writeln!(out, "## Memory"); + match shared.memory.verdict() { + Some(verdict) => { + let _ = writeln!( + out, + "- resident: start {:.0} MiB, end {:.0} MiB, peak {:.0} MiB ([memory.csv](memory.csv))", + verdict.start_rss_mib, verdict.end_rss_mib, verdict.peak_hwm_mib + ); + let _ = match verdict.slope_mib_per_100k { + Some(slope) => writeln!( + out, + "- growth after warm-up: {slope:.1} MiB per 100k executions over {} executions (limit {:.1})", + verdict.executions_evaluated, verdict.limit_mib_per_100k + ), + None => writeln!( + out, + "- growth after warm-up: not evaluated: {}", + verdict.not_evaluated.as_deref().unwrap_or("no slope") + ), + }; + if let Some(rise) = verdict.stall_rise_mib { + let _ = writeln!( + out, + "- rise while completions were stalled: {rise:.0} MiB (limit {:.0})", + verdict.stall_limit_mib + ); + } + } + None => { + let _ = writeln!(out, "- not sampled"); + } + } + let _ = writeln!(out); + let findings = shared.findings.lock().expect("findings"); let _ = writeln!(out, "## Findings, not fatal ({})", findings.len()); for finding in findings.iter() { @@ -214,10 +248,15 @@ pub fn write(shared: &Shared, outcome: &Outcome) -> std::io::Result<()> { "personas_stats": per_persona, "abandon": shared.abandon.to_json(), "run": { "counters": shared.run.counters(), "samples": shared.run.summaries() }, + "memory": shared.memory.verdict(), "violations": violations, "findings": shared.findings.lock().expect("findings").clone(), "timeline": timeline, }); + std::fs::write( + dir.join("memory.csv"), + super::memory::csv(&shared.memory.samples()), + )?; std::fs::write( dir.join("report.json"), serde_json::to_string_pretty(&summary).unwrap_or_default(), diff --git a/awa/tests/persona_soak/run.rs b/awa/tests/persona_soak/run.rs index 849b94321..ab493accb 100644 --- a/awa/tests/persona_soak/run.rs +++ b/awa/tests/persona_soak/run.rs @@ -135,6 +135,12 @@ pub async fn run_soak(config: SoakConfig) -> Result, String> { format!("load phase started: {}s", shared.config.load.as_secs()), ); shared.started_load(Instant::now()); + let load_started_ms = shared.now_ms(); + let memory_sampler = tokio::spawn(super::memory::sample( + shared.clone(), + shared.drain_stop.clone(), + )); + let completion_stall = tokio::spawn(super::memory::stall_completions(shared.clone())); let load_deadline = Instant::now() + shared.config.load; let mut next_progress = Instant::now() + Duration::from_secs(30); @@ -201,6 +207,13 @@ pub async fn run_soak(config: SoakConfig) -> Result, String> { } shared.drain_stop.trigger(); + completion_stall + .await + .map_err(|error| format!("completion stall task: {error}"))?; + memory_sampler + .await + .map_err(|error| format!("memory sampler task: {error}"))?; + super::memory::evaluate(&shared, load_started_ms); for (persona, task) in tasks { match tokio::time::timeout( crate::ci_timing::scaled_timeout(Duration::from_secs(180)), diff --git a/awa/tests/persona_soak/shared.rs b/awa/tests/persona_soak/shared.rs index 2bd038081..639debb2c 100644 --- a/awa/tests/persona_soak/shared.rs +++ b/awa/tests/persona_soak/shared.rs @@ -135,6 +135,8 @@ pub struct Shared { pub shardops: super::shard_oracle::ShardOpsState, /// State of the claim-and-abandon persona and the checks that go with it. pub abandon: super::abandon_oracle::AbandonState, + /// Resident memory of this process over the run, and the growth gate's verdict. + pub memory: super::memory::MemoryLog, chaos_until_ms: AtomicI64, } @@ -166,6 +168,7 @@ impl Shared { fleets: Mutex::new(HashMap::new()), shardops: Default::default(), abandon: Default::default(), + memory: Default::default(), chaos_until_ms: AtomicI64::new(0), }) } @@ -198,6 +201,14 @@ impl Shared { self.chaos_until_ms.fetch_max(until, Ordering::Relaxed); } + /// Recovery time the time-based checks allow during a disruptive window: the configured + /// slack, plus the completion stall while one has been run, since claiming and promotion + /// wait on the stalled tables for its whole length. + pub fn chaos_slack(&self) -> std::time::Duration { + let stalled = self.memory.stalled_for(); + self.config.chaos_slack + stalled + } + pub fn chaos_active(&self) -> bool { self.now_ms() < self.chaos_until_ms.load(Ordering::Relaxed) } diff --git a/docs/test-plan.md b/docs/test-plan.md index 30e96fc3a..b507751c0 100644 --- a/docs/test-plan.md +++ b/docs/test-plan.md @@ -342,6 +342,7 @@ The oracle checks, while the load runs and again after it stops: - a cancelled job nobody asked to cancel, an operator cancel that overwrote a completion, and a handler that started after its operator cancel committed - a queue with claimable jobs and idle workers whose claim cursor stops advancing (stall), and scheduled jobs overdue beyond a threshold (promotion stall) - a lost job: enqueued, never terminal after the system quiesces, and not removed by an operator action; a cron fire that produced more than one job +- resident memory of the soak process that keeps growing with executions after the warm-up (`memory_growth`; see `AWA_SOAK_MEMORY_SLOPE_MIB_PER_100K`) - shard operations: every job enqueued before, during and after a shard-count change runs and completes exactly once (a repeat is allowed only after a stale-heartbeat rescue and is reported as a finding), every shard drains including those above a lowered count, claim cursors keep advancing on every shard that has backlog while workers are idle (the per-shard lag query from ADR-025), `ordering_key` jobs start in order per key within a window in which every runtime had observed the shard count, and no role raises a permission-denied or missing-lane error (found in the server log, with the role, lane, sequence owner and statement; needs `AWA_SOAK_PG_CONTAINER` or `AWA_SOAK_PG_LOG`, and `log_line_prefix` containing `%u@%a`, as `scripts/persona-soak.sh` sets) - claim and abandon: after each SIGKILL the rescue backlog (open claims whose last heartbeat or claim time is older than the staleness window, or whose deadline passed, counted every few seconds from the claim tables and leases) clears within `AWA_SOAK_ABANDON_BOUND_SECONDS`, with the age and count of the oldest unrescued expired claim sampled into `abandon-backlog.csv` and the time to clear reported per kill; the claim ring does not stay still for longer than the bound once the slot it must enter next holds no claim whose owner is alive (a ring pinned by a live process is expected); no job is run again while the process that held it is alive (every start is recorded with the process that ran it, and a repeat is allowed only after that process was killed); no attempt exceeds `max_attempts`, no job is dead-lettered early, and a job abandoned on its last attempt ends in the DLQ - a transaction the Python application rolled back that left a job in the store, and database sessions left stranded by the Python `migrate()` path @@ -361,6 +362,8 @@ Environment: | `AWA_SOAK_ABANDON_CLAIMS`, `AWA_SOAK_ABANDON_KILL_SECONDS`, `AWA_SOAK_ABANDON_KILLS` | the claim-and-abandon persona's scale, independent of `AWA_SOAK_SCALE`: claims each killed process holds (default 4000; 25,000 to 70,000 reproduces a large backlog), seconds between kills (default 180, long enough for the process to claim everything and heartbeat for a while) and how many processes are killed (default 4; the last kill is the end of the load phase if that comes first) | | `AWA_SOAK_ABANDON_BOUND_SECONDS`, `AWA_SOAK_ABANDON_RESTART_SECONDS` | the bound after a kill within which the rescue backlog must clear and the claim ring must be free to rotate (default 120), and the pause before the replacement process starts (default 3) | | `AWA_SOAK_HEARTBEAT_SECONDS`, `AWA_SOAK_STALENESS_SECONDS` | heartbeat interval (default 2) and heartbeat staleness after which a claim may be rescued (default 12) of every runtime, including the claim-and-abandon process; the library's own defaults are 30 and 90. The claim-and-abandon persona also records the duration of each heartbeat pass of its process against the number of handlers running, read from the timestamps the heartbeats write (`abandon-heartbeat-passes.csv`, and percentiles by handler count in the report) | +| `AWA_SOAK_MEMORY_SLOPE_MIB_PER_100K`, `AWA_SOAK_MEMORY_WARMUP_SECONDS`, `AWA_SOAK_MEMORY_MIN_EXECUTIONS`, `AWA_SOAK_MEMORY_SAMPLE_SECONDS` | the memory growth gate. The resident memory of the soak process, which hosts every Rust runtime, is sampled every `AWA_SOAK_MEMORY_SAMPLE_SECONDS` (default 5) into `memory.csv` with the executions so far. After `AWA_SOAK_MEMORY_WARMUP_SECONDS` of load (default 120), the least-squares slope of resident memory over executions must stay below `AWA_SOAK_MEMORY_SLOPE_MIB_PER_100K` (default 64 MiB per 100k executions), or the run fails with `memory_growth`; with fewer than `AWA_SOAK_MEMORY_MIN_EXECUTIONS` executions after the warm-up (default 20000), or more than 1.25 jobs enqueued per execution over that window, the gate is reported as not evaluated, since the harness's own ledger grows with every enqueued job and a host that cannot keep up would read that as a leak | +| `AWA_SOAK_COMPLETION_STALL_SECONDS` | halfway through the load, holds the queue-storage completion tables (`lease_claim_closures`, `receipt_completion_batches`) in `SHARE` mode for this many seconds, so completions cannot commit while claims and handlers carry on; resident memory may rise by at most `AWA_SOAK_COMPLETION_STALL_MEMORY_MIB` (default 128) from the stall until as long again after it (`memory_growth_under_stall`), and that window is left out of the growth slope. Stalls over about 45 s also trip the promotion-overdue check, since promotion waits on the same tables. Default 0, off | | `AWA_SOAK_RESCUE_INTERVAL_SECONDS` | heartbeat and deadline rescue interval of every runtime; unset keeps the harness's 3 s and 2 s, the library default for heartbeat rescue is 30 s | | `AWA_SOAK_QUEUE_ROTATE_MS`, `AWA_SOAK_CLAIM_ROTATE_MS`, `AWA_SOAK_PRUNE_LOCK_TIMEOUT_MS` | queue and claim ring rotation intervals and the prune lock timeout of every in-process runtime; unset keeps the library's 1000, 1000 and 10. Setting either interval fixes the other at 1000 unless it is set too. The python and claim-and-abandon processes keep the library defaults, so leave those personas out when comparing cadences | | `AWA_SOAK_SHARDOPS_SCALE`, `AWA_SOAK_SHARDOPS_SCHEMA_CREATE` | the shard operator's rate multiplier, independent of `AWA_SOAK_SCALE` (default 1), and whether its application roles may create objects in the schema (default 1: lanes are created on the fly; 0 withholds schema `CREATE` and the operator provisions every lane before raising a count) | diff --git a/scripts/persona-soak.sh b/scripts/persona-soak.sh index 8a1590469..bd10b9179 100755 --- a/scripts/persona-soak.sh +++ b/scripts/persona-soak.sh @@ -33,6 +33,15 @@ # AWA_SOAK_HEARTBEAT_SECONDS, AWA_SOAK_STALENESS_SECONDS # heartbeat interval and heartbeat staleness of every runtime (default 2 s # and 12 s, so rescues happen often; the library's are 30 s and 90 s) +# AWA_SOAK_MEMORY_SLOPE_MIB_PER_100K, AWA_SOAK_MEMORY_WARMUP_SECONDS, +# AWA_SOAK_MEMORY_MIN_EXECUTIONS, AWA_SOAK_MEMORY_SAMPLE_SECONDS +# memory growth gate: largest accepted growth of the soak process's +# resident memory per 100k executions after the warm-up (default 64 MiB, +# 120 s warm-up, evaluated from 20000 executions, sampled every 5 s) +# AWA_SOAK_COMPLETION_STALL_SECONDS, AWA_SOAK_COMPLETION_STALL_MEMORY_MIB +# halfway through the load, hold back queue-storage completions for this +# many seconds while claims continue (default 0: off); resident memory +# may rise by at most the given MiB over the stall (default 128) # SOAK_RESULTS_DIR output directory (default ./soak-results/) # SOAK_BASE_PORT first host port; engines use consecutive ports (default 15471) # SOAK_MAX_CONNECTIONS server max_connections (default 400)