vj-sim 0.3.0

//! SCHED::IO_GATE — the GVT-gated `defer_io` forwarding seam, the
//! IoLog↔Commit-Queue reconciliation (SC-124), and `request_finish` as a
//! deferred effect (SC-075). See `docs/CONTRACTS.md` deviation item 3.
//!
//! Per SC-124 (already settled by `docs/CONTRACTS.md`'s crate-graph:
//! `vj-sim` cannot depend on `vj-runtime` — the dependency edge runs
//! the other way), this track does **not** build a second independent
//! GVT-gated queue with its own drain logic. It builds the one seam —
//! [`IoSink`] — through which every `defer_io` call is forwarded
//! immediately (never buffered here); [`DeferIoDispatcher`] proves this by
//! construction (no field but `sink`/`seq`, no drain method at all). The
//! real GVT-gated queue that `IoSink` forwards into in production is
//! `vj-runtime`'s Commit Queue (built by the runtime track against this
//! same `IoSink` shape); `GatedFakeSink` (this module's own test double,
//! scoped to `#[cfg(test)]`) stands in for it purely to exercise the gating
//! semantics DT-012/SC-074 require, without this crate owning a second real
//! queue.

use std::sync::atomic::{AtomicU64, Ordering};

use vj_abi::Region;

use crate::sched::{DeferredIoCallback, SchedTime};

/// The one forwarding target every `defer_io` call lands in (SC-124). A real
/// engine constructs its [`DeferIoDispatcher`] against `vj-runtime`'s
/// Commit Queue, which implements this trait (docs/CONTRACTS.md deviation
/// item 3); this crate never implements it for production use itself — only
/// this module's own `#[cfg(test)]` fake does, purely to exercise the
/// gating contract.
pub trait IoSink: Send + Sync {
    fn enqueue(&self, at: SchedTime, region: Region, seq: u64, cb: DeferredIoCallback);
}

/// Implements the `defer_io` half of `SchedulerOps` by construction
/// (SC-072, SC-073, SC-124): [`Self::defer_io`] does exactly one thing —
/// assigns the next monotonic `seq` and forwards to `sink.enqueue` — no
/// local buffering, no drain method on this struct at all. There is
/// deliberately nowhere to buffer into (SC-124's "thin forwarding call, not
/// a parallel queue with its own drain logic").
pub struct DeferIoDispatcher<'a> {
    sink: &'a dyn IoSink,
    seq: AtomicU64,
}

impl<'a> DeferIoDispatcher<'a> {
    pub fn new(sink: &'a dyn IoSink) -> Self {
        Self {
            sink,
            seq: AtomicU64::new(0),
        }
    }

    /// The `SchedulerOps::defer_io` primitive (SC-072, SC-073, SC-124):
    /// never invokes `cb` itself, never buffers it — assigns the next
    /// sequence number and forwards immediately to the configured
    /// [`IoSink`]. Any concrete `SchedulerOps::defer_io` implementation must
    /// route through this (mirrors `ops::commit_blocking_write`'s role as
    /// the one sanctioned body for its `SchedulerOps` method).
    pub fn defer_io(&self, at: SchedTime, region: Region, cb: DeferredIoCallback) {
        let seq = self.seq.fetch_add(1, Ordering::SeqCst);
        self.sink.enqueue(at, region, seq, cb);
    }
}

