vj-sim 0.2.0

//! DETERMINISM (`vj-sim::determinism`) — cross-track contract and, as of
//! Task 3, its v1 implementation: the pluggable ordering/rollback interface
//! the scheduler hands ready-sets to (scheduler.md §6, ARCHITECTURE §4). The
//! scheduler "never itself assigns threads or implements rollback" — every
//! concrete controller (the v1 degenerate sequential oracle, and later WOCB)
//! is this track's own implementation, not materialized in `sched`.
//!
//! **SC-008 module-boundary note:** `sched` (regions/wheel — Tasks 4–5) and
//! `determinism` (this trait plus [`SequentialController`]) are both
//! re-exported from the same `vj-sim` crate root, satisfying "expose all
//! five named responsibilities (regions, time wheel, `DeterminismController`
//! trait, SoA arena, undo-log) as part of the same module boundary" for the
//! two of the five this track owns; the SoA arena and undo-log are STATE's
//! (`crate::state`), cited here, not re-implemented.
//!
//! ## WOCB tension — how this wave resolves it (DT-033)
//!
//! ARCHITECTURE.md §4's staged WOCB plan resolves the tension between a
//! future multi-threaded optimistic Time-Warp kernel and the bit-identical-
//! determinism-across-thread-counts MUST-HAVE gate by shipping v1 as
//! deliberately conservative: `LOOKAHEAD_CYCLES = 0` (no per-region
//! optimism — nothing this wave ever runs ahead of `gvt()`, since
//! [`SequentialController::gvt`] always equals the `now` last passed to
//! `partition`), one canonical event key ([`CanonicalEventKey`], Task 2's
//! derived total order, `sched::order::sort_ready_set` applies it to every
//! ready set before a controller ever sees it), and a GVT-gated,
//! canonically-ordered commit stream (`sched::IoSink`/`DeferIoDispatcher`,
//! Task 16) that a debugger, RUNTIME, and FST would all consume identically.
//! Concretely, as implemented (not aspiration):
//! - [`SequentialController`] is the only [`DeterminismController`] this
//!   wave ships, and it never spawns a thread
//!   (`sequential::tests::dt006_no_thread_spawn_or_rollback_code_outside_
//!   determinism_module`) — with exactly one thread, a same-thread-count
//!   "sweep" passes trivially, because there is no second thread count to
//!   diverge from.
//! - `sched::resume::tests::resume_is_never_called_concurrently_for_the_
//!   same_process_id` (Task 12) already proves `ResumeDriver`'s own locking
//!   discipline holds under genuine OS-thread parallelism — the seam a real
//!   multi-threaded controller will dispatch through — and
//!   `sched::order::tests::sort_ready_set_is_insertion_order_independent`
//!   (Task 2, this wave's SC-013 reduction) proves the canonical key, not
//!   physical arrival order, determines ready-set order, the necessary
//!   precondition for any thread count to converge to it.
//! - The real `{1,4,8,16,31}`-thread-count blake3 bit-diff gate itself
//!   (SC-013's literal acceptance) is `vj-harness`'s Tier-A/nightly job
//!   (`sched::order`'s own "SC-013 this-wave reduction" note, Task 20's
//!   DT-028 split) once a real multi-threaded controller exists — not this
//!   track, not this wave. This wave's own closest substitute is this
//!   module's `sequential_and_levelized_controllers_converge_to_the_same_
//!   canonical_effect_order` test (`tests/region_order_fixture.rs`): two
//!   controllers with genuinely different internal dispatch logic, diffed
//!   against `SequentialController`'s canonically-sorted output as ground
//!   truth (SC-009).
//! - [`SequentialController::rollback_count_handle`] backs
//!   `dt033_conservative_v1_default_produces_zero_rollback_events`
//!   (`tests/region_order_fixture.rs`): a full run of Task 21's 3-process
//!   fixture through a real `Engine` produces zero rollback events, the
//!   literal DT-001/DT-033 acceptance text — honestly a property of this
//!   controller's *design* (it has no rollback code path to trigger) rather
//!   than of anything this specific fixture exercises, exactly like
//!   `dt006_...`'s scan above; both are real regression guards against a
//!   future edit reintroducing what DT-007 says must not exist here.
//!
//! ## SC-130 pluggability review
//!
//! SC-130 requires that this wave's execution model — the 5-region
//! stratified kernel, this trait, and the canonical event-key scheme — must
//! not preclude adding a static-scheduling/levelized fast-path controller
//! later, even though v1 defers that fast path. Walking
//! [`DeterminismController`]'s 5 dispatch-relevant methods
//! (`kind` is a Task 18 testability tag only, never part of any ordering/
//! commit decision, so it is excluded from this argument):
//! - `partition` already receives the **full, canonically-sorted** ready set
//!   (`sched::order::sort_ready_set` runs before any controller sees it,
//!   Task 2/DT-005) and returns `Vec<WorkItem>` in whatever order the
//!   controller chooses — a levelized controller can statically sort by
//!   level instead of trusting the caller's pass-through order, exactly what
//!   this module's own `hypothetical_levelized_controller_compiles_against_
//!   the_unmodified_trait` test below (and `tests/region_order_fixture.rs`'s
//!   `FakeLevelizedController`) do, against the trait **as shipped**, with
//!   zero trait edits.
//! - `on_read`/`on_write` are no-ops a levelized controller is free to
//!   ignore entirely: static levelization computes its schedule from the
//!   design's own dependency structure at elaboration time, not from
//!   runtime conflict tracking, so it never needs to call into either hook
//!   for anything.
//! - `gvt` can simply track the last `now` it was handed, the same
//!   `Mutex<SchedTime>` pattern [`SequentialController::gvt`] already uses —
//!   a levelized controller that never runs ahead of `now` has the same
//!   trivial `gvt() == now` obligation.
//! - `drain_and_resolve` can run its levels **in parallel internally**
//!   (`rayon`/`crossbeam-channel`, SC-112) without the trait, `Engine`, or
//!   any caller needing to know — the trait's signature (`Vec<WorkItem> ->
//!   ResolvedBatch`) says nothing about how the work is actually executed.
//! - `Engine::set_determinism_controller` (Task 18) already accepts any `Box<
//!   dyn DeterminismController>`, proven by `sched::control::tests::set_
//!   determinism_controller_overrides_the_default` (a second, independent
//!   implementer swapped in at runtime) and this module's own
//!   `hypothetical_levelized_controller_compiles_against_the_unmodified_
//!   trait` test, which additionally round-trips a levelized stand-in
//!   through that exact seam.
//!
//! **One residual risk, accepted not hidden:** `docs/design/scheduler.md`'s
//! own Decision Record Conflicts #3 already flags that "static levelization
//! [is] deferred... a multi-threaded event kernel from v1... the
//! multi-threading goal will visibly under-deliver on comb-dominated
//! netlists until static levelization lands post-v1" — this review proves
//! the *seam* is ready for that later controller, not that v1 already
//! delivers its performance.
//!
//! ## DT-028 — CI tiering policy for the Gate-1 thread-count sweep (Task 20)
//!
//! DT-028 requires an explicit, recorded CI policy for Gate 1's thread-count
//! sweep, not an implicit default. Recorded here (this crate is where the
//! tiebreak the sweep diffs against is published, `tiebreak`) because
//! `vj-harness`'s own CI YAML is a different track's file, out of this
//! crate's scope:
//! - The full `{1, 4, 8, 16, 31}`-thread-count blake3 bit-diff sweep
//!   (SC-013's literal acceptance) is **nightly-only** — it requires a real
//!   multi-threaded [`DeterminismController`], which this wave does not
//!   ship ([`SequentialController`] never spawns a thread, see
//!   `sequential::tests::dt006_no_thread_spawn_or_rollback_code_outside_
//!   determinism_module`).
//! - A cheap **Tier-A subset is PR-blocking**: `{1, 4}` thread counts only,
//!   satisfied this wave by two checks already in this crate —
//!   `sched::resume::tests::resume_is_never_called_concurrently_for_the_
//!   same_process_id` (Task 12's real 8-OS-thread reentrancy stress test)
//!   plus a 2-controller [`committed_effect_tiebreak`] ordering
//!   check (this module's own tests and `tests/region_order_fixture.rs`'s
//!   `sequential_and_levelized_controllers_converge_to_the_same_canonical_
//!   effect_order`, Task 22).
//!
//! See [`committed_effect_tiebreak`] for DT-017's fixed ordering rule and
//! DT-027's "Gate 1 is blocked until this is published" dependency.

