//! Stress harness 2 — Postgres chaos (gated on `DATABASE_URL`).
//!
//! Simulates a fleet of saga-runner "nodes" sharing one Postgres,
//! pushing a faulty multi-branch pipeline through churn:
//!
//! - N runner instances, each a fresh `PgSagaRunner` (a "node"). State
//!   lives entirely in Postgres, so any node can act on any instance —
//!   the whole point of the durable model.
//! - M saga instances triggered across random nodes.
//! - External responses delivered from RANDOM nodes (not the one that
//!   triggered) — proving cross-process resumption.
//! - Nodes are dropped and recreated mid-flight ("come and go").
//! - The SAME response delivered concurrently from two nodes — exactly
//!   one wins (OCC), no corruption, no double-processing.
//! - A fraction of instances hit a failing payment handler → full
//!   compensation cascade.
//! - Timeouts ticked from random nodes.
//!
//! Invariants asserted at the end:
//! - every instance reached a terminal state (Succeeded or Failed)
//! - succeeded instances have all steps Completed
//! - failed instances have their completed steps Compensated
//! - no `es_saga_awaits` rows leak for terminal instances
//!
//! Deterministic: a seeded LCG drives all "random" choices so a failure
//! reproduces.

#![allow(dead_code)]

use mire::{Aggregate, EventData};
use mire_sagas::{
    Emit, Output, PgSagaRunner, Req, Saga, SagaBuilder, SagaInstanceStatus, SagaStepStatus,
    SagaWorker, Start, State, comp, task,
};
use serde::{Deserialize, Serialize};
use sqlx::{PgPool, Row};

// =====================================================================
// Deterministic PRNG (LCG) — no external dep, reproducible.
// =====================================================================

struct Lcg(u64);
impl Lcg {
    fn next(&mut self) -> u64 {
        self.0 = self
            .0
            .wrapping_mul(6364136223846793005)
            .wrapping_add(1442695040888963407);
        self.0
    }
    fn below(&mut self, n: usize) -> usize {
        (self.next() >> 33) as usize % n
    }
    fn chance(&mut self, pct: u64) -> bool {
        self.next() % 100 < pct
    }
}

// =====================================================================
// Faulty pipeline saga.
//
//   trigger ─▶ inventory ─┐
//             shipping  ─┼─▶ charge ─▶ confirm(terminal)
//             (parallel) ─┘
//
// inventory + shipping: emit + await EXTERNAL responses (suspend).
// charge: in-process handler that FAILS for "doomed" orders → cascade.
// compensations release inventory + shipping.
// =====================================================================

#[derive(Default, Clone)]
struct Order;

#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.order-placed")]
struct OrderPlaced {
    id: String,
    doomed: bool,
    /// "Abandoned": no external worker will ever deliver this order's
    /// inventory/shipping responses, so its steps suspend with a SHORT
    /// timeout and are reaped by `tick_timeouts` → compensation. Healthy
    /// orders get a long timeout that never fires during a test run.
    #[serde(default)]
    abandoned: bool,
}

impl Aggregate for Order {
    type Event = OrderPlaced;
    fn stream_category() -> &'static str {
        "chaos-order"
    }
    fn apply(&mut self, _: &Self::Event) {}
}

#[derive(Default, Clone, Serialize, Deserialize)]
struct OrderState {
    id: String,
    doomed: bool,
    #[serde(default)]
    abandoned: bool,
}

/// Per-instance await timeout: abandoned orders expire in 1s (so the
/// timeout→compensation path runs continuously under soak load);
/// everything else gets an hour (never fires within a test run).
fn order_timeout(abandoned: bool) -> std::time::Duration {
    if abandoned {
        std::time::Duration::from_secs(1)
    } else {
        std::time::Duration::from_secs(3600)
    }
}