/// `SchedulerOps::request_finish(code)` (SC-075): routes `$finish`/`$stop`
/// through the *same* `defer_io` path as any other effect — never a
/// synchronous stop. Any concrete `SchedulerOps::request_finish`
/// implementation must call this (mirrors `ops::commit_blocking_write`'s
/// role as the one sanctioned body for its `SchedulerOps` method).
///
/// `code` is the requested `$finish`/`$stop` exit code. This wave's
/// `IoCtx` is an empty marker trait (RUNTIME's own concern to fill in with
/// a real finish-code sink), so there is nowhere production-side for this
/// function itself to hand `code` *through* — it is a named parameter
/// (rather than folded silently into `on_finish`'s closure alone) so every
/// `request_finish` call site stays self-documenting about which exit code
/// it queued, and so a future `IoCtx` that gains a finish-code sink has one
/// obvious place to thread it from.
pub fn request_finish_as_deferred_effect(
    dispatcher: &DeferIoDispatcher<'_>,
    at: SchedTime,
    region: Region,
    code: i32,
    on_finish: DeferredIoCallback,
) {
    let _ = code;
    dispatcher.defer_io(at, region, on_finish);
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::AtomicBool;
    use std::sync::{Arc, Mutex};

    /// This task's own in-crate test double for the GVT-gated queue
    /// `IoSink` forwards into in production (`vj-runtime`'s Commit
    /// Queue) — models just enough of scheduler.md's "IoLog" shape
    /// (per-(time,region) deferred entries, flushed strictly once `gvt()`
    /// passes each entry) to prove `DeferIoDispatcher`'s forwarding and the
    /// gating semantics DT-012/SC-074 require, without this crate owning a
    /// second real queue (SC-124).
    struct GatedFakeSink {
        gvt: Mutex<SchedTime>,
        pending: Mutex<Vec<(SchedTime, Region, u64, /* fired */ bool)>>,
        flush_log: Mutex<Vec<u64>>,
    }

    impl GatedFakeSink {
        fn new(initial_gvt: SchedTime) -> Self {
            Self {
                gvt: Mutex::new(initial_gvt),
                pending: Mutex::new(Vec::new()),
                flush_log: Mutex::new(Vec::new()),
            }
        }

        /// "Fires" (marks flushed) every still-pending, not-yet-fired entry
        /// with `at <= new_gvt`, processed in `(at, region, seq)` order, and
        /// returns how many were newly fired by this call (an already-fired
        /// entry is never fired/counted twice).
        fn advance_gvt(&self, new_gvt: SchedTime) -> usize {
            *self.gvt.lock().unwrap() = new_gvt;
            let mut pending = self.pending.lock().unwrap();
            pending.sort_by_key(|(at, region, seq, _)| (*at, *region as u32, *seq));
            let mut flush_log = self.flush_log.lock().unwrap();
            let mut fired = 0usize;
            for entry in pending.iter_mut() {
                let (at, _region, seq, already_fired) = entry;
                if !*already_fired && *at <= new_gvt {
                    *already_fired = true;
                    flush_log.push(*seq);
                    fired += 1;
                }
            }
            fired
        }

        fn entries(&self) -> Vec<(SchedTime, Region, u64, bool)> {
            self.pending.lock().unwrap().clone()
        }

        fn flush_order(&self) -> Vec<u64> {
            self.flush_log.lock().unwrap().clone()
        }
    }

    impl IoSink for GatedFakeSink {
        fn enqueue(&self, at: SchedTime, region: Region, seq: u64, cb: DeferredIoCallback) {
            // Models the GVT-gating metadata this module's tests check (when
            // an entry becomes eligible to flush); it is not a second real
            // effect-execution engine, so `cb` is intentionally dropped
            // unexecuted rather than stored/invoked later — no test in this
            // module depends on `cb` actually running once flushed.
            drop(cb);
            self.pending.lock().unwrap().push((at, region, seq, false));
        }
    }

    #[test]
    fn defer_io_never_executes_the_callback_synchronously() {
        let sink = GatedFakeSink::new(SchedTime { ticks: 0 });
        let dispatcher = DeferIoDispatcher::new(&sink);
        let fired = Arc::new(AtomicBool::new(false));
        let fired_clone = fired.clone();
        let cb: DeferredIoCallback = Box::new(move |_ctx| {
            fired_clone.store(true, Ordering::SeqCst);
        });
        dispatcher.defer_io(SchedTime { ticks: 3 }, Region::Active, cb);
        assert!(
            !fired.load(Ordering::SeqCst),
            "defer_io must never invoke its callback synchronously"
        );
    }

    #[test]
    fn defer_io_is_a_thin_forward_no_local_buffering() {
        assert_eq!(
            core::mem::size_of::<DeferIoDispatcher<'static>>(),
            core::mem::size_of::<&dyn IoSink>() + core::mem::size_of::<AtomicU64>(),
            "DeferIoDispatcher must carry nothing beyond its sink reference and seq counter"
        );

        let sink = GatedFakeSink::new(SchedTime { ticks: 0 });
        let dispatcher = DeferIoDispatcher::new(&sink);
        dispatcher.defer_io(SchedTime { ticks: 1 }, Region::Active, Box::new(|_| {}));
        dispatcher.defer_io(SchedTime { ticks: 2 }, Region::Active, Box::new(|_| {}));

        let entries = sink.entries();
        assert_eq!(
            entries.len(),
            2,
            "each defer_io call must reach the sink exactly once, nothing coalesced or dropped locally"
        );
        assert!(
            entries[0].2 < entries[1].2,
            "seq must strictly increase across defer_io calls"
        );
    }

    #[test]
    fn under_the_degenerate_controller_effects_flush_immediately() {
        // Task 3's SequentialController contract: gvt() == now always.
        let now = SchedTime { ticks: 5 };
        let sink = GatedFakeSink::new(now);
        let dispatcher = DeferIoDispatcher::new(&sink);
        dispatcher.defer_io(now, Region::Active, Box::new(|_| {}));

        let fired = sink.advance_gvt(now);
        assert_eq!(
            fired, 1,
            "under the degenerate controller gvt==now, so the entry fires as soon as gvt is advanced to now"
        );
    }

    #[test]
    fn under_a_speculative_mock_controller_effects_wait_for_gvt_to_pass() {
        let sink = GatedFakeSink::new(SchedTime { ticks: 0 });
        let dispatcher = DeferIoDispatcher::new(&sink);
        dispatcher.defer_io(SchedTime { ticks: 10 }, Region::Active, Box::new(|_| {}));

        assert_eq!(
            sink.advance_gvt(SchedTime { ticks: 5 }),
            0,
            "gvt has not yet reached the entry's scheduled tick"
        );
        assert_eq!(
            sink.advance_gvt(SchedTime { ticks: 10 }),
            1,
            "gvt reaching the entry's tick (>=) is the flush condition"
        );

        // Same-tick two-entry sub-case: flush order must match seq order.
        let sink2 = GatedFakeSink::new(SchedTime { ticks: 0 });
        let dispatcher2 = DeferIoDispatcher::new(&sink2);
        dispatcher2.defer_io(SchedTime { ticks: 20 }, Region::Active, Box::new(|_| {}));
        dispatcher2.defer_io(SchedTime { ticks: 20 }, Region::Active, Box::new(|_| {}));
        let entries = sink2.entries();
        let (first_seq, second_seq) = (entries[0].2, entries[1].2);
        assert_eq!(sink2.advance_gvt(SchedTime { ticks: 20 }), 2);
        assert_eq!(
            sink2.flush_order(),
            vec![first_seq, second_seq],
            "two entries at the same tick must flush in seq order"
        );
    }

    #[test]
    fn request_finish_does_not_halt_other_same_timestep_activations() {
        let sink = GatedFakeSink::new(SchedTime { ticks: 0 });
        let dispatcher = DeferIoDispatcher::new(&sink);
        let now = SchedTime { ticks: 4 };

        request_finish_as_deferred_effect(&dispatcher, now, Region::Active, 0, Box::new(|_| {}));
        assert_eq!(
            sink.entries().len(),
            1,
            "request_finish must enqueue exactly one deferred entry, not execute synchronously"
        );

        // Nothing about calling request_finish disables the dispatcher for
        // ordinary defer_io calls made "after" it in the same timestep.
        dispatcher.defer_io(now, Region::Active, Box::new(|_| {}));
        dispatcher.defer_io(now, Region::Active, Box::new(|_| {}));
        assert_eq!(
            sink.entries().len(),
            3,
            "unrelated defer_io calls after request_finish must still enqueue normally"
        );
    }
}