kithara-platform 0.0.1-alpha5

use std::{
    env, fs,
    future::Future,
    io,
    path::PathBuf,
    pin::pin,
    process,
    sync::{
        Arc, Mutex,
        atomic::{AtomicUsize, Ordering},
    },
    task::{Context, Poll, Waker},
    thread,
    time::{Duration, Instant},
};

use cpu_time::ThreadTime;
use kithara_test_utils::kithara;
use tracing_subscriber::fmt::MakeWriter;

use super::{
    clock::force_cpu_elapsed,
    mode::{Mode, force_blanket_budget, force_log_path, force_mode, force_no_log_path},
    *,
};

const FIRST_LOG_FILE_ID: usize = 0;
const BLANKET_TEST_BUDGET_MS: u64 = 10;
const BLANKET_TEST_SPIN_MS: u64 = 50;
const CENSUS_LOG_BUDGET_MS: u64 = 10_000;
const CENSUS_LOG_SLEEP_MS: u64 = 1;
const WORK_TEST_BUDGET_MS: u64 = 10;
const WORK_TEST_SPIN_CPU_MS: u64 = 50;

static LOG_FILE_ID: AtomicUsize = AtomicUsize::new(FIRST_LOG_FILE_ID);

fn poll_once<F: Future>(fut: F) -> Poll<F::Output> {
    let mut fut = pin!(fut);
    let waker = Waker::noop();
    let mut cx = Context::from_waker(waker);
    fut.as_mut().poll(&mut cx)
}

fn temp_log_path(name: &str) -> PathBuf {
    let mut path = env::temp_dir();
    let id = LOG_FILE_ID.fetch_add(1, Ordering::Relaxed);
    path.push(format!(
        "kithara-no-block-{name}-{}-{id}.log",
        process::id()
    ));
    path
}

fn spin_for(d: Duration) {
    let start = Instant::now();
    while start.elapsed() < d {
        std::hint::spin_loop();
    }
}

/// Spend `cpu` of this thread's own CPU time.
///
/// A work budget reads CPU, and wall time buys an unknown share of it: a
/// loaded runner can hand a 50 ms wall spin less than the 10 ms of CPU the
/// budget is asking about, and the poll under test would then have spent
/// nothing to flag.
fn spin_cpu_for(cpu: Duration) {
    let start = ThreadTime::try_now().expect("thread CPU clock");
    while start.try_elapsed().expect("thread CPU clock") < cpu {
        std::hint::spin_loop();
    }
}

/// Drive one over-budget poll, which reports a single census observation.
fn census_once(task: &'static str) {
    let fut = watch_budget(task, CENSUS_LOG_BUDGET_MS, async {
        crate::thread::sleep(Duration::from_millis(CENSUS_LOG_SLEEP_MS));
    });
    let _ = poll_once(fut);
}

#[derive(Clone)]
struct TracingSink(Arc<Mutex<Vec<u8>>>);

impl io::Write for TracingSink {
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }

    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.0.lock().expect("tracing sink").extend_from_slice(buf);
        Ok(buf.len())
    }
}

impl<'a> MakeWriter<'a> for TracingSink {
    type Writer = Self;

    fn make_writer(&'a self) -> Self::Writer {
        self.clone()
    }
}

/// What `run` emits through `tracing`, as text.
///
/// The subscriber is thread-local, so it takes precedence over whichever one
/// the test harness installed globally and sees only this call.
fn capture_tracing(run: impl FnOnce()) -> String {
    let sink = Arc::new(Mutex::new(Vec::new()));
    let subscriber = tracing_subscriber::fmt()
        .with_ansi(false)
        .with_writer(TracingSink(Arc::clone(&sink)))
        .finish();
    tracing::subscriber::with_default(subscriber, run);
    let captured = sink.lock().expect("tracing sink").clone();
    String::from_utf8(captured).expect("tracing output is utf8")
}

#[kithara::test(native, flash(false))]
fn budget_flags_over_budget_poll() {
    force_mode(Mode::Panic);

    let caught = std::panic::catch_unwind(|| {
        let fut = watch_budget("spin_task", 10, async {
            spin_for(Duration::from_millis(50));
        });
        let _ = poll_once(fut);
    });
    let err = caught.expect_err("over-budget spin poll must panic");
    let msg = err.downcast_ref::<String>().expect("panic payload");
    assert!(msg.contains("[no_block]"), "got: {msg}");
    assert!(msg.contains("spin_task"), "got: {msg}");
    assert!(msg.contains("budget"), "got: {msg}");
}