use vj_abi::Region;

use crate::sched::{PendingActivation, ProcessId, SchedTime};
use crate::state::{FourStateValue, StateCellId};

mod kind_guard;
mod policy;
#[cfg(test)]
pub(crate) mod scan_support;
mod sequential;
mod tiebreak;
pub use kind_guard::*;
pub use policy::*;
pub use sequential::SequentialController;
pub use tiebreak::committed_effect_tiebreak;

/// Same representation as `sched::ActivationId` (both `u64`) — re-exported
/// rather than independently redeclared (Task 1 review fix, docs/CONTRACTS.md
/// r3 item 2) so there is exactly one `ActivationId` type in the crate, not
/// two definitionally-identical-but-nominally-distinct ones.
pub use crate::sched::ActivationId;

/// The one ordering scheme used everywhere (ARCHITECTURE §4 item 3, the
/// faceoffs.md WOCB-verdict graft): intra-region delta-cycle commit, inter-region
/// barrier delivery, and the side-effect/VPI/FST commit stream all key on this
/// tuple.
///
/// `derive(PartialOrd, Ord)` is intentional and is DT-010's total order: field
/// declaration order below already matches the 5-tuple's own significance
/// order (`vtime, region_ordinal, reconv_iter, canonical_process_id,
/// local_send_seq`), so the lexicographic order the derive produces *is* the
/// canonical order — comparing/sorting `CanonicalEventKey` values (e.g.
/// `sched::order::WriteBuffer::resolve_last_write_wins`'s last-write-wins
/// resolution, SC-010/SC-012) never needs a hand-written `Ord` impl or a
/// second key type. Reordering these fields would silently change the total
/// order this type defines — don't.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)]
pub struct CanonicalEventKey {
    pub vtime: SchedTime,
    pub region_ordinal: u32,
    pub reconv_iter: u32,
    pub canonical_process_id: ProcessId,
    pub local_send_seq: u64,
}

