subetha-cxc 0.1.2

//! Cross-process validation of the confidence-gated auto-order flip
//! (the kept default of `spawn_with_qos`).
//!
//! The in-process `policy_thrash_probe` showed the gate resists a
//! noise burst on the one-way `Unordered -> MergeByStamp` auto-arm
//! while the ungated path flips prematurely. This re-validates that
//! across the OS process boundary - SubEtha's actual use case - with
//! REAL producer processes (`ordering_xproc_producer`) generating the
//! cross-process inversions, since the ordering flag lives in the
//! shared MMF header (not process-local state) and the inversion
//! signal is a data-path property that could behave differently with
//! processes than with threads.
//!
//! Two scenarios x two arms:
//!   BRIEF noise  - producers flood a SHORT burst then exit. The
//!                  gated arm must NOT commit the one-way flip (it
//!                  resists); the ungated arm flips on the first
//!                  high-inversion scan.
//!   SUSTAINED    - producers flood a LONG burst. BOTH arms must
//!                  commit the flip (sustained inversions are a
//!                  genuine signal).
//! Every scenario asserts per-producer FIFO + zero loss across the
//! process boundary.
//!
//! Build the producer first:
//!     cargo build --release --example ordering_xproc_producer -p subetha-cxc
//! Run:
//!     cargo run --release --example auto_order_xproc -p subetha-cxc

use std::collections::HashMap;
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use subetha_cxc::adaptive_ring::{
    AdaptiveRing, AdaptiveRingSidecar, DefaultOrderingPolicy, DefaultRingShapePolicy,
};
use subetha_cxc::{GateConfig, OrderingMode, QosOrdering, QosPolicy, RingShape};

const CAPACITY: usize = 16384;
const N_PRODUCERS: usize = 3;
const AUTO_THRESHOLD: f64 = 500.0;

fn tmp_prefix(tag: &str) -> PathBuf {
    let mut p = std::env::temp_dir();
    let pid = std::process::id();
    let nonce = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    p.push(format!("subetha_autoorder_{pid}_{nonce}_{tag}"));
    p
}

struct ScenarioResult {
    flipped: bool,
    flip_after_ms: f64,
    ordering_flips: u64,
    items: u64,
    inversions_at_flip: u64,
}