#[kithara::test(native, flash(false))]
fn blanket_wait_over_budget_logs_not_panics() {
    const BLANKET_TEST_SLEEP_MS: u64 = 50;

    force_mode(Mode::Panic);
    force_blanket_budget(Duration::from_millis(BLANKET_TEST_BUDGET_MS));

    let path = temp_log_path("blanket-wait");
    let _ = fs::remove_file(&path);
    force_log_path(path.clone());

    let caught = std::panic::catch_unwind(|| {
        let fut = watch_blanket("blanket_wait_task", async {
            thread::sleep(Duration::from_millis(BLANKET_TEST_SLEEP_MS));
        });
        let _ = poll_once(fut);
    });
    if let Err(err) = caught {
        let msg = err
            .downcast_ref::<String>()
            .map_or("non-string panic payload", String::as_str);
        panic!("blanket wait must log instead of panic: {msg}");
    }

    let contents = fs::read_to_string(&path).expect("read blanket census log");
    assert!(contents.contains("[no_block][census]"), "got: {contents}");
    assert!(contents.contains("blanket_wait_task"), "got: {contents}");
    let _ = fs::remove_file(path);
}

#[kithara::test(native, flash(false))]
fn blanket_spin_over_budget_panics() {
    const FORCED_SPIN_CPU_MS: u64 = 10_000;

    force_mode(Mode::Panic);
    force_blanket_budget(Duration::from_millis(BLANKET_TEST_BUDGET_MS));
    force_cpu_elapsed(Some(Duration::from_millis(FORCED_SPIN_CPU_MS)));

    let caught = std::panic::catch_unwind(|| {
        let fut = watch_blanket("blanket_spin_task", async {
            spin_for(Duration::from_millis(BLANKET_TEST_SPIN_MS));
        });
        let _ = poll_once(fut);
    });
    force_cpu_elapsed(None);

    let err = caught.expect_err("blanket CPU spin must panic");
    let msg = err.downcast_ref::<String>().expect("panic payload");
    assert!(msg.contains("blanket_spin_task"), "got: {msg}");
    assert!(msg.contains("CPU spin"), "got: {msg}");
}

#[kithara::test(native, flash(false))]
fn off_mode_skips_blocking_checks_and_budget() {
    force_mode(Mode::Off);

    let caught_off = std::panic::catch_unwind(|| {
        let fut = watch_budget("off_task", 10, async {
            crate::thread::sleep(Duration::from_millis(1));
        });
        let _ = poll_once(fut);
    });
    assert!(
        caught_off.is_ok(),
        "off mode must skip forbid and budget checks: {caught_off:?}"
    );

    force_mode(Mode::Panic);
    let caught_panic = std::panic::catch_unwind(|| {
        let fut = watch_budget("panic_task", 10, async {
            crate::thread::sleep(Duration::from_millis(1));
        });
        let _ = poll_once(fut);
    });
    let err = caught_panic.expect_err("sleep in panic mode must hit forbid");
    let msg = err.downcast_ref::<String>().expect("panic payload");
    assert!(msg.contains("thread::sleep"), "got: {msg}");
    assert!(msg.contains("panic_task"), "got: {msg}");
}

#[kithara::test(native, flash(false))]
fn budget_ignores_paused_time() {
    force_mode(Mode::Panic);

    let fut = watch_budget("paused_task", 10, async {
        let _p = permit();
        spin_for(Duration::from_millis(50));
    });
    let _ = poll_once(fut);
}