/// Review fix: `activation.id` (`sched::PendingActivation::id`) is the defined
/// correlation between a `WorkItem` handed out by `partition()` and the
/// `ActivationId` that `on_read`/`on_write`/`ResolvedBatch::completed` expect
/// — those hooks previously had no way to name the activation a `WorkItem`
/// came from. `key.reconv_iter` (from [`CanonicalEventKey`]) must equal
/// `activation.reconv_iter` (from `PendingActivation`) for the `WorkItem` that
/// produced it; both are threaded through independently because
/// `PendingActivation` is scheduler-owned ready-set data and
/// `CanonicalEventKey` is the WOCB-verdict ordering key, but they describe the
/// same delta-loop pass.
#[derive(Clone, Copy, Debug)]
pub struct WorkItem {
    pub activation: PendingActivation,
    pub key: CanonicalEventKey,
}

#[derive(Clone, Debug, Default)]
pub struct ResolvedBatch {
    pub completed: Vec<ActivationId>,
}

/// A pluggable rollback/ordering mechanism (scheduler.md §6; ledger SC-008,
/// SC-009, SC-013, SC-030, SC-074, SC-076). v1 ships a degenerate sequential
/// controller (`gvt() == now` always, no rollback) as its own test oracle; WOCB
/// plugs in here per-region once each region's own cross-thread-count fuzz
/// harness is green (ARCHITECTURE §4).
pub trait DeterminismController: Send + Sync {
    fn partition(
        &self,
        ready: &[PendingActivation],
        region: Region,
        now: SchedTime,
    ) -> Vec<WorkItem>;
    /// `act` is the `ActivationId` of the `WorkItem` (`partition()`'s own
    /// output, `WorkItem.activation.id`) currently executing the read/write.
    fn on_read(&self, act: ActivationId, cell: StateCellId, version: u64);
    fn on_write(&self, act: ActivationId, cell: StateCellId, prior: FourStateValue, version: u64);
    /// The commit horizon: nothing at/after this may be observed externally.
    fn gvt(&self) -> SchedTime;
    fn drain_and_resolve(&self, work: Vec<WorkItem>) -> ResolvedBatch;
    /// **NEW (Task 18, `wA-sched` plan, SC-083's `Engine::controller_kind`
    /// testability getter — see `sched::control`'s deviations note).** A
    /// short, stable, human-readable tag identifying the concrete controller
    /// implementation, with no other production use (never parsed, never
    /// part of any ordering/commit decision) — added purely so
    /// `Engine::controller_kind()` can prove, from outside, which concrete
    /// `Box<dyn DeterminismController>` is currently installed (the default
    /// `SequentialController` vs. a caller-supplied override), without this
    /// crate resorting to `Any`/downcasting for a testability-only need.
    /// `determinism` is owned by the same `sched` track as Task 18
    /// (`docs/CONTRACTS.md`), so this is an in-track, additive trait
    /// extension, not a foreign-contract change.
    fn kind(&self) -> &'static str;
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sched::ResumeArg;
    use vj_abi::Region;