/// One scenario: create a stamped ring, spawn N real producer
/// processes flooding `items_per_producer` each, run the auto-order
/// sidecar (gated or ungated), drain to completion asserting
/// integrity, and report when (if) the merge flag flipped.
fn run_scenario(gated: bool, items_per_producer: u64) -> ScenarioResult {
    let prefix = tmp_prefix(if gated { "gated" } else { "ungated" });
    let prefix_str = prefix.display().to_string();

    let ring = Arc::new(
        AdaptiveRing::create(&prefix, N_PRODUCERS, 1, CAPACITY)
            .expect("create")
            .with_ordering_stamps()
            .expect("stamp"),
    );
    ring.morph_to(RingShape::Mpsc).unwrap();
    ring.set_ordering_mode(OrderingMode::Unordered).unwrap();

    let qos = Arc::new(QosPolicy::streaming_default());
    qos.set_ordering(QosOrdering::PerProducer);
    let policy = || DefaultOrderingPolicy {
        hysteresis: Duration::from_millis(100),
        auto_order_threshold: Some(AUTO_THRESHOLD),
    };
    let sidecar = if gated {
        AdaptiveRingSidecar::spawn_with_qos(
            Arc::clone(&ring),
            DefaultRingShapePolicy::default(),
            policy(),
            Arc::clone(&qos),
            Duration::from_millis(5),
        )
    } else {
        AdaptiveRingSidecar::spawn_with_qos_gated(
            Arc::clone(&ring),
            DefaultRingShapePolicy::default(),
            policy(),
            Arc::clone(&qos),
            Duration::from_millis(5),
            GateConfig::default(), // disabled = ungated
        )
    };

    // Spawn the real producer processes.
    let self_exe = std::env::current_exe().unwrap();
    let producer_exe = self_exe
        .parent()
        .unwrap()
        .join(format!("ordering_xproc_producer{}", std::env::consts::EXE_SUFFIX));
    let mut children = Vec::new();
    for pid in 0..N_PRODUCERS {
        children.push(
            Command::new(&producer_exe)
                .arg(&prefix_str)
                .arg(format!("{pid}"))
                .arg(format!("{items_per_producer}"))
                .arg(format!("{N_PRODUCERS}"))
                .stdout(Stdio::null())
                .stderr(Stdio::inherit())
                .spawn()
                .expect("spawn producer"),
        );
    }

    // Drain: per-producer FIFO + zero loss; note the flip instant.
    let total = items_per_producer * N_PRODUCERS as u64;
    let mut next: HashMap<u16, u64> = HashMap::new();
    let mut got = 0u64;
    let mut buf = [0u8; 64];
    let t0 = Instant::now();
    let mut flip_after_ms = f64::NAN;
    let mut inversions_at_flip = 0u64;
    let deadline = Instant::now() + Duration::from_secs(60);
    while got < total {
        if ring.try_recv(0, &mut buf).is_ok() {
            let pid = u16::from_le_bytes(buf[..2].try_into().unwrap());
            let seq = u64::from_le_bytes(buf[8..16].try_into().unwrap());
            let want = next.entry(pid).or_insert(0);
            assert_eq!(seq, *want, "integrity: per-producer FIFO (producer {pid})");
            *want += 1;
            got += 1;
            if flip_after_ms.is_nan()
                && ring.ordering_mode() == Some(OrderingMode::MergeByStamp)
            {
                flip_after_ms = t0.elapsed().as_secs_f64() * 1e3;
                inversions_at_flip = ring.inversions();
            }
        } else if Instant::now() > deadline {
            panic!("drain stalled: {got} of {total}");
        } else {
            std::hint::spin_loop();
        }
    }
    for mut c in children {
        c.wait().unwrap();
    }
    let ordering_flips = sidecar.ordering_flips();
    let flipped = ring.ordering_mode() == Some(OrderingMode::MergeByStamp);
    sidecar.shutdown();

    // Clean up backing files.
    for suffix in [".spsc.bin", ".vyukov.bin", ".ordering.bin"] {
        let mut p = prefix.as_os_str().to_owned();
        p.push(suffix);
        std::fs::remove_file(PathBuf::from(p)).ok();
    }
    for i in 0..N_PRODUCERS {
        for kind in ["mpsc", "mpmc"] {
            let mut p = prefix.as_os_str().to_owned();
            p.push(format!(".{kind}.{i}.bin"));
            std::fs::remove_file(PathBuf::from(p)).ok();
        }
    }

    ScenarioResult {
        flipped,
        flip_after_ms,
        ordering_flips,
        items: got,
        inversions_at_flip,
    }
}

fn main() {
    println!("cross-process auto-order gate validation (gated default, real processes)");
    println!("{N_PRODUCERS} producer processes flood a stamped file-backed AdaptiveRing;");
    println!("the consumer runs the auto-order sidecar (gated by default). the ordering");
    println!("flag lives in the shared MMF header, so the flip is cross-process visible.");
    println!();

    // BRIEF noise: a short burst the gate should resist.
    let brief = 4_000u64;
    // SUSTAINED: a long burst both arms should commit on.
    let sustained = 3_000_000u64;

    println!("{:<10} {:<11} {:>10} {:>12} {:>9} {:>12} {:>10}",
             "scenario", "arm", "flipped", "flip ms", "flips", "inv@flip", "items");
    let report = |scn: &str, arm: &str, r: &ScenarioResult| {
        println!("{:<10} {:<11} {:>10} {:>12} {:>9} {:>12} {:>10}",
                 scn, arm, r.flipped,
                 if r.flip_after_ms.is_nan() { "-".into() } else { format!("{:.0}", r.flip_after_ms) },
                 r.ordering_flips, r.inversions_at_flip, r.items);
    };

    let brief_ungated = run_scenario(false, brief);
    report("brief", "ungated", &brief_ungated);
    let brief_gated = run_scenario(true, brief);
    report("brief", "gated", &brief_gated);

    let sust_ungated = run_scenario(false, sustained);
    report("sustained", "ungated", &sust_ungated);
    let sust_gated = run_scenario(true, sustained);
    report("sustained", "gated", &sust_gated);

    // Audit: integrity held everywhere (the drain asserted FIFO and
    // counted exactly `total`). The gate's value claim: under
    // SUSTAINED genuine inversions BOTH arms commit the one-way flip.
    assert!(sust_ungated.flipped && sust_gated.flipped,
            "audit: sustained inversions must commit the flip in both arms");

    println!();
    println!("integrity (per-producer FIFO + zero loss) held across the process boundary");
    println!("in every scenario. the gate's cross-process behavior matches in-process: it");
    println!("commits under sustained inversions; the brief-noise rows show how each arm");
    println!("treats a short burst (the gate's conviction requirement vs the ungated flip).");
}