/// The CPU twin of [`budget_ignores_paused_time`]. A pause takes its region out
/// of the poll's wall, so the CPU that region burned has to leave with it:
/// weighing a net wall against a gross CPU reads every sanctioned pass of real
/// arithmetic as a spin, and the blanket tier panics on exactly that label. The
/// Cochlea oracle poll reported 129ms of CPU inside 2.8ms of wall on that
/// arithmetic.
#[kithara::test(native, flash(false))]
fn budget_ignores_paused_cpu() {
    const PAUSED_CPU_SLEEP_MS: u64 = 20;

    force_mode(Mode::Census);
    force_no_log_path();
    force_blanket_budget(Duration::from_millis(BLANKET_TEST_BUDGET_MS));

    let traced = capture_tracing(|| {
        let fut = watch_blanket("paused_cpu_task", async {
            {
                let _p = permit();
                spin_for(Duration::from_millis(BLANKET_TEST_SPIN_MS));
            }
            thread::sleep(Duration::from_millis(PAUSED_CPU_SLEEP_MS));
        });
        let _ = poll_once(fut);
    });

    let line = traced
        .lines()
        .find(|line| line.contains("single poll took"))
        .expect("over-budget census line");
    assert!(line.contains("paused_cpu_task"), "got: {line}");
    assert!(line.contains("blocked wait"), "got: {line}");
}

/// A poll that sat without working does not spend a work budget.
///
/// This is the shape a loaded runner produces: 58.7ms of wall against 42us of
/// CPU, which a wall budget reads as an overrun and a work budget reads as the
/// deschedule it was.
#[kithara::test(native, flash(false))]
fn a_work_budget_ignores_a_poll_that_did_no_work() {
    const WORK_TEST_SLEEP_MS: u64 = 50;

    force_mode(Mode::Panic);

    let fut = watch_cpu_budget("descheduled_task", WORK_TEST_BUDGET_MS, async {
        thread::sleep(Duration::from_millis(WORK_TEST_SLEEP_MS));
    });
    let _ = poll_once(fut);
}

/// Sanctioned arithmetic leaves the work budget where it found it, so the
/// budget still reads what the unsanctioned remainder spent.
#[kithara::test(native, flash(false))]
fn a_work_budget_ignores_sanctioned_work() {
    force_mode(Mode::Panic);

    let fut = watch_cpu_budget("sanctioned_work_task", WORK_TEST_BUDGET_MS, async {
        let _p = permit();
        spin_cpu_for(Duration::from_millis(WORK_TEST_SPIN_CPU_MS));
    });
    let _ = poll_once(fut);
}

#[kithara::test(native, flash(false))]
fn a_work_budget_flags_a_poll_that_spent_it() {
    force_mode(Mode::Panic);

    let caught = std::panic::catch_unwind(|| {
        let fut = watch_cpu_budget("work_task", WORK_TEST_BUDGET_MS, async {
            spin_cpu_for(Duration::from_millis(WORK_TEST_SPIN_CPU_MS));
        });
        let _ = poll_once(fut);
    });
    let err = caught.expect_err("a poll that spent the work budget must panic");
    let msg = err.downcast_ref::<String>().expect("panic payload");
    assert!(msg.contains("work_task"), "got: {msg}");
    assert!(msg.contains("budget"), "got: {msg}");
}

#[kithara::test(native, flash(false))]
fn fast_poll_passes() {
    force_mode(Mode::Panic);

    assert!(matches!(
        poll_once(watch_budget("ok", 10, async {})),
        Poll::Ready(())
    ));
}

#[kithara::test(native, flash(false))]
fn census_writes_to_forced_log_path() {
    force_mode(Mode::Census);

    let path = temp_log_path("census");
    let _ = fs::remove_file(&path);
    force_log_path(path.clone());

    let fut = watch_budget("census_file_task", CENSUS_LOG_BUDGET_MS, async {
        crate::thread::sleep(Duration::from_millis(CENSUS_LOG_SLEEP_MS));
    });
    let _ = poll_once(fut);

    let contents = fs::read_to_string(&path).expect("read census log");
    assert!(
        contents.starts_with(&super::report::nextest_prefix()),
        "census line must carry current nextest correlation: {contents}"
    );
    assert!(contents.contains("census_file_task"), "got: {contents}");
    let _ = fs::remove_file(path);
}

#[kithara::test(native, flash(false))]
fn census_panics_when_configured_log_cannot_be_written() {
    force_mode(Mode::Census);

    let missing_parent = temp_log_path("missing-parent");
    let _ = fs::remove_dir_all(&missing_parent);
    let path = missing_parent.join("census.log");
    force_log_path(path.clone());

    let caught = std::panic::catch_unwind(|| {
        let fut = watch_budget("unwritable_census_task", CENSUS_LOG_BUDGET_MS, async {
            crate::thread::sleep(Duration::from_millis(CENSUS_LOG_SLEEP_MS));
        });
        let _ = poll_once(fut);
    });

    let err = caught.expect_err("configured census write failure must panic");
    let msg = err.downcast_ref::<String>().expect("panic payload");
    assert!(
        msg.contains("failed to write census log"),
        "unexpected panic: {msg}"
    );
    assert!(
        msg.contains(&path.display().to_string()),
        "panic must identify the configured path: {msg}"
    );
}

