vj_sim/determinism/

sequential.rs

//! DETERMINISM::SEQUENTIAL — the v1 default `DeterminismController`: a
//! degenerate, strictly-sequential dispatcher (ledger SC-009, DT-004, DT-006,
//! DT-007) that runs the given, already-canonically-sorted (Task 2, DT-005)
//! ready set in the order it was handed, on the caller's own thread, with
//! `gvt() == now` always and no rollback ever. It is also v1's own
//! correctness oracle: its committed output is the ground truth a later real
//! WOCB controller's output is diffed against (Task 22's cross-controller
//! test).

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

use vj_abi::Region;

use super::{ActivationId, CanonicalEventKey, DeterminismController, ResolvedBatch, WorkItem};
use crate::sched::{PendingActivation, SchedTime, region_ordinal};
use crate::state::{FourStateValue, StateCellId};

/// The v1 degenerate sequential controller (SC-009/DT-007): a single global
/// `Mutex<SchedTime>` for `gvt` and one monotonically increasing
/// `local_send_seq` counter. Never spawns a thread (DT-006) and never
/// records a conflict set — `on_read`/`on_write` are no-ops because a
/// strictly sequential dispatcher has nothing to detect a conflict against;
/// that absence of any rollback machinery *is* DT-007's "no rollback ever
/// needed" made concrete, not an oversight.
pub struct SequentialController {
    gvt: Mutex<SchedTime>,
    next_seq: AtomicU64,
    /// **Task 22 (DT-033) testability hook — see [`Self::record_rollback`]'s
    /// own doc comment for why this can only ever read `0` in this wave, and
    /// why that is still a real, not-vacuous regression guard.** `Arc`-backed
    /// (not a bare `AtomicU64`) so a caller can clone a handle *before*
    /// boxing this controller and handing it to `Engine::
    /// set_determinism_controller` — `Engine` only ever exposes the boxed
    /// `dyn DeterminismController` back out via `controller_kind()`'s string
    /// tag, never the concrete type, so this is the one way a test outside
    /// this module can still observe the counter after a run.
    rollback_count: Arc<AtomicU64>,
}

impl SequentialController {
    pub fn new() -> Self {
        Self {
            gvt: Mutex::new(SchedTime { ticks: 0 }),
            next_seq: AtomicU64::new(0),
            rollback_count: Arc::new(AtomicU64::new(0)),
        }
    }

    /// The engine's `set_determinism_controller` fallback (Task 18) calls
    /// this when no controller has been configured — SC-009's "selected by
    /// default when no other controller is configured" half.
    pub fn default_boxed() -> Box<dyn DeterminismController> {
        Box::new(Self::new())
    }

    /// A clonable handle onto this controller's rollback-event counter,
    /// readable after this controller has been moved into an `Engine` (see
    /// [`Self::rollback_count`]'s doc comment for why this indirection is
    /// necessary at all). Task 22's `dt033_conservative_v1_default_produces_
    /// zero_rollback_events` integration test (`tests/region_order_fixture.rs`)
    /// clones this *before* calling `set_determinism_controller`.
    pub fn rollback_count_handle(&self) -> Arc<AtomicU64> {
        self.rollback_count.clone()
    }

    /// **DT-033/DT-001's "zero rollback events" acceptance, made into a real
    /// counter rather than only prose.** Nothing in this file's own
    /// production code (`partition`/`on_read`/`on_write`/`gvt`/
    /// `drain_and_resolve` above) calls this method anywhere — a strictly
    /// sequential, single-thread dispatcher with `gvt() == now` always has no
    /// conflict to roll back from by construction (DT-007), so this counter
    /// can only ever read `0` for *this* controller in *this* wave. That is
    /// not the same as the test being unable to fail: it is a real,
    /// wired-through counter (not a hardcoded `0` the test reads), so a
    /// future edit that mistakenly adds rollback-triggering logic to this
    /// controller (e.g. a well-meaning "fix" that starts calling this from
    /// `on_write` on a detected write-write conflict, forgetting `DT-007`'s
    /// whole point is that this controller must never need to) trips this
    /// counter and fails Task 22's test — the same "currently-always-passes,
    /// genuinely-can-regress" shape as `dt006_no_thread_spawn_or_rollback_
    /// code_outside_determinism_module` above.
    pub fn record_rollback(&self) {
        self.rollback_count.fetch_add(1, Ordering::SeqCst);
    }
}

impl Default for SequentialController {
    fn default() -> Self {
        Self::new()
    }
}