    /// Review fix: prove a `WorkItem`'s `ActivationId` (what `on_read`/
    /// `on_write`/`ResolvedBatch::completed` expect) is actually recoverable
    /// from what `partition()` hands back, via `activation.id`.
    #[test]
    fn work_item_activation_id_correlates_back_to_the_activation() {
        let now = SchedTime { ticks: 5 };
        let activation = PendingActivation {
            id: 99,
            proc: 1,
            arg: ResumeArg::Start,
            reconv_iter: 0,
            logical_order: (now, Region::Active, 0),
        };
        let key = CanonicalEventKey {
            vtime: now,
            region_ordinal: Region::Active as u32,
            reconv_iter: activation.reconv_iter,
            canonical_process_id: activation.proc,
            local_send_seq: 0,
        };
        let item = WorkItem { activation, key };
        let resolved = ResolvedBatch {
            completed: vec![item.activation.id],
        };
        assert_eq!(resolved.completed[0], item.activation.id);
        assert_eq!(item.key.reconv_iter, item.activation.reconv_iter);
    }

    #[test]
    fn canonical_event_key_is_a_plain_totally_ordered_tuple() {
        let k1 = CanonicalEventKey {
            vtime: SchedTime { ticks: 0 },
            region_ordinal: Region::Active as u32,
            reconv_iter: 0,
            canonical_process_id: 0,
            local_send_seq: 0,
        };
        let k2 = k1;
        assert_eq!(k1, k2);
        assert_eq!(k1.region_ordinal, 0);
    }