#[kithara::test(native, flash(false))]
fn a_configured_census_log_is_the_only_sink() {
    force_mode(Mode::Census);

    let path = temp_log_path("census-sole-sink");
    let _ = fs::remove_file(&path);
    force_log_path(path.clone());

    let traced = capture_tracing(|| census_once("census_sole_sink_task"));

    assert!(
        !traced.contains("census_sole_sink_task"),
        "a configured log takes the stream; a second copy lands in the JUnit: {traced}"
    );
    let _ = fs::remove_file(path);
}

#[kithara::test(native, flash(false))]
fn census_without_a_configured_log_reaches_the_tracing_sink() {
    force_mode(Mode::Census);
    force_no_log_path();

    let traced = capture_tracing(|| census_once("census_traced_task"));

    assert!(traced.contains("census_traced_task"), "got: {traced}");
}

#[kithara::test(native, flash(false))]
fn forbid_fires_on_platform_sleep_inside_poll() {
    force_mode(Mode::Panic);

    let caught = std::panic::catch_unwind(|| {
        let fut = watch_budget("sleeper", 10_000, async {
            crate::thread::sleep(Duration::from_millis(1));
        });
        let _ = poll_once(fut);
    });
    let err = caught.expect_err("platform sleep inside poll must hit forbid");
    let msg = err.downcast_ref::<String>().expect("panic payload");
    assert!(msg.contains("thread::sleep"), "got: {msg}");
    assert!(msg.contains("sleeper"), "got: {msg}");
    assert!(
        msg.contains("tests.rs"),
        "forbid must attribute the call site, got: {msg}"
    );
}

#[kithara::test(native, flash(false))]
fn allow_block_permit_suppresses_forbid() {
    force_mode(Mode::Panic);

    let fut = watch_budget("permitted_sleeper", 10_000, async {
        let _permit = permit();
        crate::thread::sleep(Duration::from_millis(1));
    });
    let _ = poll_once(fut);
}

#[kithara::test(native, flash(false))]
fn permit_poll_suppresses_forbid_and_budget() {
    force_mode(Mode::Panic);

    let fut = watch_budget(
        "outer",
        10,
        permit_poll(async {
            crate::thread::sleep(Duration::from_millis(30));
        }),
    );
    let _ = poll_once(fut);
}

#[kithara::test(native, flash(false))]
fn forbid_still_fires_after_permit_poll_scope_ends() {
    force_mode(Mode::Panic);

    let permitted = watch_budget(
        "permitted",
        10,
        permit_poll(async {
            crate::thread::sleep(Duration::from_millis(1));
        }),
    );
    let _ = poll_once(permitted);

    let caught = std::panic::catch_unwind(|| {
        let fut = watch_budget("plain", 10_000, async {
            crate::thread::sleep(Duration::from_millis(1));
        });
        let _ = poll_once(fut);
    });
    let err = caught.expect_err("plain sleep after permit_poll must hit forbid");
    let msg = err.downcast_ref::<String>().expect("panic payload");
    assert!(msg.contains("thread::sleep"), "got: {msg}");
    assert!(msg.contains("plain"), "got: {msg}");
}

#[kithara::test(native, flash(false))]
fn sleep_outside_poll_is_untouched() {
    force_mode(Mode::Panic);

    crate::thread::sleep(Duration::from_millis(1));
}

#[kithara::test(native, flash(false))]
fn snapshot_rate_limits_thread_cpu_reads() {
    force_mode(Mode::Panic);

    clock::force_snapshot_refresh_count(0);

    let t0 = Instant::now();
    let _first = clock::snapshot(t0);
    let _second = clock::snapshot(t0);
    assert_eq!(
        clock::snapshot_refresh_count(),
        1,
        "same instant should refresh once"
    );

    let _third = clock::snapshot(t0 + Duration::from_millis(2));
    assert_eq!(
        clock::snapshot_refresh_count(),
        2,
        "2ms later should force a second refresh"
    );
}