impl DeterminismController for SequentialController {
    fn partition(
        &self,
        ready: &[PendingActivation],
        region: Region,
        now: SchedTime,
    ) -> Vec<WorkItem> {
        *self
            .gvt
            .lock()
            .expect("SequentialController gvt mutex poisoned") = now;
        ready
            .iter()
            .map(|activation| {
                let seq = self.next_seq.fetch_add(1, Ordering::SeqCst);
                let key = CanonicalEventKey {
                    vtime: activation.logical_order.0,
                    region_ordinal: region_ordinal(region),
                    reconv_iter: activation.reconv_iter,
                    canonical_process_id: activation.proc,
                    local_send_seq: seq,
                };
                WorkItem {
                    activation: *activation,
                    key,
                }
            })
            .collect()
    }

    fn on_read(&self, _act: ActivationId, _cell: StateCellId, _version: u64) {}

    fn on_write(
        &self,
        _act: ActivationId,
        _cell: StateCellId,
        _prior: FourStateValue,
        _version: u64,
    ) {
    }

    fn gvt(&self) -> SchedTime {
        *self
            .gvt
            .lock()
            .expect("SequentialController gvt mutex poisoned")
    }

    fn drain_and_resolve(&self, work: Vec<WorkItem>) -> ResolvedBatch {
        ResolvedBatch {
            completed: work.iter().map(|w| w.activation.id).collect(),
        }
    }

    /// Task 18's `Engine::controller_kind()` testability tag (DT-007's
    /// "selected by default when no other controller is configured" half,
    /// made observable from outside).
    fn kind(&self) -> &'static str {
        "sequential"
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sched::{ProcessId, ResumeArg, canonical_key_of, sort_ready_set};

    fn activation(
        id: ActivationId,
        proc: ProcessId,
        ticks: u64,
        reconv_iter: u32,
        registration_order: u64,
    ) -> PendingActivation {
        PendingActivation {
            id,
            proc,
            arg: ResumeArg::Start,
            reconv_iter,
            logical_order: (SchedTime { ticks }, Region::Active, registration_order),
        }
    }

    #[test]
    fn sequential_controller_gvt_equals_now_after_partition() {
        let controller = SequentialController::new();
        let out = controller.partition(&[], Region::Active, SchedTime { ticks: 9 });
        assert!(out.is_empty());
        assert_eq!(controller.gvt(), SchedTime { ticks: 9 });
    }

    #[test]
    fn sequential_controller_never_reorders_the_given_ready_set() {
        // Feed a pre-sorted 3-element ready set (Task 2's sort_ready_set)
        // whose canonical (post-sort) proc order is NEITHER ascending NOR
        // descending in numeric ProcessId ([20, 10, 30]), so a `partition`
        // that secretly re-derives order from ProcessId -- ascending OR
        // descending -- produces a different, and therefore caught, output.
        // (Registration order 0,1,2 was previously assigned to procs
        // 30,10,20 in construction order, which sorts to [10,20,30] --
        // coincidentally ascending-by-ProcessId, so a "canonicalize by
        // ascending ProcessId" bug would have passed undetected. Assigning
        // registration order 0,1,2 to procs 20,10,30 instead makes the
        // sorted sequence [20,10,30], which is monotonic in neither
        // direction.)
        let mut ready = vec![
            activation(3, 30, 5, 0, 2),
            activation(2, 20, 5, 0, 0),
            activation(1, 10, 5, 0, 1),
        ];
        sort_ready_set(&mut ready);
        // sort_ready_set orders by (time, region, reconv_iter, registration);
        // all three share time/region/reconv_iter here, so registration order
        // (0,1,2) decides: proc 20, then 10, then 30.
        assert_eq!(
            ready.iter().map(|a| a.proc).collect::<Vec<_>>(),
            vec![20, 10, 30]
        );

        let controller = SequentialController::new();
        let out = controller.partition(&ready, Region::Active, SchedTime { ticks: 5 });

        let procs: Vec<ProcessId> = out.iter().map(|w| w.activation.proc).collect();
        assert_eq!(
            procs,
            vec![20, 10, 30],
            "partition must preserve the caller's given order, never reorder itself \
             (this order is neither ascending nor descending in ProcessId, so a \
             reorder-by-ProcessId bug in either direction is caught)"
        );
        for (item, act) in out.iter().zip(ready.iter()) {
            let (vtime, region_ordinal, reconv_iter, proc) = canonical_key_of(act);
            assert_eq!(item.key.vtime, vtime);
            assert_eq!(item.key.region_ordinal, region_ordinal);
            assert_eq!(item.key.reconv_iter, reconv_iter);
            assert_eq!(item.key.canonical_process_id, proc);
        }
    }

