haematite 0.7.0

//! Explicit-run TTL scheduler and plain-put decode measurement harness.
//!
//! This follows the two existing documented harness precedents:
//! `tests/lazy_shard_materialisation.rs::many_shard_open_timing_harness` and
//! `tests/multi_reader_writer.rs::multi_reader_writer_stress`. Like both, it is a
//! normal Rust test marked `#[ignore]`, documents its exact opt-in command, prints
//! raw observations with `--nocapture`, and never burdens the default suite.
//!
//! Run from the repository root:
//! `cargo test -p haematite expiry_scheduler_and_plain_put_measurement_harness -- --ignored --nocapture`
//!
//! Scheduler samples use the real threaded Beamr scheduler and call
//! `NativeContext::schedule` through `native::NativeDeadlineScheduler`, the exact
//! production adapter used by the shard actor. Synchronization is only a bounded
//! `recv_timeout` waiting for actual delivery; the harness never sleeps or polls.
//! The late-wall classification (latency >= 900 ms) counts observations in the
//! structurally certified ~1.024 s genus; it is reporting, not a pass threshold.

use std::error::Error;
use std::hint::black_box;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, SyncSender};
use std::time::{Duration, Instant};

use beamr::atom::AtomTable;
use beamr::module::ModuleRegistry;
use beamr::process::ExitReason;
use beamr::scheduler::{Scheduler, SchedulerConfig};
use beamr::{NativeContext, NativeHandler, NativeOutcome};

use super::ShardActor;
use super::native;
use crate::store::MemoryStore;
use crate::wal::{DurableWal, FsyncPolicy};

const DELIVERY_SAMPLES: usize = 128;
const WALL_SAMPLES: usize = 64;
const DECODE_SAMPLES: usize = 256;
const DELIVERY_WAIT: Duration = Duration::from_secs(3);
const LATE_WALL_FLOOR: Duration = Duration::from_millis(900);

#[derive(Clone, Copy)]
struct ProbePlan {
    delay: Duration,
    samples: usize,
    busy_scheduler: bool,
}

struct SchedulerProbe {
    plan: ProbePlan,
    sender: SyncSender<Duration>,
    armed_at: Option<Instant>,
    delivered: usize,
}

impl SchedulerProbe {
    fn new(plan: ProbePlan, sender: SyncSender<Duration>) -> Self {
        Self {
            plan,
            sender,
            armed_at: None,
            delivered: 0,
        }
    }

    fn arm(
        &mut self,
        context: &mut NativeContext<'_>,
    ) -> Result<(), super::expiry_index::ArmError> {
        let token = u64::try_from(self.delivered)
            .unwrap_or(u64::MAX)
            .saturating_add(1);
        self.armed_at = Some(Instant::now());
        native::schedule_probe(context, self.plan.delay, token)?;
        Ok(())
    }
}

impl NativeHandler for SchedulerProbe {
    fn handle(&mut self, context: &mut NativeContext<'_>) -> NativeOutcome {
        if self.armed_at.is_none() && self.arm(context).is_err() {
            return NativeOutcome::Stop(ExitReason::Error);
        }
        while let Some(message) = context.recv() {
            let Some(expected) = self
                .delivered
                .checked_add(1)
                .and_then(|value| i64::try_from(value).ok())
            else {
                return NativeOutcome::Stop(ExitReason::Error);
            };
            if message.as_small_int() != Some(expected) {
                return NativeOutcome::Stop(ExitReason::Error);
            }
            let Some(armed_at) = self.armed_at.take() else {
                return NativeOutcome::Stop(ExitReason::Error);
            };
            if self.sender.send(armed_at.elapsed()).is_err() {
                return NativeOutcome::Stop(ExitReason::Error);
            }
            self.delivered = self.delivered.saturating_add(1);
            if self.delivered == self.plan.samples {
                return NativeOutcome::Stop(ExitReason::Normal);
            }
            if self.arm(context).is_err() {
                return NativeOutcome::Stop(ExitReason::Error);
            }
        }
        NativeOutcome::Wait
    }
}

struct LoadHandler;

impl NativeHandler for LoadHandler {
    fn handle(&mut self, context: &mut NativeContext<'_>) -> NativeOutcome {
        for _ in 0..64 {
            let Some(message) = context.recv() else { break };
            black_box(message);
        }
        NativeOutcome::Wait
    }
}

fn scheduler() -> Result<Arc<Scheduler>, Box<dyn Error>> {
    let scheduler = Scheduler::new(
        SchedulerConfig {
            thread_count: Some(1),
            ..SchedulerConfig::default()
        },
        Arc::new(ModuleRegistry::new()),
    )
    .map_err(|message| -> Box<dyn Error> { message.into() })?;
    Ok(Arc::new(scheduler))
}