// --- inventory branch (external) ---
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.inv-req")]
struct InvReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.inv-held")]
struct InvHeld {
    hold: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct InvReserved {
    hold: String,
}
fn reserve_inv(State(s): State<OrderState>) -> Emit {
    Emit::event(InvReq { id: s.id.clone() }).timeout(order_timeout(s.abandoned))
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.inv-release")]
struct InvRelease {
    hold: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.inv-released")]
struct InvReleased;
fn release_inv(Output(o): Output<InvReserved>) -> Emit {
    Emit::event(InvRelease { hold: o.hold })
}
async fn handle_inv_release(Req(_): Req<InvRelease>) -> InvReleased {
    InvReleased
}

// --- shipping branch (external) ---
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.ship-req")]
struct ShipReq {
    id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.ship-held")]
struct ShipHeld {
    hold: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct ShipReserved {
    hold: String,
}
fn reserve_ship(State(s): State<OrderState>) -> Emit {
    Emit::event(ShipReq { id: s.id.clone() }).timeout(order_timeout(s.abandoned))
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.ship-release")]
struct ShipRelease {
    hold: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.ship-released")]
struct ShipReleased;
fn release_ship(Output(o): Output<ShipReserved>) -> Emit {
    Emit::event(ShipRelease { hold: o.hold })
}
async fn handle_ship_release(Req(_): Req<ShipRelease>) -> ShipReleased {
    ShipReleased
}

// --- charge step (in-process, can fail) ---
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.charge-req")]
struct ChargeReq {
    id: String,
    doomed: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.charge-ok")]
struct ChargeOk;
#[derive(Debug, Clone, Serialize, Deserialize, EventData)]
#[mire(event_type = "chaos.charge-declined")]
struct ChargeDeclined {
    reason: String,
}
impl std::fmt::Display for ChargeDeclined {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.reason)
    }
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Charged;
fn charge(
    State(s): State<OrderState>,
    Output(_inv): Output<InvReserved>,
    Output(_ship): Output<ShipReserved>,
) -> Emit {
    Emit::event(ChargeReq {
        id: s.id.clone(),
        doomed: s.doomed,
    })
}
async fn handle_charge(Req(r): Req<ChargeReq>) -> Result<ChargeOk, ChargeDeclined> {
    if r.doomed {
        Err(ChargeDeclined {
            reason: "card declined".into(),
        })
    } else {
        Ok(ChargeOk)
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Confirmed;
fn confirm(Output(_): Output<Charged>) -> Confirmed {
    Confirmed
}

// Shared topology. `tick_timeouts` is saga-WIDE (it expires every
// instance of a saga *name*), so any two tests that both call it must
// use DISTINCT saga names or one would reap the other's in-flight
// instances. A generic build helper keeps the pipeline identical while
// each test gets its own name.
fn build_order_topology<S>(b: SagaBuilder<S>) -> SagaBuilder<S>
where
    S: Saga<State = OrderState>,
{
    b.start_on::<Order, _>(|e: &OrderPlaced| {
        Some(Start {
            instance_id: e.id.clone(),
            state: OrderState {
                id: e.id.clone(),
                doomed: e.doomed,
                abandoned: e.abandoned,
            },
        })
    })
    .step(
        "inventory",
        task(reserve_inv)
            .on_response(|h: InvHeld| InvReserved { hold: h.hold })
            .compensate(comp(release_inv).on_response::<InvReleased>()),
    )
    .step(
        "shipping",
        task(reserve_ship)
            .on_response(|h: ShipHeld| ShipReserved { hold: h.hold })
            .compensate(comp(release_ship).on_response::<ShipReleased>()),
    )
    .step(
        "charge",
        task(charge)
            .after(["inventory", "shipping"])
            .on_response(|_: ChargeOk| Charged),
    )
    .step("confirm", task(confirm).after("charge").terminal_success())
    .effect(handle_charge)
    .effect(handle_inv_release)
    .effect(handle_ship_release)
}

// The one-shot tests (chaos / mayhem / scaling) share this saga name —
// only the chaos test ticks, and it does so at real `now` after its own
// instances are terminal, so they don't reap each other.
struct ChaosSaga;
impl Saga for ChaosSaga {
    type State = OrderState;
    fn name() -> &'static str {
        "chaos-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        build_order_topology(b)
    }
}

// The soak gets its OWN saga name: it ticks aggressively (including
// far-future ticks in its final drain), which must never touch another
// test's instances.
struct SoakSaga;
impl Saga for SoakSaga {
    type State = OrderState;
    fn name() -> &'static str {
        "soak-saga"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        build_order_topology(b)
    }
}

async fn maybe_store() -> Option<mire::EventStore> {
    let url = std::env::var("DATABASE_URL")
        .ok()
        .filter(|s| !s.is_empty())?;
    let pool = PgPool::connect(&url).await.expect("connect");
    let store = mire::EventStore::new(pool);
    store.migrate().await.expect("mire migrate");
    mire_sagas::migrate(&store).await.expect("saga migrate");
    Some(store)
}

fn new_node(store: &mire::EventStore) -> PgSagaRunner<ChaosSaga> {
    PgSagaRunner::new(
        ChaosSaga::build(SagaBuilder::<ChaosSaga>::for_saga()),
        store.clone(),
    )
    // Short processing-lease TTL so a crashed node's instance lock lapses
    // quickly and reconciliation can take over within the test's budget.
    // Handlers here are instant, so a live call never approaches this.
    .with_lease_ttl(std::time::Duration::from_millis(750))
}

fn new_soak_node(store: &mire::EventStore) -> PgSagaRunner<SoakSaga> {
    PgSagaRunner::new(
        SoakSaga::build(SagaBuilder::<SoakSaga>::for_saga()),
        store.clone(),
    )
}

// Recovery-chaos saga: its OWN name so its aggressive worker-driven timeout
// ticking never reaps another test's instances (see the note above).
struct RecoverChaosSaga;
impl Saga for RecoverChaosSaga {
    type State = OrderState;
    fn name() -> &'static str {
        "recover-chaos"
    }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        build_order_topology(b)
    }
}

// Sharded runner with a SHORT instance-lease TTL so a hard-aborted worker's
// per-instance lock lapses fast and a peer's timeout/recovery tick can take
// over within the test budget.
fn new_recover_node(store: &mire::EventStore, num_shards: u32) -> PgSagaRunner<RecoverChaosSaga> {
    PgSagaRunner::new(
        RecoverChaosSaga::build(SagaBuilder::<RecoverChaosSaga>::for_saga()),
        store.clone(),
    )
    .with_num_shards(num_shards)
    .with_lease_ttl(std::time::Duration::from_millis(500))
}

// =====================================================================
// The chaos run.
// =====================================================================

#[tokio::test]
async fn chaos_fleet_drives_faulty_pipeline_through_churn() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    let mut rng = Lcg(0x5151_2323_8787_dead);
    const NUM_INSTANCES: usize = 40;
    let run_tag = uuid::Uuid::new_v4().to_string();

    // A pool of "nodes". We rebuild entries mid-run to simulate churn.
    let mut nodes: Vec<PgSagaRunner<ChaosSaga>> = (0..4).map(|_| new_node(&store)).collect();

    // Track which instances we created + whether they're doomed.
    let mut ids: Vec<(String, bool)> = Vec::new();

    // 1. Trigger NUM_INSTANCES across random nodes; ~30% doomed.
    for i in 0..NUM_INSTANCES {
        let doomed = rng.chance(30);
        let id = format!("{run_tag}-{i}");
        let node = &nodes[rng.below(nodes.len())];
        node.trigger::<Order>(&OrderPlaced {
            id: id.clone(),
            doomed,
            abandoned: false,
        })
        .await
        .expect("trigger");
        ids.push((id, doomed));

        // Churn: occasionally replace a node mid-triggering.
        if rng.chance(20) {
            let victim = rng.below(nodes.len());
            nodes[victim] = new_node(&store);
        }
    }

    // 2. Deliver inventory + shipping responses from random nodes, in
    //    random order, with churn interleaved. Each instance suspends
    //    at inventory + shipping (external). Charge is in-process so it
    //    fires as soon as both complete.
    //
    //    Build a shuffled work list of (id, branch) deliveries.
    let mut work: Vec<(String, u8)> = Vec::new();
    for (id, _) in &ids {
        work.push((id.clone(), 0)); // inventory
        work.push((id.clone(), 1)); // shipping
    }
    // Fisher-Yates with the LCG.
    for i in (1..work.len()).rev() {
        let j = rng.below(i + 1);
        work.swap(i, j);
    }

    for (id, branch) in &work {
        let node = &nodes[rng.below(nodes.len())];
        let consumed = match branch {
            0 => node
                .deliver(
                    id,
                    &InvHeld {
                        hold: format!("inv-{id}"),
                    },
                )
                .await
                .expect("deliver inv"),
            1 => node
                .deliver(
                    id,
                    &ShipHeld {
                        hold: format!("ship-{id}"),
                    },
                )
                .await
                .expect("deliver ship"),
            _ => unreachable!(),
        };
        assert!(consumed, "every branch delivery should be consumed once");

        // Churn during delivery.
        if rng.chance(15) {
            let victim = rng.below(nodes.len());
            nodes[victim] = new_node(&store);
        }
    }

    // 3. Concurrent duplicate delivery: pick a still-running instance
    //    and fire the SAME (already-consumed) inventory response again
    //    from two nodes at once. Both must be safe no-ops (the await is
    //    already gone) — exactly-once consumption held.
    if let Some((dup_id, _)) = ids.first() {
        let n1 = new_node(&store);
        let n2 = new_node(&store);
        let inv_dup = InvHeld { hold: "dup".into() };
        let ship_dup = ShipHeld { hold: "dup".into() };
        let (r1, r2) = tokio::join!(n1.deliver(dup_id, &inv_dup), n2.deliver(dup_id, &ship_dup),);
        // Both already consumed earlier → both are no-ops (false), and
        // neither corrupts the instance.
        assert!(!r1.unwrap(), "duplicate inv delivery must be a no-op");
        assert!(!r2.unwrap(), "duplicate ship delivery must be a no-op");
    }

    // 4. Tick timeouts from a random node at real `now` (all awaits were
    //    delivered, so these instances are terminal and have no pending
    //    deadlines — it must simply run cleanly). We use real `now`, not
    //    a far-future time, so this saga-wide tick can't reap a
    //    concurrently-running sibling test's still-suspended instances.
    let now = chrono::Utc::now();
    nodes[rng.below(nodes.len())]
        .tick_timeouts(now)
        .await
        .expect("tick");

    // 5. Invariants. Use a fresh node to read everything back — proving
    //    a node that saw none of the action can still observe correct
    //    final state.
    let observer = new_node(&store);
    let mut succeeded = 0;
    let mut failed = 0;
    for (id, doomed) in &ids {
        let env = observer
            .load(id)
            .await
            .expect("load")
            .expect("instance exists");

        match env.status {
            SagaInstanceStatus::SucceededTerminal => {
                assert!(!doomed, "doomed order {id} must not succeed");
                for step in ["inventory", "shipping", "charge", "confirm"] {
                    assert_eq!(
                        env.step_status(step),
                        SagaStepStatus::Completed,
                        "{id}: step {step} should be Completed on success"
                    );
                }
                succeeded += 1;
            }
            SagaInstanceStatus::FailedTerminal => {
                assert!(doomed, "non-doomed order {id} must not fail");
                // charge failed → inventory + shipping compensated.
                assert_eq!(env.step_status("charge"), SagaStepStatus::Failed, "{id}");
                assert_eq!(
                    env.step_status("inventory"),
                    SagaStepStatus::Compensated,
                    "{id}: inventory should be compensated"
                );
                assert_eq!(
                    env.step_status("shipping"),
                    SagaStepStatus::Compensated,
                    "{id}: shipping should be compensated"
                );
                failed += 1;
            }
            other => panic!("{id} not terminal: {other:?}"),
        }

        // No await rows leak for a terminal instance.
        let leaked: i64 = sqlx::query(
            "SELECT count(*) AS c FROM es_saga_awaits \
             WHERE saga_name = 'chaos-saga' AND instance_id = $1",
        )
        .bind(id)
        .fetch_one(store.pool())
        .await
        .unwrap()
        .get("c");
        assert_eq!(leaked, 0, "{id}: await rows must not leak after terminal");
    }

    assert_eq!(succeeded + failed, NUM_INSTANCES);
    assert!(
        succeeded > 0 && failed > 0,
        "run should exercise both paths"
    );
    eprintln!("chaos run: {succeeded} succeeded, {failed} compensated, through node churn");
}

// =====================================================================
// Concurrent chaos: genuine parallelism. Many nodes deliver the SAME
// instances' responses simultaneously via tokio::spawn — exercising
// real OCC contention (inv + ship for one instance racing to enable
// charge) and duplicate-delivery races at scale. The deliver()
// OCC-retry loop must make every branch land exactly once and every
// instance reach the correct terminal state.
// =====================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn concurrent_fleet_races_deliveries_with_occ_contention() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    const NUM_INSTANCES: usize = 60;
    let run_tag = uuid::Uuid::new_v4().to_string();
    let mut rng = Lcg(0xC0FF_EE12_3456_7890);

    // Trigger all instances (sequential; ~40% doomed).
    let trigger_node = new_node(&store);
    let mut ids: Vec<(String, bool)> = Vec::new();
    for i in 0..NUM_INSTANCES {
        let doomed = rng.chance(40);
        let id = format!("conc-{run_tag}-{i}");
        trigger_node
            .trigger::<Order>(&OrderPlaced {
                id: id.clone(),
                doomed,
                abandoned: false,
            })
            .await
            .expect("trigger");
        ids.push((id, doomed));
    }

    // Fire ALL inventory + shipping deliveries concurrently, each from
    // its own fresh node. For a third of instances, fire a DUPLICATE
    // of each branch too — so 2 or 3 nodes race the same await. Exactly
    // one delivery per (instance, branch) may consume; the rest must be
    // safe no-ops, and no instance may be corrupted.
    let mut handles = Vec::new();
    for (i, (id, _)) in ids.iter().enumerate() {
        let dup = i % 3 == 0;
        let copies = if dup { 2 } else { 1 };
        for _ in 0..copies {
            // inventory
            {
                let store = store.clone();
                let id = id.clone();
                handles.push(tokio::spawn(async move {
                    let node = new_node(&store);
                    node.deliver(
                        &id,
                        &InvHeld {
                            hold: format!("inv-{id}"),
                        },
                    )
                    .await
                    .map_err(|e| format!("inv {id}: {e:?}"))
                }));
            }
            // shipping
            {
                let store = store.clone();
                let id = id.clone();
                handles.push(tokio::spawn(async move {
                    let node = new_node(&store);
                    node.deliver(
                        &id,
                        &ShipHeld {
                            hold: format!("ship-{id}"),
                        },
                    )
                    .await
                    .map_err(|e| format!("ship {id}: {e:?}"))
                }));
            }
        }
    }

    // Join everything. Each deliver must succeed (Ok) — true if it
    // consumed the await, false if a concurrent copy beat it. No errors
    // allowed: the OCC-retry loop must absorb all contention.
    let mut consumed_count = 0usize;
    for h in handles {
        match h.await.expect("task panicked") {
            Ok(true) => consumed_count += 1,
            Ok(false) => {} // a racing duplicate beat this one — fine
            Err(e) => panic!("deliver errored under contention: {e}"),
        }
    }
    // Exactly 2 consumptions per instance (one inv, one ship), no matter
    // how many duplicate copies raced.
    assert_eq!(
        consumed_count,
        NUM_INSTANCES * 2,
        "each instance's two branches must each be consumed exactly once"
    );

    // Verify every instance reached the correct terminal state.
    let observer = new_node(&store);
    let mut succeeded = 0;
    let mut failed = 0;
    for (id, doomed) in &ids {
        let env = observer.load(id).await.unwrap().unwrap();
        match env.status {
            SagaInstanceStatus::SucceededTerminal => {
                assert!(!doomed, "{id}: doomed must not succeed");
                assert_eq!(env.step_status("confirm"), SagaStepStatus::Completed);
                succeeded += 1;
            }
            SagaInstanceStatus::FailedTerminal => {
                assert!(doomed, "{id}: non-doomed must not fail");
                assert_eq!(env.step_status("inventory"), SagaStepStatus::Compensated);
                assert_eq!(env.step_status("shipping"), SagaStepStatus::Compensated);
                failed += 1;
            }
            other => panic!("{id} not terminal: {other:?}"),
        }
        // No await leak.
        let leaked: i64 = sqlx::query(
            "SELECT count(*) AS c FROM es_saga_awaits \
             WHERE saga_name = 'chaos-saga' AND instance_id = $1",
        )
        .bind(id)
        .fetch_one(store.pool())
        .await
        .unwrap()
        .get("c");
        assert_eq!(leaked, 0, "{id}: await rows leaked");
    }

    assert_eq!(succeeded + failed, NUM_INSTANCES);
    assert!(succeeded > 0 && failed > 0);
    eprintln!(
        "concurrent chaos: {succeeded} succeeded, {failed} compensated, \
         {consumed_count} branch-consumptions under parallel OCC contention"
    );
}

// =====================================================================
// Mayhem: the most hostile environment we can simulate without the
// (not-yet-built) lease worker loop.
//
//   - A fluctuating pool of worker tasks ("nodes") that come and go.
//   - Nodes are asked to stop gracefully (a stop flag) but some are
//     ABORTED mid-operation instead (tokio JoinHandle::abort) — a hard
//     crash, possibly mid-deliver / mid-DB-transaction.
//   - Injected latency: nodes sleep random amounts (slow workers), and
//     some "take a while to wake up" (startup delay before first act).
//   - Failures: ~25% of orders are doomed (charge declines → cascade);
//     workers also randomly skip work (simulating transient inability).
//   - Duplicate / out-of-order deliveries are the norm, not the
//     exception — every node hammers a shared work list.
//
// Correctness is guaranteed by two pillars, not by the chaos behaving:
//   1. `deliver` is idempotent (OCC + await-consumption) — duplicate
//      and racing deliveries are safe.
//   2. A final single-threaded reconciliation sweep delivers every
//      (instance, branch) once more, guaranteeing completion regardless
//      of how much work the chaotic phase dropped on the floor.
//
// We then assert every instance reached the correct terminal state and
// print throughput/latency so we can watch it scale across runs.
// =====================================================================

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn mayhem_crashing_nodes_latency_and_failures() {
    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    const NUM_INSTANCES: usize = 80;
    let run_tag = uuid::Uuid::new_v4().to_string();
    let mut rng = Lcg(0xDEAD_BEEF_CAFE_F00D);

    // 1. Trigger all instances up front (mix of doomed).
    let trig = new_node(&store);
    let mut ids: Vec<(String, bool)> = Vec::new();
    for i in 0..NUM_INSTANCES {
        let doomed = rng.chance(25);
        let id = format!("mayhem-{run_tag}-{i}");
        trig.trigger::<Order>(&OrderPlaced {
            id: id.clone(),
            doomed,
            abandoned: false,
        })
        .await
        .expect("trigger");
        ids.push((id, doomed));
    }

    // Shared work list: every (instance, branch) that an external worker
    // must eventually deliver. Workers pick at random; duplicates are
    // safe (idempotent deliver).
    let work: Arc<Vec<(String, u8)>> = Arc::new(
        ids.iter()
            .flat_map(|(id, _)| [(id.clone(), 0u8), (id.clone(), 1u8)])
            .collect(),
    );

    // Metrics.
    let deliveries_attempted = Arc::new(AtomicU64::new(0));
    let deliveries_consumed = Arc::new(AtomicU64::new(0));
    let nodes_crashed = Arc::new(AtomicU64::new(0));
    let nodes_retired = Arc::new(AtomicU64::new(0));

    let start = Instant::now();

    // 2. Chaotic phase: spawn a churning pool of worker "nodes". Each
    //    node loops delivering random work items with latency; the
    //    supervisor randomly aborts some (crash) and lets others finish
    //    (graceful retire), continuously spawning replacements.
    const CHURN_ROUNDS: usize = 40;
    const POOL: usize = 6;

    // A worker node: wakes up (maybe slowly), then delivers random work
    // items with inter-op latency until told to stop or aborted.
    fn spawn_node(
        store: mire::EventStore,
        work: Arc<Vec<(String, u8)>>,
        seed: u64,
        attempted: Arc<AtomicU64>,
        consumed: Arc<AtomicU64>,
        retired: Arc<AtomicU64>,
        ops_before_retire: u64,
    ) -> tokio::task::JoinHandle<()> {
        tokio::spawn(async move {
            let mut rng = Lcg(seed);
            // "Takes a while to wake up."
            tokio::time::sleep(Duration::from_millis(rng.below(25) as u64)).await;
            let node = new_node(&store);
            for _ in 0..ops_before_retire {
                // Inter-op latency (slow worker).
                tokio::time::sleep(Duration::from_millis(rng.below(8) as u64)).await;
                // Sometimes a worker just can't act this tick.
                if rng.chance(15) {
                    continue;
                }
                let (id, branch) = &work[rng.below(work.len())];
                attempted.fetch_add(1, Ordering::Relaxed);
                let res = match branch {
                    0 => {
                        node.deliver(
                            id,
                            &InvHeld {
                                hold: format!("inv-{id}"),
                            },
                        )
                        .await
                    }
                    _ => {
                        node.deliver(
                            id,
                            &ShipHeld {
                                hold: format!("ship-{id}"),
                            },
                        )
                        .await
                    }
                };
                if let Ok(true) = res {
                    consumed.fetch_add(1, Ordering::Relaxed);
                }
                // Errors (e.g. OCC exhaustion under extreme contention)
                // are tolerated here — the final sweep reconciles. We
                // only assert no PANIC escaped.
            }
            retired.fetch_add(1, Ordering::Relaxed);
        })
    }

    let mut pool: Vec<tokio::task::JoinHandle<()>> = Vec::new();
    for s in 0..POOL {
        pool.push(spawn_node(
            store.clone(),
            work.clone(),
            0x1000 + s as u64,
            deliveries_attempted.clone(),
            deliveries_consumed.clone(),
            nodes_retired.clone(),
            rng.below(20) as u64 + 5,
        ));
    }

    for round in 0..CHURN_ROUNDS {
        // Let the pool churn briefly.
        tokio::time::sleep(Duration::from_millis(5)).await;

        // Pick a victim node. ~40% of the time: ABORT it mid-flight
        // (hard crash). Otherwise: leave it to retire gracefully.
        if !pool.is_empty() && rng.chance(40) {
            let victim = rng.below(pool.len());
            let handle = pool.swap_remove(victim);
            handle.abort(); // crash, possibly mid-deliver / mid-transaction
            nodes_crashed.fetch_add(1, Ordering::Relaxed);
        }

        // Reap finished nodes; spawn replacements to keep pressure up.
        pool.retain(|h| !h.is_finished());
        while pool.len() < POOL {
            pool.push(spawn_node(
                store.clone(),
                work.clone(),
                0x2000 + (round as u64) * 16 + pool.len() as u64,
                deliveries_attempted.clone(),
                deliveries_consumed.clone(),
                nodes_retired.clone(),
                rng.below(20) as u64 + 5,
            ));
        }
    }

    // Drain the pool: abort whatever's still running (final crash wave).
    for h in pool {
        h.abort();
    }

    let chaos_elapsed = start.elapsed();

    // 3. Reconciliation sweep — single node, deterministic, delivers
    //    every (instance, branch) once more. Idempotency makes already-
    //    consumed deliveries no-ops; anything the chaos dropped now
    //    completes. Retry around transient OCC just in case.
    let reconciler = new_node(&store);
    for (id, branch) in work.iter() {
        for _attempt in 0..5 {
            let r = match branch {
                0 => {
                    reconciler
                        .deliver(
                            id,
                            &InvHeld {
                                hold: format!("inv-{id}"),
                            },
                        )
                        .await
                }
                _ => {
                    reconciler
                        .deliver(
                            id,
                            &ShipHeld {
                                hold: format!("ship-{id}"),
                            },
                        )
                        .await
                }
            };
            if r.is_ok() {
                break;
            }
        }
    }

    // 4. Invariants: every instance terminal + correct, no await leak.
    let observer = new_node(&store);
    let mut succeeded = 0;
    let mut failed = 0;
    for (id, doomed) in &ids {
        let env = observer
            .load(id)
            .await
            .expect("load")
            .expect("instance exists");
        match env.status {
            SagaInstanceStatus::SucceededTerminal => {
                assert!(!doomed, "{id}: doomed must not succeed");
                assert_eq!(env.step_status("confirm"), SagaStepStatus::Completed);
                succeeded += 1;
            }
            SagaInstanceStatus::FailedTerminal => {
                assert!(doomed, "{id}: healthy must not fail");
                assert_eq!(
                    env.step_status("inventory"),
                    SagaStepStatus::Compensated,
                    "{id}"
                );
                assert_eq!(
                    env.step_status("shipping"),
                    SagaStepStatus::Compensated,
                    "{id}"
                );
                failed += 1;
            }
            other => panic!("{id} not terminal after reconciliation: {other:?}"),
        }
        let leaked: i64 = sqlx::query(
            "SELECT count(*) AS c FROM es_saga_awaits \
             WHERE saga_name = 'chaos-saga' AND instance_id = $1",
        )
        .bind(id)
        .fetch_one(store.pool())
        .await
        .unwrap()
        .get("c");
        assert_eq!(leaked, 0, "{id}: await rows leaked");
    }

    assert_eq!(succeeded + failed, NUM_INSTANCES);
    assert!(
        succeeded > 0 && failed > 0,
        "run should exercise both paths"
    );

    eprintln!(
        "mayhem: {NUM_INSTANCES} instances → {succeeded} ok / {failed} compensated | \
         chaos phase {chaos_elapsed:?} | nodes: {} crashed, {} retired | \
         deliveries: {} attempted, {} consumed | survived to a fully-correct terminal state",
        nodes_crashed.load(Ordering::Relaxed),
        nodes_retired.load(Ordering::Relaxed),
        deliveries_attempted.load(Ordering::Relaxed),
        deliveries_consumed.load(Ordering::Relaxed),
    );
}

// =====================================================================
// P3.6 — recovery under worker churn.
//
// The `mayhem` test above proves the DELIVERY path survives churn, but it
// leans on a final harness-driven reconciliation sweep + manual tick. This
// test removes BOTH crutches: a fleet of autonomous `SagaWorker`s is the ONLY
// thing driving the saga forward — their timeout tick fails the (never-
// delivered, abandoned) instances and their recovery tick (P3.3) heals any
// instance left half-driven by a hard-aborted peer.
//
// Every instance is `abandoned` (1s step deadline, no external delivery), so
// each must time out → compensation cascade → FailedTerminal entirely under
// worker control. Workers are ABORTED mid-flight (hard crash, possibly
// between `tick_one`'s StepFailed-claim commit and its cascade commit — the
// exact gap P3.3 closes). With no harness re-driving, the run COMPLETING is
// the proof: a stranded instance that recovery couldn't heal would hang past
// the budget and fail the test.
// =====================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn recovery_heals_stranded_instances_under_worker_churn() {
    use std::time::{Duration, Instant};
    use tokio_util::sync::CancellationToken;

    let Some(store) = maybe_store().await else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    const NUM_INSTANCES: usize = 40;
    const NUM_SHARDS: u32 = 4;
    let run_tag = uuid::Uuid::new_v4().to_string();
    let mut rng = Lcg(0x5EED_F00D_1234_ABCD);

    // 1. Trigger all instances as ABANDONED (so they can only finish via the
    //    worker timeout → cascade path, never via a delivery).
    let trig = new_recover_node(&store, NUM_SHARDS);
    let mut ids: Vec<String> = Vec::new();
    for i in 0..NUM_INSTANCES {
        let id = format!("recover-{run_tag}-{i}");
        trig.trigger::<Order>(&OrderPlaced {
            id: id.clone(),
            doomed: false,
            abandoned: true,
        })
        .await
        .expect("trigger");
        ids.push(id);
    }

    // 2. Spawn a churning fleet of autonomous workers. Each owns a fresh
    //    runner; the fleet self-partitions the 4 shards via shard leases.
    //    Fast ticks so timeouts/recovery fire well inside the budget.
    fn spawn_worker(store: mire::EventStore) -> (tokio::task::JoinHandle<()>, CancellationToken) {
        let cancel = CancellationToken::new();
        let c = cancel.clone();
        let handle = tokio::spawn(async move {
            let worker = SagaWorker::new(new_recover_node(&store, NUM_SHARDS))
                .poll_interval(Duration::from_millis(50))
                .shard_lease_ttl(Duration::from_secs(2))
                // Phase 7: recovery is lease-expiry-driven. A crash-stranded
                // instance's lease (500ms TTL on this node) lapses and the next
                // scan steals + re-drives it; a live holder keeps its lease via
                // heartbeat and is never re-driven out from under itself.
                .recovery_interval(Duration::from_millis(150));
            // A graceful cancel returns Ok; a hard abort just drops the future.
            let _ = worker.run(c).await;
        });
        (handle, cancel)
    }

    const POOL: usize = 4;
    let mut fleet: Vec<(tokio::task::JoinHandle<()>, CancellationToken)> =
        (0..POOL).map(|_| spawn_worker(store.clone())).collect();

    let start = Instant::now();
    let mut aborted = 0u64;

    // 3. Churn: repeatedly hard-ABORT a random worker (crash) and respawn a
    //    replacement, keeping the fleet at POOL. Run long enough to span the
    //    1s timeout + a few recovery cycles.
    for _ in 0..30 {
        tokio::time::sleep(Duration::from_millis(120)).await;
        if rng.chance(60) {
            let victim = rng.below(fleet.len());
            let (h, _c) = fleet.swap_remove(victim);
            h.abort(); // hard crash — possibly mid tick_one, between commits
            aborted += 1;
        }
        fleet.retain(|(h, _)| !h.is_finished());
        while fleet.len() < POOL {
            fleet.push(spawn_worker(store.clone()));
        }
    }

    // 4. Let the (still-churning-free) fleet drain. Poll for every instance
    //    terminal — driven ONLY by the workers (no manual tick / reconcile).
    let mut all_terminal = false;
    for _ in 0..400 {
        let remaining: i64 = sqlx::query(
            "SELECT count(*) AS c FROM es_saga_instances \
             WHERE saga_name = 'recover-chaos' AND instance_id LIKE $1 \
               AND status IN ('not-started','running')",
        )
        .bind(format!("recover-{run_tag}-%"))
        .fetch_one(store.pool())
        .await
        .unwrap()
        .get("c");
        if remaining == 0 {
            all_terminal = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    let elapsed = start.elapsed();

    // Gracefully stop the fleet.
    for (h, c) in &fleet {
        c.cancel();
        let _ = h;
    }

    assert!(
        all_terminal,
        "every instance must reach terminal via worker timeout+recovery alone \
         (no harness re-driving); some hung — recovery failed to heal a stranded instance"
    );

    // 5. Invariants: every instance FailedTerminal (abandoned → cascade), no
    //    await rows leaked.
    let observer = new_recover_node(&store, NUM_SHARDS);
    for id in &ids {
        let env = observer.load(id).await.unwrap().expect("instance exists");
        assert_eq!(
            env.status,
            SagaInstanceStatus::FailedTerminal,
            "{id}: abandoned instance must end FailedTerminal, got {:?}",
            env.status
        );
        let leaked: i64 = sqlx::query(
            "SELECT count(*) AS c FROM es_saga_awaits \
             WHERE saga_name = 'recover-chaos' AND instance_id = $1",
        )
        .bind(id)
        .fetch_one(store.pool())
        .await
        .unwrap()
        .get("c");
        assert_eq!(leaked, 0, "{id}: await rows leaked");
    }

    eprintln!(
        "recovery-churn: {NUM_INSTANCES} abandoned instances → all FailedTerminal in {elapsed:?} \
         | {aborted} worker crashes | driven purely by SagaWorker timeout+recovery (no harness)"
    );
}

// =====================================================================
// Scaling: how smoothly does the runner scale with load + concurrency?
//
// Drives an increasing number of healthy instances through the full
// external-suspend pipeline (inventory + shipping suspend, charge in-
// process, confirm) with a bounded-concurrency worker pool, and reports
// throughput (instances/sec) and per-deliver latency percentiles at
// each load level. No crashes here — this isolates steady-state
// scaling behaviour from the chaos suite.
//
// Not a pass/fail perf gate (hardware varies); it asserts correctness
// and prints a scaling table so regressions / scaling cliffs are
// visible across runs.
// =====================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn scaling_throughput_and_latency() {
    use sqlx::postgres::PgPoolOptions;
    use std::sync::Arc;
    use std::time::Instant;
    use tokio::sync::{Mutex, Semaphore};

    let Some(url) = std::env::var("DATABASE_URL").ok() else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    // A roomier pool than the sqlx default (10) so concurrency isn't
    // bottlenecked below the level we're trying to measure.
    const POOL_SIZE: u32 = 32;
    const CONCURRENCY: usize = 24;

    let pool = PgPoolOptions::new()
        .max_connections(POOL_SIZE)
        .connect(&url)
        .await
        .expect("connect");
    let store = mire::EventStore::new(pool.clone());
    store.migrate().await.expect("mire migrate");
    mire_sagas::migrate(&store).await.expect("saga migrate");

    // Pre-warm the connection pool so the first measured level reflects
    // steady-state, not lazy connection establishment. Fire POOL_SIZE
    // concurrent trivial queries to force every connection open.
    {
        let mut warm = Vec::new();
        for _ in 0..POOL_SIZE {
            let pool = pool.clone();
            warm.push(tokio::spawn(async move {
                let _ = sqlx::query("SELECT 1").execute(&pool).await;
            }));
        }
        for h in warm {
            let _ = h.await;
        }
    }

    eprintln!(
        "scaling (pool={POOL_SIZE}, concurrency={CONCURRENCY}):  \
         N    triggers/s   drive/s    deliver p50/p99 (µs)"
    );

    for &n in &[50usize, 100, 200] {
        let run_tag = uuid::Uuid::new_v4().to_string();

        // Trigger N healthy instances; time it (trigger throughput).
        let trig = new_node(&store);
        let ids: Vec<String> = (0..n).map(|i| format!("scale-{run_tag}-{i}")).collect();
        let t0 = Instant::now();
        for id in &ids {
            trig.trigger::<Order>(&OrderPlaced {
                id: id.clone(),
                doomed: false,
                abandoned: false,
            })
            .await
            .expect("trigger");
        }
        let trigger_secs = t0.elapsed().as_secs_f64();

        // Build the 2N deliveries and fire them with bounded concurrency,
        // recording per-deliver latency. Idempotent delivers + healthy
        // pipeline → once both branches of an instance land, charge +
        // confirm run inline → terminal.
        let work: Vec<(String, u8)> = ids
            .iter()
            .flat_map(|id| [(id.clone(), 0u8), (id.clone(), 1u8)])
            .collect();
        let sem = Arc::new(Semaphore::new(CONCURRENCY));
        let lat = Arc::new(Mutex::new(Vec::<u64>::with_capacity(work.len())));

        let t1 = Instant::now();
        let mut handles = Vec::new();
        for (id, branch) in work {
            let store = store.clone();
            let sem = sem.clone();
            let lat = lat.clone();
            handles.push(tokio::spawn(async move {
                let _permit = sem.acquire().await.unwrap();
                let node = new_node(&store);
                let d0 = Instant::now();
                let r = match branch {
                    0 => {
                        node.deliver(
                            &id,
                            &InvHeld {
                                hold: format!("inv-{id}"),
                            },
                        )
                        .await
                    }
                    _ => {
                        node.deliver(
                            &id,
                            &ShipHeld {
                                hold: format!("ship-{id}"),
                            },
                        )
                        .await
                    }
                };
                r.expect("deliver under scaling load");
                lat.lock().await.push(d0.elapsed().as_micros() as u64);
            }));
        }
        for h in handles {
            h.await.expect("task");
        }
        let drive_secs = t1.elapsed().as_secs_f64();

        // Latency percentiles.
        let mut samples = Arc::try_unwrap(lat).unwrap().into_inner();
        samples.sort_unstable();
        let pct = |p: f64| -> u64 {
            if samples.is_empty() {
                return 0;
            }
            let idx = ((samples.len() as f64 - 1.0) * p).round() as usize;
            samples[idx]
        };
        let p50 = pct(0.50);
        let p99 = pct(0.99);

        // Correctness: every instance terminal-success.
        let observer = new_node(&store);
        for id in &ids {
            let env = observer.load(id).await.unwrap().unwrap();
            assert_eq!(
                env.status,
                SagaInstanceStatus::SucceededTerminal,
                "{id} should be terminal-success under scaling load"
            );
        }

        eprintln!(
            "  {n:>4}   {:>9.0}   {:>7.0}    {p50:>6}/{p99}",
            n as f64 / trigger_secs,
            n as f64 / drive_secs,
        );
    }
}

// =====================================================================
// Soak: a continuously-running simulated environment, designed to run
// at full throttle for minutes-to-hours and prove the runner stays
// correct + bounded under sustained chaotic load.
//
// Unlike the one-shot chaos/mayhem tests, this models a LIVE system:
//   - A producer continuously triggers new sagas for the whole run.
//   - A churning worker pool ("nodes") delivers responses with latency,
//     crashes (tokio abort, mid-transaction), graceful retirement, and
//     slow wake-up — continuously respawned to keep pressure up.
//   - ~20% of sagas are doomed (fail mid-flight → compensation cascade).
//   - A periodic reaper reconciles + VERIFIES completed instances and
//     drops them, keeping memory bounded (proving no unbounded growth).
//   - A metrics line prints every report interval so you can watch
//     throughput, backlog, crash count, and error rate over time.
//
// Correctness rests on idempotent `deliver` + the reaper's reconciling
// sweep; the chaos only affects timing, never the eventual outcome.
//
// GATING: ignored by default (a soak is not a unit test). Runs only when
// BOTH are set:
//   MIRE_SOAK_SECONDS=<n>   how long to run at full throttle
//   DATABASE_URL=<url>      the target Postgres
// Invoke:
//   MIRE_SOAK_SECONDS=3600 DATABASE_URL=... \
//     cargo test -p mire-sagas --test stress_pg_chaos soak -- --ignored --nocapture
// =====================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore = "long-running soak; set MIRE_SOAK_SECONDS to run"]
async fn soak_simulated_environment() {
    use sqlx::postgres::PgPoolOptions;
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
    use std::sync::{Arc, Mutex};
    use std::time::{Duration, Instant};

    let Some(url) = std::env::var("DATABASE_URL").ok() else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    let soak_secs: u64 = std::env::var("MIRE_SOAK_SECONDS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(0);
    if soak_secs == 0 {
        eprintln!("skipping: set MIRE_SOAK_SECONDS=<n> to run the soak");
        return;
    }

    const POOL_SIZE: u32 = 32;
    const WORKERS: usize = 8;
    const REPORT_EVERY: Duration = Duration::from_secs(15);
    let producer_pace = Duration::from_millis(2); // ~500 triggers/s target

    let pool = PgPoolOptions::new()
        .max_connections(POOL_SIZE)
        .connect(&url)
        .await
        .expect("connect");
    let store = mire::EventStore::new(pool);
    store.migrate().await.expect("mire migrate");
    mire_sagas::migrate(&store).await.expect("saga migrate");

    // A "flaky link" store pointed at an unreachable address. Workers
    // route a fraction of their deliveries through it to model transient
    // connectivity loss + the latency of a failing connection. Every op
    // through it errors; the healthy reconciler recovers the work.
    let flaky = mire::EventStore::new(
        PgPoolOptions::new()
            .max_connections(4)
            .acquire_timeout(Duration::from_millis(100))
            .connect_lazy("postgres://mire:mire@127.0.0.1:1/mire")
            .expect("flaky lazy pool"),
    );

    let run_tag = uuid::Uuid::new_v4().to_string();
    let deadline = Instant::now() + Duration::from_secs(soak_secs);

    // Shared in-flight set: id -> (doomed, abandoned). The reaper drains
    // it; staying bounded across the whole run is itself a tested
    // property. Abandoned instances are never delivered — they're
    // reaped by tick_timeouts → compensation.
    let inflight: Arc<Mutex<HashMap<String, (bool, bool)>>> = Arc::new(Mutex::new(HashMap::new()));
    let seq = Arc::new(AtomicU64::new(0));
    let stop = Arc::new(AtomicBool::new(false));

    // Counters.
    let triggered = Arc::new(AtomicU64::new(0));
    let verified_ok = Arc::new(AtomicU64::new(0));
    let verified_failed = Arc::new(AtomicU64::new(0)); // doomed → charge declined
    let verified_timedout = Arc::new(AtomicU64::new(0)); // abandoned → timeout
    let crashes = Arc::new(AtomicU64::new(0));
    let deliver_errs = Arc::new(AtomicU64::new(0));
    let max_backlog = Arc::new(AtomicU64::new(0));
    // Per-deliver latency samples (µs) for successful healthy-path
    // deliveries, drained each report window so we can watch p50/p99
    // OVER TIME — tail-latency creep across a long run is the signal.
    let lat_us: Arc<Mutex<Vec<u64>>> = Arc::new(Mutex::new(Vec::new()));

    // ---- Producer: continuously trigger new sagas until the deadline.
    let producer = {
        let store = store.clone();
        let inflight = inflight.clone();
        let seq = seq.clone();
        let triggered = triggered.clone();
        let stop = stop.clone();
        let run_tag = run_tag.clone();
        tokio::spawn(async move {
            let node = new_soak_node(&store);
            let mut rng = Lcg(0xA11CE);
            while !stop.load(Ordering::Relaxed) {
                let i = seq.fetch_add(1, Ordering::Relaxed);
                let doomed = rng.chance(20);
                let abandoned = rng.chance(10); // ~10% are never delivered
                let id = format!("soak-{run_tag}-{i}");
                if node
                    .trigger::<Order>(&OrderPlaced {
                        id: id.clone(),
                        doomed,
                        abandoned,
                    })
                    .await
                    .is_ok()
                {
                    inflight.lock().unwrap().insert(id, (doomed, abandoned));
                    triggered.fetch_add(1, Ordering::Relaxed);
                }
                tokio::time::sleep(producer_pace).await;
            }
        })
    };

    // ---- Worker pool: deliver responses for random in-flight instances
    //      with latency. Spawned via a factory so the supervisor can
    //      crash + respawn them.
    fn spawn_worker(
        store: mire::EventStore,
        flaky: mire::EventStore,
        inflight: Arc<Mutex<HashMap<String, (bool, bool)>>>,
        stop: Arc<std::sync::atomic::AtomicBool>,
        deliver_errs: Arc<AtomicU64>,
        lat_us: Arc<Mutex<Vec<u64>>>,
        seed: u64,
    ) -> tokio::task::JoinHandle<()> {
        tokio::spawn(async move {
            let mut rng = Lcg(seed);
            tokio::time::sleep(Duration::from_millis(rng.below(30) as u64)).await; // slow wake
            let node = new_soak_node(&store);
            let flaky_node = new_soak_node(&flaky);
            while !stop.load(std::sync::atomic::Ordering::Relaxed) {
                tokio::time::sleep(Duration::from_millis(rng.below(6) as u64)).await;
                // Snapshot a random in-flight id, skipping abandoned ones
                // (those are meant to time out, not be delivered).
                let pick = {
                    let map = inflight.lock().unwrap();
                    if map.is_empty() {
                        None
                    } else {
                        let n = rng.below(map.len());
                        map.iter()
                            .nth(n)
                            .filter(|(_, (_, abandoned))| !abandoned)
                            .map(|(k, _)| k.clone())
                    }
                };
                let Some(id) = pick else {
                    tokio::time::sleep(Duration::from_millis(2)).await;
                    continue;
                };
                // ~15% of the time, route through the flaky (unreachable)
                // link — transient connectivity loss. Those ops error +
                // burn the connect timeout (latency); the reconciler
                // recovers the work.
                let use_flaky = rng.chance(15);
                let via = if use_flaky { &flaky_node } else { &node };
                let branch = rng.below(2);
                let t0 = std::time::Instant::now();
                let r = if branch == 0 {
                    via.deliver(
                        &id,
                        &InvHeld {
                            hold: format!("inv-{id}"),
                        },
                    )
                    .await
                } else {
                    via.deliver(
                        &id,
                        &ShipHeld {
                            hold: format!("ship-{id}"),
                        },
                    )
                    .await
                };
                match r {
                    Err(_) => {
                        deliver_errs.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    }
                    Ok(_) if !use_flaky => {
                        // Record latency of successful healthy-path
                        // deliveries only — the flaky link's timeout
                        // latency would swamp the real pipeline signal.
                        lat_us.lock().unwrap().push(t0.elapsed().as_micros() as u64);
                    }
                    Ok(_) => {}
                }
            }
        })
    }

    let mut workers: Vec<tokio::task::JoinHandle<()>> = (0..WORKERS)
        .map(|w| {
            spawn_worker(
                store.clone(),
                flaky.clone(),
                inflight.clone(),
                stop.clone(),
                deliver_errs.clone(),
                lat_us.clone(),
                0x5EED + w as u64,
            )
        })
        .collect();

    // ---- Supervisor + reaper + reporter, all on this task.
    let mut rng = Lcg(0xC0DE_F00D);
    let mut last_report = Instant::now();
    let mut last_tick = Instant::now();
    let mut respawn_seed = 0x9000u64;
    let reconciler = new_soak_node(&store);

    while Instant::now() < deadline {
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Periodically expire abandoned instances (their 1s deadline).
        // This is the timeout→compensation path running under load.
        if last_tick.elapsed() >= Duration::from_millis(750) {
            let _ = reconciler.tick_timeouts(chrono::Utc::now()).await;
            last_tick = Instant::now();
        }

        // Chaos: ~每 tick, maybe crash a worker (hard abort, possibly
        // mid-deliver) and respawn; occasionally let one retire and
        // respawn (graceful).
        if rng.chance(30) && !workers.is_empty() {
            let v = rng.below(workers.len());
            let h = workers.swap_remove(v);
            h.abort(); // crash
            crashes.fetch_add(1, Ordering::Relaxed);
        }
        workers.retain(|h| !h.is_finished());
        while workers.len() < WORKERS {
            respawn_seed += 1;
            workers.push(spawn_worker(
                store.clone(),
                flaky.clone(),
                inflight.clone(),
                stop.clone(),
                deliver_errs.clone(),
                lat_us.clone(),
                respawn_seed,
            ));
        }

        // Reaper: snapshot in-flight, reconcile + verify the done ones,
        // drop them. Keeps the in-flight set bounded.
        let snapshot: Vec<(String, bool, bool)> = {
            let map = inflight.lock().unwrap();
            let cur = map.len() as u64;
            max_backlog.fetch_max(cur, Ordering::Relaxed);
            map.iter().map(|(k, (d, a))| (k.clone(), *d, *a)).collect()
        };
        for (id, doomed, abandoned) in snapshot {
            // Abandoned instances are NOT delivered — they reach terminal
            // only via tick_timeouts (above). Just check + reap.
            if !abandoned {
                // Reconcile: idempotently deliver both branches (no-ops
                // if already consumed) so nothing stalls forever.
                let _ = reconciler
                    .deliver(
                        &id,
                        &InvHeld {
                            hold: format!("inv-{id}"),
                        },
                    )
                    .await;
                let _ = reconciler
                    .deliver(
                        &id,
                        &ShipHeld {
                            hold: format!("ship-{id}"),
                        },
                    )
                    .await;
            }
            if let Ok(Some(env)) = reconciler.load(&id).await {
                match env.status {
                    SagaInstanceStatus::SucceededTerminal => {
                        assert!(!doomed && !abandoned, "{id}: only healthy orders succeed");
                        verified_ok.fetch_add(1, Ordering::Relaxed);
                        inflight.lock().unwrap().remove(&id);
                    }
                    SagaInstanceStatus::FailedTerminal => {
                        assert!(doomed || abandoned, "{id}: only doomed/abandoned may fail");
                        if abandoned {
                            verified_timedout.fetch_add(1, Ordering::Relaxed);
                        } else {
                            verified_failed.fetch_add(1, Ordering::Relaxed);
                        }
                        inflight.lock().unwrap().remove(&id);
                    }
                    _ => {} // still running; reap next cycle
                }
            }
        }

        if last_report.elapsed() >= REPORT_EVERY {
            let remaining = deadline.saturating_duration_since(Instant::now());
            // Drain this window's latency samples and compute p50/p99.
            // Per-window (not cumulative) so tail-latency creep over the
            // run is visible rather than averaged away.
            let (p50, p99, n_samples) = {
                let mut s = std::mem::take(&mut *lat_us.lock().unwrap());
                s.sort_unstable();
                let pct = |p: f64| -> u64 {
                    if s.is_empty() {
                        0
                    } else {
                        s[(((s.len() - 1) as f64) * p).round() as usize]
                    }
                };
                (pct(0.50), pct(0.99), s.len())
            };
            eprintln!(
                "soak[{:>5}s left] triggered={} ok={} failed={} timedOut={} backlog={} maxBacklog={} crashes={} deliverErr={} | deliver p50/p99={}/{}µs (n={})",
                remaining.as_secs(),
                triggered.load(Ordering::Relaxed),
                verified_ok.load(Ordering::Relaxed),
                verified_failed.load(Ordering::Relaxed),
                verified_timedout.load(Ordering::Relaxed),
                inflight.lock().unwrap().len(),
                max_backlog.load(Ordering::Relaxed),
                crashes.load(Ordering::Relaxed),
                deliver_errs.load(Ordering::Relaxed),
                p50,
                p99,
                n_samples,
            );
            last_report = Instant::now();
        }
    }

    // ---- Shutdown: stop producer + workers, then drain everything to a
    //      verified terminal state.
    stop.store(true, Ordering::Relaxed);
    let _ = producer.await;
    for h in &workers {
        h.abort();
    }

    // Final drain: bring every remaining in-flight instance to a
    // verified terminal state. Delivered branches for non-abandoned
    // orders; abandoned orders are forced to expire via a tick with a
    // `now` well past their 1s deadline. Loop until the set is empty
    // (bounded — no new work is being produced).
    loop {
        let snapshot: Vec<(String, bool, bool)> = {
            let map = inflight.lock().unwrap();
            map.iter().map(|(k, (d, a))| (k.clone(), *d, *a)).collect()
        };
        if snapshot.is_empty() {
            break;
        }
        // Force all abandoned deadlines to expire now.
        let _ = reconciler
            .tick_timeouts(chrono::Utc::now() + chrono::Duration::seconds(5))
            .await;
        for (id, doomed, abandoned) in snapshot {
            if !abandoned {
                for _ in 0..5 {
                    let a = reconciler
                        .deliver(
                            &id,
                            &InvHeld {
                                hold: format!("inv-{id}"),
                            },
                        )
                        .await;
                    let b = reconciler
                        .deliver(
                            &id,
                            &ShipHeld {
                                hold: format!("ship-{id}"),
                            },
                        )
                        .await;
                    if a.is_ok() && b.is_ok() {
                        break;
                    }
                }
            }
            let env = reconciler.load(&id).await.unwrap().unwrap();
            match env.status {
                SagaInstanceStatus::SucceededTerminal => {
                    assert!(!doomed && !abandoned, "{id}: only healthy orders succeed");
                    verified_ok.fetch_add(1, Ordering::Relaxed);
                }
                SagaInstanceStatus::FailedTerminal => {
                    assert!(doomed || abandoned, "{id}: only doomed/abandoned may fail");
                    if abandoned {
                        verified_timedout.fetch_add(1, Ordering::Relaxed);
                    } else {
                        verified_failed.fetch_add(1, Ordering::Relaxed);
                    }
                }
                other => panic!("{id}: not terminal after final drain: {other:?}"),
            }
            inflight.lock().unwrap().remove(&id);
        }
    }

    let total_triggered = triggered.load(Ordering::Relaxed);
    let total_verified = verified_ok.load(Ordering::Relaxed)
        + verified_failed.load(Ordering::Relaxed)
        + verified_timedout.load(Ordering::Relaxed);
    eprintln!(
        "soak DONE: ran {soak_secs}s | triggered={total_triggered} verified={total_verified} \
         (ok={} failed={} timedOut={}) | maxBacklog={} crashes={} deliverErr={}",
        verified_ok.load(Ordering::Relaxed),
        verified_failed.load(Ordering::Relaxed),
        verified_timedout.load(Ordering::Relaxed),
        max_backlog.load(Ordering::Relaxed),
        crashes.load(Ordering::Relaxed),
        deliver_errs.load(Ordering::Relaxed),
    );

    // The timeout→compensation path must actually have exercised under
    // load — otherwise the abandoned-instance machinery is silently dead.
    assert!(
        verified_timedout.load(Ordering::Relaxed) > 0,
        "soak should have reaped abandoned instances via timeout"
    );

    // Transient connectivity loss must have been exercised — workers
    // routed a fraction of deliveries through the unreachable flaky
    // link, and those failures must NOT have lost any work.
    assert!(
        deliver_errs.load(Ordering::Relaxed) > 0,
        "soak should have hit the flaky link (transient connectivity failures)"
    );

    // Every triggered instance must have been verified to a correct
    // terminal state — nothing lost, nothing stuck — despite node
    // crashes, timeouts, AND transient connectivity loss.
    assert_eq!(
        total_verified, total_triggered,
        "every triggered saga must reach a verified terminal state"
    );
}

// =====================================================================
// DB connectivity chaos: a subset of "nodes" run against a FRAGILE
// connection pool (1 connection, 50ms acquire timeout). Under
// concurrency they constantly fail to even reach the database —
// modelling a flaky/saturated/partitioned DB link. Deliveries from
// fragile nodes fail cleanly (a transaction that never began can't
// corrupt anything); a healthy node then reconciles everything to a
// correct terminal state.
//
// Proves: degraded DB connectivity surfaces as clean errors, never
// partial/torn state, and the system fully recovers once a healthy
// connection drives the work.
// =====================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn db_connectivity_chaos_recovers() {
    use sqlx::postgres::PgPoolOptions;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::Duration;

    let Some(url) = std::env::var("DATABASE_URL").ok() else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    // Healthy pool — for triggering + final reconciliation.
    let healthy = mire::EventStore::new(
        PgPoolOptions::new()
            .max_connections(16)
            .connect(&url)
            .await
            .expect("healthy connect"),
    );
    healthy.migrate().await.expect("migrate");
    mire_sagas::migrate(&healthy).await.expect("saga migrate");

    // Fragile pool — points at an UNREACHABLE address via `connect_lazy`,
    // modelling a network partition / DB outage for these nodes: every
    // operation fails with a connection error, deterministically, on any
    // machine. A short acquire timeout makes the failure fast instead of
    // hanging on connect retries. (Port 1 is reserved/closed.)
    let bad_url = "postgres://mire:mire@127.0.0.1:1/mire";
    let fragile = mire::EventStore::new(
        PgPoolOptions::new()
            .max_connections(2)
            .acquire_timeout(Duration::from_millis(200))
            .connect_lazy(bad_url)
            .expect("fragile lazy pool"),
    );

    const N: usize = 40;
    let run_tag = uuid::Uuid::new_v4().to_string();

    // Trigger N healthy instances on the healthy pool.
    let trig = new_node(&healthy);
    let ids: Vec<String> = (0..N).map(|i| format!("dbc-{run_tag}-{i}")).collect();
    for id in &ids {
        trig.trigger::<Order>(&OrderPlaced {
            id: id.clone(),
            doomed: false,
            abandoned: false,
        })
        .await
        .expect("trigger");
    }

    // Fire every delivery from FRAGILE nodes, concurrently. Many will
    // fail with pool-acquire timeouts; that's the point.
    let work: Vec<(String, u8)> = ids
        .iter()
        .flat_map(|id| [(id.clone(), 0u8), (id.clone(), 1u8)])
        .collect();
    let ok = Arc::new(AtomicU64::new(0));
    let err = Arc::new(AtomicU64::new(0));

    // No semaphore: fire all deliveries at once so they pile onto the
    // fragile pool's single connection and contend for it.
    let mut handles = Vec::new();
    for (id, branch) in work {
        let fragile = fragile.clone();
        let ok = ok.clone();
        let err = err.clone();
        handles.push(tokio::spawn(async move {
            let node = new_node(&fragile);
            let r = if branch == 0 {
                node.deliver(
                    &id,
                    &InvHeld {
                        hold: format!("inv-{id}"),
                    },
                )
                .await
            } else {
                node.deliver(
                    &id,
                    &ShipHeld {
                        hold: format!("ship-{id}"),
                    },
                )
                .await
            };
            match r {
                Ok(_) => ok.fetch_add(1, Ordering::Relaxed),
                Err(_) => err.fetch_add(1, Ordering::Relaxed),
            };
        }));
    }
    for h in handles {
        h.await.expect("task");
    }

    let errors = err.load(Ordering::Relaxed);
    eprintln!(
        "db-connectivity: fragile deliveries → {} ok, {} failed (DB unreachable)",
        ok.load(Ordering::Relaxed),
        errors,
    );
    // The fragile pool must actually have induced connectivity failures,
    // otherwise the test isn't exercising what it claims.
    assert!(
        errors > 0,
        "fragile pool should have produced connectivity failures"
    );

    // Recovery: a healthy node reconciles every (instance, branch). OCC
    // retry + idempotency bring everything home despite the earlier
    // failures.
    let healer = new_node(&healthy);
    for id in &ids {
        for _ in 0..5 {
            let a = healer
                .deliver(
                    id,
                    &InvHeld {
                        hold: format!("inv-{id}"),
                    },
                )
                .await;
            let b = healer
                .deliver(
                    id,
                    &ShipHeld {
                        hold: format!("ship-{id}"),
                    },
                )
                .await;
            if a.is_ok() && b.is_ok() {
                break;
            }
        }
    }

    // Every instance reached terminal-success — connectivity loss caused
    // delay, never corruption or lost work.
    for id in &ids {
        let env = healer.load(id).await.unwrap().unwrap();
        assert_eq!(
            env.status,
            SagaInstanceStatus::SucceededTerminal,
            "{id} must recover to terminal-success after connectivity chaos"
        );
        // No torn state: every step Completed, no leaked awaits.
        for step in ["inventory", "shipping", "charge", "confirm"] {
            assert_eq!(
                env.step_status(step),
                SagaStepStatus::Completed,
                "{id}/{step}"
            );
        }
    }
    let leaked: i64 = sqlx::query_scalar(
        "SELECT count(*) FROM es_saga_awaits WHERE saga_name = 'chaos-saga' \
         AND instance_id LIKE $1",
    )
    .bind(format!("dbc-{run_tag}-%"))
    .fetch_one(healthy.pool())
    .await
    .unwrap();
    assert_eq!(leaked, 0, "no await rows may leak after recovery");
}

// =====================================================================
// Scaling curve: fix the instance count, vary the CONCURRENCY level,
// and chart throughput at each. Reveals how the runner scales with
// added parallelism and where it saturates (connection pool / DB).
// Complements `scaling_throughput_and_latency`, which varies load at
// fixed concurrency — this varies concurrency at fixed load.
//
// Not a pass/fail perf gate (hardware varies); it asserts correctness
// and prints the curve so scaling regressions are visible across runs.
// =====================================================================

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn scaling_concurrency_curve() {
    use sqlx::postgres::PgPoolOptions;
    use std::sync::Arc;
    use std::time::Instant;
    use tokio::sync::Semaphore;

    let Some(url) = std::env::var("DATABASE_URL").ok() else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };

    const POOL_SIZE: u32 = 32;
    const N: usize = 120; // instances per concurrency level

    let pool = PgPoolOptions::new()
        .max_connections(POOL_SIZE)
        .connect(&url)
        .await
        .expect("connect");
    let store = mire::EventStore::new(pool.clone());
    store.migrate().await.expect("mire migrate");
    mire_sagas::migrate(&store).await.expect("saga migrate");

    // Pre-warm the pool so the first level isn't penalised by lazy
    // connection establishment.
    {
        let mut warm = Vec::new();
        for _ in 0..POOL_SIZE {
            let pool = pool.clone();
            warm.push(tokio::spawn(async move {
                let _ = sqlx::query("SELECT 1").execute(&pool).await;
            }));
        }
        for h in warm {
            let _ = h.await;
        }
    }

    eprintln!("scaling curve (N={N} per level):  concurrency   drive/s   speedup-vs-C1");
    let mut baseline: Option<f64> = None;

    for &c in &[1usize, 2, 4, 8, 16, 32] {
        let run_tag = uuid::Uuid::new_v4().to_string();

        // Trigger N healthy instances (untimed; suspend on inv + ship).
        let trig = new_node(&store);
        let ids: Vec<String> = (0..N).map(|i| format!("cc-{c}-{run_tag}-{i}")).collect();
        for id in &ids {
            trig.trigger::<Order>(&OrderPlaced {
                id: id.clone(),
                doomed: false,
                abandoned: false,
            })
            .await
            .expect("trigger");
        }

        // Drive all 2N deliveries with exactly C-way concurrency.
        let work: Vec<(String, u8)> = ids
            .iter()
            .flat_map(|id| [(id.clone(), 0u8), (id.clone(), 1u8)])
            .collect();
        let sem = Arc::new(Semaphore::new(c));

        let t0 = Instant::now();
        let mut handles = Vec::new();
        for (id, branch) in work {
            let store = store.clone();
            let sem = sem.clone();
            handles.push(tokio::spawn(async move {
                let _permit = sem.acquire().await.unwrap();
                let node = new_node(&store);
                let r = if branch == 0 {
                    node.deliver(
                        &id,
                        &InvHeld {
                            hold: format!("inv-{id}"),
                        },
                    )
                    .await
                } else {
                    node.deliver(
                        &id,
                        &ShipHeld {
                            hold: format!("ship-{id}"),
                        },
                    )
                    .await
                };
                r.expect("deliver");
            }));
        }
        for h in handles {
            h.await.expect("task");
        }
        let drive_secs = t0.elapsed().as_secs_f64();
        let tput = N as f64 / drive_secs;

        // Correctness at every level.
        let observer = new_node(&store);
        for id in &ids {
            assert_eq!(
                observer.load(id).await.unwrap().unwrap().status,
                SagaInstanceStatus::SucceededTerminal,
                "{id} must be terminal-success at concurrency {c}"
            );
        }

        let speedup = match baseline {
            None => {
                baseline = Some(tput);
                1.0
            }
            Some(b) => tput / b,
        };
        eprintln!("  {c:>11}   {tput:>7.0}   {speedup:>5.2}x");
    }
}

/// Multi-NODE scale-out (weak scaling). The concurrency curve above shares
/// ONE pool, so it conflates concurrency with a fixed connection budget. A
/// real deployment is N processes, each with its own pool, driving
/// *disjoint* instances. This measures that: K independent nodes (own pool
/// each) each drive their own partition of `N_PER_NODE` instances. Because
/// the partitions are disjoint, the per-instance lease + OCC never contend
/// across nodes (principle 6), so aggregate throughput should grow ~linearly
/// with K until Postgres itself is the ceiling. Sub-linearity here points at
/// a *shared* bottleneck (PG write path, or — once they exist — unsharded
/// global scans), which is the thing to attack for scale-out.
///
/// Run with: `MIRE_SCALEOUT=1 cargo test --release -p mire-sagas --test
/// stress_pg_chaos scaling_node_fanout -- --nocapture`. Gated behind the env
/// var (and DATABASE_URL) so it doesn't run in the normal suite.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn scaling_node_fanout() {
    use sqlx::postgres::PgPoolOptions;
    use std::sync::Arc;
    use std::time::Instant;
    use tokio::sync::Semaphore;

    let Some(url) = std::env::var("DATABASE_URL").ok() else {
        eprintln!("skipping: DATABASE_URL not set");
        return;
    };
    if std::env::var("MIRE_SCALEOUT").is_err() {
        eprintln!("skipping scaling_node_fanout; set MIRE_SCALEOUT=1 to run");
        return;
    }

    const N_PER_NODE: usize = 60; // instances each node owns
    const PER_NODE_POOL: u32 = 16;
    const PER_NODE_CONC: usize = 12;

    // Migrate ONCE up front on a dedicated connection (principle 7/10) so the
    // per-node pools never race DDL.
    {
        let p = PgPoolOptions::new()
            .max_connections(1)
            .connect(&url)
            .await
            .expect("connect (migrate)");
        let s = mire::EventStore::new(p);
        s.migrate().await.expect("mire migrate");
        mire_sagas::migrate(&s).await.expect("saga migrate");
    }

    eprintln!(
        "node fan-out (weak scaling, {N_PER_NODE} instances/node, pool={PER_NODE_POOL}, \
         conc={PER_NODE_CONC}):"
    );
    eprintln!("  nodes   total   agg-drive/s   per-node/s   efficiency-vs-1node");
    let mut base_per_node: Option<f64> = None;

    for &k in &[1usize, 2, 4] {
        let run_tag = uuid::Uuid::new_v4().simple().to_string();

        // Build k independent nodes, each with its OWN pool (≈ separate
        // processes sharing the database).
        let mut nodes = Vec::with_capacity(k);
        for _ in 0..k {
            let pool = PgPoolOptions::new()
                .max_connections(PER_NODE_POOL)
                .connect(&url)
                .await
                .expect("connect (node)");
            let store = mire::EventStore::new(pool);
            nodes.push(Arc::new(PgSagaRunner::new(
                ChaosSaga::build(SagaBuilder::<ChaosSaga>::for_saga()),
                store,
            )));
        }

        // Disjoint partition: node j owns ids [j*N .. (j+1)*N). Trigger all
        // (untimed) — each instance suspends on inv + ship.
        let ids_per_node: Vec<Vec<String>> = (0..k)
            .map(|j| {
                (0..N_PER_NODE)
                    .map(|i| format!("nf-{k}-{run_tag}-{j}-{i}"))
                    .collect()
            })
            .collect();
        for (j, node) in nodes.iter().enumerate() {
            for id in &ids_per_node[j] {
                node.trigger::<Order>(&OrderPlaced {
                    id: id.clone(),
                    doomed: false,
                    abandoned: false,
                })
                .await
                .expect("trigger");
            }
        }

        // TIMED: every node drives its own partition's 2 deliveries/instance
        // concurrently (own pool, own semaphore). All nodes run at once.
        let t0 = Instant::now();
        let mut node_tasks = Vec::with_capacity(k);
        for (j, node) in nodes.iter().enumerate() {
            let node = node.clone();
            let ids = ids_per_node[j].clone();
            node_tasks.push(tokio::spawn(async move {
                let sem = Arc::new(Semaphore::new(PER_NODE_CONC));
                let mut hs = Vec::new();
                for id in ids {
                    for branch in 0u8..2 {
                        let node = node.clone();
                        let sem = sem.clone();
                        let id = id.clone();
                        hs.push(tokio::spawn(async move {
                            let _p = sem.acquire().await.unwrap();
                            if branch == 0 {
                                node.deliver(
                                    &id,
                                    &InvHeld {
                                        hold: format!("inv-{id}"),
                                    },
                                )
                                .await
                                .expect("deliver inv");
                            } else {
                                node.deliver(
                                    &id,
                                    &ShipHeld {
                                        hold: format!("ship-{id}"),
                                    },
                                )
                                .await
                                .expect("deliver ship");
                            }
                        }));
                    }
                }
                for h in hs {
                    h.await.expect("deliver task");
                }
            }));
        }
        for t in node_tasks {
            t.await.expect("node task");
        }
        let drive_secs = t0.elapsed().as_secs_f64();

        let total = (k * N_PER_NODE) as f64;
        let agg = total / drive_secs;
        let per_node = agg / k as f64;
        let efficiency = match base_per_node {
            None => {
                base_per_node = Some(per_node);
                1.0
            }
            Some(b) => per_node / b,
        };

        // Correctness: every instance terminal-success.
        for (j, node) in nodes.iter().enumerate() {
            for id in &ids_per_node[j] {
                assert_eq!(
                    node.load(id).await.unwrap().unwrap().status,
                    SagaInstanceStatus::SucceededTerminal,
                    "{id} must be terminal-success with {k} nodes"
                );
            }
        }

        eprintln!("  {k:>5}   {total:>5.0}   {agg:>11.0}   {per_node:>10.0}   {efficiency:>5.2}x");
    }
}