    /// Task 1 review fix (SC-129/SC-008): a compile-level proof that
    /// `DeterminismController::partition`'s `now` and `gvt`'s return type are
    /// `sched::SchedTime`, not the ABI-facing `vj_abi::SimTime`. This would
    /// fail to compile against the old `SimTime`-typed signatures (`SchedTime`
    /// has no `delta` field and does not coerce to `SimTime`), which is the
    /// actual guard.
    #[test]
    fn determinism_controller_trait_uses_sched_time_for_now_and_gvt() {
        struct NoopController {
            gvt: SchedTime,
        }
        impl DeterminismController for NoopController {
            fn partition(
                &self,
                _ready: &[PendingActivation],
                _region: Region,
                _now: SchedTime,
            ) -> Vec<WorkItem> {
                vec![]
            }
            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
            }
            fn drain_and_resolve(&self, _work: Vec<WorkItem>) -> ResolvedBatch {
                ResolvedBatch::default()
            }
            fn kind(&self) -> &'static str {
                "noop"
            }
        }

        let controller = NoopController {
            gvt: SchedTime { ticks: 7 },
        };
        let out = controller.partition(&[], Region::Active, SchedTime { ticks: 7 });
        assert!(out.is_empty());
        assert_eq!(controller.gvt(), SchedTime { ticks: 7 });
    }

    /// Task 22 (SC-130) — the actual executable proof, not only the doc
    /// review above: a **second**, test-only `DeterminismController` whose
    /// `partition` sorts by a fake "level" (`canonical_process_id % 3`,
    /// standing in for a real static levelization) instead of executing
    /// sequentially, compiles and satisfies the trait with zero changes to
    /// this file's trait definition, and round-trips through `Engine::
    /// set_determinism_controller` without error. This
    /// `FakeLevelizedController` is local to this test and independently
    /// declared from `tests/region_order_fixture.rs`'s own copy (see that
    /// file's doc comment: an integration test cannot reach a private
    /// in-crate test type, so both are separate instantiations of the same
    /// trivial shape, not one implementation shared across a crate
    /// boundary — the established pattern this plan already uses for
    /// `TestArena`/`FakeArena`).
    #[test]
    fn hypothetical_levelized_controller_compiles_against_the_unmodified_trait() {
        struct FakeLevelizedController;

        impl DeterminismController for FakeLevelizedController {
            fn partition(
                &self,
                ready: &[PendingActivation],
                region: Region,
                _now: SchedTime,
            ) -> Vec<WorkItem> {
                let mut sorted: Vec<PendingActivation> = ready.to_vec();
                sorted.sort_by_key(|a| a.proc % 3);
                sorted
                    .into_iter()
                    .enumerate()
                    .map(|(i, activation)| {
                        let key = CanonicalEventKey {
                            vtime: activation.logical_order.0,
                            region_ordinal: crate::sched::region_ordinal(region),
                            reconv_iter: activation.reconv_iter,
                            canonical_process_id: activation.proc,
                            local_send_seq: i as u64,
                        };
                        WorkItem { 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 {
                SchedTime { ticks: 0 }
            }
            fn drain_and_resolve(&self, work: Vec<WorkItem>) -> ResolvedBatch {
                ResolvedBatch {
                    completed: work.iter().map(|w| w.activation.id).collect(),
                }
            }
            fn kind(&self) -> &'static str {
                "fake-levelized"
            }
        }

        // Round-trip through Engine::set_determinism_controller (Task 18) --
        // needs a minimal SignalAccess/IoSink pair Engine::new requires;
        // independently declared (Engine's own #[cfg(test)] fixtures are
        // private to sched::control, unreachable from here).
        struct NoopArena;
        impl crate::state::SignalAccess for NoopArena {
            fn read(&self, _cell: StateCellId) -> FourStateValue {
                FourStateValue::Narrow(0)
            }
            fn write(&mut self, _cell: StateCellId, _value: FourStateValue) -> bool {
                true
            }
            fn subscribe(
                &mut self,
                _cell: StateCellId,
                _role: crate::state::ListenerRole,
                _sink: Box<dyn crate::state::ValueChangeSink + Send>,
            ) {
            }
        }
        struct NoopIoSink;
        impl crate::sched::IoSink for NoopIoSink {
            fn enqueue(
                &self,
                _at: SchedTime,
                _region: Region,
                _seq: u64,
                cb: crate::sched::DeferredIoCallback,
            ) {
                drop(cb);
            }
        }

        use crate::sched::SchedulerControl;
        let mut engine = crate::sched::Engine::new(Box::new(NoopArena), Box::new(NoopIoSink));
        assert_eq!(
            engine.controller_kind(),
            "sequential",
            "engine must still default to SequentialController before this test installs \
             the hypothetical levelized one"
        );
        engine.set_determinism_controller(Box::new(FakeLevelizedController));
        assert_eq!(
            engine.controller_kind(),
            "fake-levelized",
            "set_determinism_controller must have genuinely swapped the installed \
             controller -- this is the round-trip proof, not just a compile check"
        );
    }

    /// Task 22 (SC-042 review fix, sixth review pass) — the crate-wide
    /// completion of Task 7/8's own single-file checks: `notify_change` is
    /// called from exactly two sanctioned production sites in the whole
    /// crate (`notify.rs`'s `commit_nba`, `ops.rs`'s
    /// `commit_blocking_write`), never a third ad-hoc caller.
    ///
    /// **Extended, B-sched Task 6 (ledger SC-069):** `control.rs`'s
    /// `Engine::advance_time_to` is now a real, SANCTIONED third call site —
    /// a delayed named-event trigger (`->>#N my_event;`'s wheel-fired
    /// `TimerAction::TriggerEvent`) fans out through this exact same
    /// `notify_change` path (never a second, hand-rolled matching
    /// implementation of its own), the literal reduction this task's own
    /// plan text names ("the exact same fan-out `EngineNotifySink::
    /// notify_change` already does for an ordinary committed write"). This
    /// is additive to SC-042's own guarantee, not a weakening of it:
    /// `SensitivityTable::matches` (the one canonical matching
    /// implementation SC-042 actually protects) still has exactly one
    /// production consumer function, `notify_change` itself — this gate now
    /// pins the count at exactly THREE call sites of that one function,
    /// still rejecting any fourth, still-ad-hoc caller.
    ///
    /// **Deviation from the plan's literal scan (documented here and in
    /// `docs/CONTRACTS.md`'s r3 Deviations section, item 10):** a raw
    /// substring count of `".notify_change("` over `scan_support::
    /// strip_cfg_test_blocks`'s output alone over-counts against this
    /// crate's *real* source: `notify.rs`'s and `ops.rs`'s own top-of-file
    /// doc comments (`//! ... .notify_change(\` call site`,
    /// `/// calls \`sink.notify_change(..)\` ...`) quote the exact call
    /// pattern in prose, outside any `#[cfg(test)]` block, which
    /// `strip_cfg_test_blocks` never touches (it only strips test *code*
    /// blocks, not comments). Left as a plain substring count, this test
    /// would read a count of 6 (3 in each of `notify.rs`/`ops.rs`: two doc
    /// mentions plus the one real call), never `2`, on the current,
    /// already-shipped, correct source — an unconditional false failure, not
    /// a real defect. Fixed by additionally stripping every line whose
    /// trimmed text starts with `//` (covers `//`, `///`, `//!`) before
    /// counting, so only text that could actually execute is scanned. This
    /// does not weaken the check: a real call site is never written as a
    /// same-line trailing comment in this crate (verified: every real
    /// `.notify_change(` call above sits on its own statement line), so no
    /// production call this test needs to catch is hidden by comment
    /// stripping; a doc comment merely mentioning the pattern (exactly what
    /// trips the unpatched version) is correctly excluded either way.
    #[test]
    fn notify_change_has_exactly_three_call_sites_in_the_crate() {
        let sched_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src/sched");
        let mut hits: Vec<(String, usize)> = Vec::new();
        scan_dir_for_notify_change_calls(&sched_dir, &mut hits);

        assert!(
            !hits.is_empty(),
            "scan matched no files under {sched_dir:?} -- the scan itself is broken"
        );

        let total: usize = hits.iter().map(|(_, n)| n).sum();
        assert_eq!(
            total, 3,
            "expected exactly three production `.notify_change(` call sites crate-wide \
             (SC-042 + B-sched Task 6's SC-069 addition); found: {hits:?}"
        );

        let count_for = |suffix: &str| -> usize {
            hits.iter()
                .filter(|(f, _)| f.ends_with(suffix))
                .map(|(_, n)| *n)
                .sum()
        };
        assert_eq!(
            count_for("notify.rs"),
            1,
            "notify.rs (Task 7's commit_nba) must be exactly one of the three sanctioned \
             call sites: {hits:?}"
        );
        assert_eq!(
            count_for("ops.rs"),
            1,
            "ops.rs (Task 8's commit_blocking_write) must be exactly one of the three \
             sanctioned call sites: {hits:?}"
        );
        assert_eq!(
            count_for("control.rs"),
            1,
            "control.rs (B-sched Task 6's advance_time_to, SC-069's TriggerEvent fan-out) \
             must be exactly one of the three sanctioned call sites: {hits:?}"
        );
        for (file, n) in &hits {
            if !file.ends_with("notify.rs")
                && !file.ends_with("ops.rs")
                && !file.ends_with("control.rs")
            {
                assert_eq!(
                    *n, 0,
                    "unexpected production `.notify_change(` call site outside \
                     notify.rs/ops.rs/control.rs: {file} (SC-042: a fourth ad-hoc caller is a \
                     design defect) -- full scan: {hits:?}"
                );
            }
        }
    }

    /// Strips every line whose trimmed text starts with `//` -- see this
    /// module's own doc comment on `notify_change_has_exactly_three_call_
    /// sites_in_the_crate` for why this is layered on top of (never a
    /// replacement for) `scan_support::strip_cfg_test_blocks`.
    fn strip_line_comments(src: &str) -> String {
        src.lines()
            .filter(|line| !line.trim_start().starts_with("//"))
            .collect::<Vec<_>>()
            .join("\n")
    }

    fn scan_dir_for_notify_change_calls(dir: &std::path::Path, hits: &mut Vec<(String, usize)>) {
        for entry in std::fs::read_dir(dir).expect("read sched/ dir for notify_change scan") {
            let entry = entry.expect("dir entry for notify_change scan");
            let path = entry.path();
            if path.is_dir() {
                scan_dir_for_notify_change_calls(&path, hits);
                continue;
            }
            if path.extension().is_none_or(|e| e != "rs") {
                continue;
            }
            let src = std::fs::read_to_string(&path)
                .unwrap_or_else(|e| panic!("read {path:?} for notify_change scan: {e}"));
            let production_only = strip_line_comments(&scan_support::strip_cfg_test_blocks(&src));
            let count = production_only.matches(".notify_change(").count();
            hits.push((path.display().to_string(), count));
        }
    }
}