fn measure_scheduler(plan: ProbePlan) -> Result<Vec<u64>, Box<dyn Error>> {
    let scheduler = scheduler()?;
    let load_stop = Arc::new(AtomicBool::new(false));
    let load_thread = if plan.busy_scheduler {
        let load_pid = scheduler.spawn_native(Box::new(|| Box::new(LoadHandler)))?;
        let atoms = AtomTable::with_common_atoms();
        let wake = atoms.intern("haematite_ttl_probe_load");
        let producer_scheduler = Arc::clone(&scheduler);
        let producer_stop = Arc::clone(&load_stop);
        Some(std::thread::spawn(move || {
            while !producer_stop.load(Ordering::Acquire) {
                if !producer_scheduler.enqueue_atom_message(load_pid, wake) {
                    break;
                }
            }
        }))
    } else {
        None
    };
    let (sender, receiver) = mpsc::sync_channel(plan.samples);
    let factory = Box::new(move || {
        Box::new(SchedulerProbe::new(plan, sender.clone())) as Box<dyn NativeHandler>
    });
    let _pid = scheduler.spawn_native(factory)?;
    let result = (|| -> Result<Vec<u64>, Box<dyn Error>> {
        let mut raw = Vec::with_capacity(plan.samples);
        for _ in 0..plan.samples {
            let elapsed = receiver.recv_timeout(DELIVERY_WAIT)?;
            raw.push(u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX));
        }
        Ok(raw)
    })();
    load_stop.store(true, Ordering::Release);
    if let Some(load_thread) = load_thread {
        load_thread
            .join()
            .map_err(|_panic| "scheduler load producer panicked")?;
    }
    result
}

fn percentile(sorted: &[u64], numerator: usize, denominator: usize) -> u64 {
    let index = sorted.len().saturating_sub(1).saturating_mul(numerator) / denominator;
    sorted[index]
}

fn report(label: &str, raw: &[u64]) {
    let mut sorted = raw.to_vec();
    sorted.sort_unstable();
    let late = raw
        .iter()
        .filter(|value| **value >= u64::try_from(LATE_WALL_FLOOR.as_nanos()).unwrap_or(u64::MAX))
        .count();
    println!("{label}.raw_ns={raw:?}");
    println!(
        "{label}.summary=count={},min_ns={},p50_ns={},p95_ns={},p99_ns={},max_ns={},late_wall_count={late}",
        sorted.len(),
        sorted[0],
        percentile(&sorted, 50, 100),
        percentile(&sorted, 95, 100),
        percentile(&sorted, 99, 100),
        sorted[sorted.len() - 1]
    );
}

fn measure_plain_put(prior_ttl: bool) -> Result<Vec<u64>, Box<dyn Error>> {
    let dir = tempfile::tempdir()?;
    let wal = DurableWal::new(dir.path().join("decode-probe.wal"), FsyncPolicy::CommitOnly)?;
    let mut actor = ShardActor::new(wal);
    let store = MemoryStore::new();
    let mut raw = Vec::with_capacity(DECODE_SAMPLES);
    for sample in 0..DECODE_SAMPLES {
        let key = sample.to_be_bytes();
        let ttl = prior_ttl.then_some(Duration::from_secs(3_600));
        actor.put_with_ttl(key, b"prior", ttl, &store)?;
        let started = Instant::now();
        actor.put_with_ttl(
            black_box(&key),
            black_box(b"plain"),
            None,
            black_box(&store),
        )?;
        raw.push(u64::try_from(started.elapsed().as_nanos()).unwrap_or(u64::MAX));
    }
    Ok(raw)
}

/// Reports real Beamr delivery distributions, idle/load late-wall frequencies,
/// and end-to-end plain-put timings with a prior non-TTL versus TTL envelope.
#[test]
#[ignore = "timing harness; run explicitly with --ignored --nocapture"]
fn expiry_scheduler_and_plain_put_measurement_harness() -> Result<(), Box<dyn Error>> {
    println!(
        "probe.contract=beamr-0.13.0; relative one-shot; immediate small-int token; certified late wall ~1.024s+idle park; observed values are non-normative"
    );
    let due_now = measure_scheduler(ProbePlan {
        delay: Duration::ZERO,
        samples: DELIVERY_SAMPLES,
        busy_scheduler: false,
    })?;
    let short_future = measure_scheduler(ProbePlan {
        delay: Duration::from_millis(2),
        samples: DELIVERY_SAMPLES,
        busy_scheduler: false,
    })?;
    let wall_idle = measure_scheduler(ProbePlan {
        delay: Duration::from_millis(1),
        samples: WALL_SAMPLES,
        busy_scheduler: false,
    })?;
    let wall_busy = measure_scheduler(ProbePlan {
        delay: Duration::from_millis(1),
        samples: WALL_SAMPLES,
        busy_scheduler: true,
    })?;
    let plain_prior = measure_plain_put(false)?;
    let ttl_prior = measure_plain_put(true)?;

    report("scheduler.due_now", &due_now);
    report("scheduler.short_future_2ms", &short_future);
    report("scheduler.late_wall_idle_1ms", &wall_idle);
    report("scheduler.late_wall_busy_1ms", &wall_busy);
    report("plain_put.prior_plain", &plain_prior);
    report("plain_put.prior_ttl", &ttl_prior);
    let floor = u64::try_from(LATE_WALL_FLOOR.as_nanos()).unwrap_or(u64::MAX);
    let idle_late = wall_idle.iter().filter(|value| **value >= floor).count();
    let busy_late = wall_busy.iter().filter(|value| **value >= floor).count();
    println!(
        "scheduler.late_wall_prediction=certified_more_likely_under_load;idle={idle_late}/{};busy={busy_late}/{};observed_relation={}",
        wall_idle.len(),
        wall_busy.len(),
        match busy_late.cmp(&idle_late) {
            std::cmp::Ordering::Greater => "supports",
            std::cmp::Ordering::Equal => "inconclusive_equal",
            std::cmp::Ordering::Less => "refutes_in_this_run",
        }
    );
    Ok(())
}