    #[test]
    fn sequential_controller_assigns_strictly_increasing_local_send_seq() {
        let controller = SequentialController::new();
        let ready1 = vec![activation(1, 1, 1, 0, 0), activation(2, 2, 1, 0, 1)];
        let ready2 = vec![activation(3, 3, 2, 0, 0)];

        let out1 = controller.partition(&ready1, Region::Active, SchedTime { ticks: 1 });
        let out2 = controller.partition(&ready2, Region::Active, SchedTime { ticks: 2 });

        let seqs: Vec<u64> = out1
            .iter()
            .chain(out2.iter())
            .map(|w| w.key.local_send_seq)
            .collect();
        let mut sorted = seqs.clone();
        sorted.sort_unstable();
        sorted.dedup();
        assert_eq!(
            seqs.len(),
            sorted.len(),
            "every local_send_seq across both partition calls must be unique: {seqs:?}"
        );
        assert!(
            seqs.windows(2).all(|w| w[0] < w[1]),
            "local_send_seq must be strictly increasing across successive partition \
             calls, not reset per call: {seqs:?}"
        );
    }

    /// Task 22 (DT-033) — proves the counter itself is real, wired-through
    /// state (not a constant `0` the getter hardcodes): starts at `0`, a
    /// `record_rollback()` call increments the *same* handle a caller cloned
    /// beforehand, and a fresh `SequentialController` starts back at `0`
    /// (the counter is per-instance, never shared across controllers). Fails
    /// if `rollback_count_handle` returns a detached copy instead of a
    /// shared `Arc`, or if `record_rollback` doesn't actually touch it.
    #[test]
    fn rollback_count_handle_is_zero_until_record_rollback_is_called() {
        let controller = SequentialController::new();
        let handle = controller.rollback_count_handle();
        assert_eq!(handle.load(Ordering::SeqCst), 0);

        controller.record_rollback();
        assert_eq!(
            handle.load(Ordering::SeqCst),
            1,
            "record_rollback must increment the same counter rollback_count_handle \
             returns a clone of"
        );
        controller.record_rollback();
        assert_eq!(handle.load(Ordering::SeqCst), 2);

        let fresh = SequentialController::new();
        assert_eq!(
            fresh.rollback_count_handle().load(Ordering::SeqCst),
            0,
            "a new SequentialController's rollback counter must start at 0, not \
             inherit a prior instance's count"
        );
    }

    #[test]
    fn sequential_controller_is_the_default_when_none_configured() {
        let controller = SequentialController::default_boxed();
        let out = controller.partition(&[], Region::Active, SchedTime { ticks: 3 });
        assert!(out.is_empty());
        assert_eq!(controller.gvt(), SchedTime { ticks: 3 });
    }

    /// DT-006: the scheduler's *production* dispatch/scheduling code never
    /// spawns a thread or does its own rollback — an architectural
    /// constraint on the code that ships, not a ban on the test-only OS-
    /// thread concurrency stress tests (`resume.rs`'s SC-061 tests) that
    /// prove a guarantee ABOUT that code. Scoped to production code only via
    /// `scan_support::strip_cfg_test_blocks` (shared, not reimplemented, by
    /// Task 22's crate-wide `notify_change` scan) so this test does not
    /// falsely trip on those already-shipped, plan-mandated
    /// `#[cfg(test)]`-only stress tests, and does not silently hide a real
    /// production violation smuggled in next to a test module either.
    #[test]
    fn dt006_no_thread_spawn_or_rollback_code_outside_determinism_module() {
        let sched_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src/sched");
        let mut scanned_files = 0usize;
        let mut violations = Vec::new();
        scan_dir_for_thread_spawn(&sched_dir, &mut scanned_files, &mut violations);

        assert!(
            scanned_files > 0,
            "scan matched no files under {sched_dir:?} — the scan itself is broken"
        );
        assert!(
            violations.is_empty(),
            "found `thread::spawn` in production (non-#[cfg(test)]) code under \
             sched/: {violations:?} — scheduler production code must never spawn a \
             thread; that is exclusively the DeterminismController's job (DT-006)"
        );
    }

    fn scan_dir_for_thread_spawn(
        dir: &std::path::Path,
        scanned_files: &mut usize,
        violations: &mut Vec<String>,
    ) {
        for entry in std::fs::read_dir(dir).expect("read sched/ dir for DT-006 scan") {
            let entry = entry.expect("dir entry for DT-006 scan");
            let path = entry.path();
            if path.is_dir() {
                scan_dir_for_thread_spawn(&path, scanned_files, violations);
                continue;
            }
            if path.extension().is_none_or(|e| e != "rs") {
                continue;
            }
            *scanned_files += 1;
            let src = std::fs::read_to_string(&path)
                .unwrap_or_else(|e| panic!("read {path:?} for DT-006 scan: {e}"));
            let production_only = crate::determinism::scan_support::strip_cfg_test_blocks(&src);
            if production_only.contains("thread::spawn") {
                violations.push(path.display().to_string());
            }
        }
    }
}