khive_runtime/
config_ledger.rs

1//! ADR-094 process-lifetime config lock ledger.
2//!
3//! Several `OnceLock`-backed config readers across packs (recall profiling,
4//! ANN overfetch rounds, context profiling, ...) resolve an environment
5//! variable exactly once per process and hold it for the process lifetime.
6//! That first resolution is itself an auditable lifecycle event ("this
7//! process locked config key K to value V"), but the `OnceLock::get_or_init`
8//! closures that produce it run synchronously, off any async/event-store
9//! context, and long before a `VerbRegistry` exists.
10//!
11//! This module bridges the two: `record_config_locked` is a plain sync
12//! function any `OnceLock` closure can call to enqueue a `(key, value)` pair
13//! into a process-wide queue. The registry's dispatch path drains the queue
14//! (`take_config_locked`) at the same gate where it already appends audit
15//! events, so `ConfigLocked` rows carry the namespace/actor of whichever
16//! dispatch happens to observe the queue non-empty first (ADR-094's accepted
17//! provenance quirk) rather than needing every `OnceLock` site to carry its
18//! own `EventStore` handle.
19
20use std::sync::atomic::{AtomicBool, Ordering};
21use std::sync::Mutex;
22
23/// `true` once at least one pair has been enqueued and not yet drained.
24///
25/// Read via `swap(false, Ordering::AcqRel)` on the dispatch hot path so the
26/// overwhelmingly common empty-ledger case pays for a single atomic
27/// read-and-clear instead of locking `LEDGER`.
28pub(crate) static PENDING: AtomicBool = AtomicBool::new(false);
29
30static LEDGER: Mutex<Vec<(String, String)>> = Mutex::new(Vec::new());
31
32/// Enqueue a config key/value pair for later `ConfigLocked` event emission.
33///
34/// Safe to call from a synchronous `OnceLock::get_or_init` closure — this
35/// only takes a `std::sync::Mutex`, never awaits, and never fails.
36pub fn record_config_locked(key: &'static str, value: impl Into<String>) {
37    LEDGER
38        .lock()
39        .unwrap_or_else(|poisoned| poisoned.into_inner())
40        .push((key.to_string(), value.into()));
41    // Set after enqueue so a dispatch that observes `true` is guaranteed to
42    // find the pair already in `LEDGER`.
43    PENDING.store(true, Ordering::Release);
44}
45
46/// Drain every queued config-locked pair for emission as `ConfigLocked` events.
47///
48/// Crate-private: only the dispatch path that owns event-store persistence
49/// should drain this queue.
50pub(crate) fn drain_config_locked() -> Vec<(String, String)> {
51    std::mem::take(
52        &mut *LEDGER
53            .lock()
54            .unwrap_or_else(|poisoned| poisoned.into_inner()),
55    )
56}
57
58/// `true` if at least one config-locked pair is queued, without draining it.
59///
60/// Test-only observability into the atomic fast path; the dispatch path
61/// itself swaps `PENDING` directly rather than calling this.
62#[cfg(test)]
63pub(crate) fn has_pending_config_locked() -> bool {
64    PENDING.load(Ordering::Acquire)
65}
66
67#[cfg(test)]
68mod tests {
69    use super::*;
70    use serial_test::serial;
71
72    // `PENDING`/`LEDGER` are process-wide singletons; serialize these tests
73    // against each other so one test's queued pairs never leak into
74    // another's assertions.
75    #[test]
76    #[serial(config_ledger)]
77    fn record_then_drain_returns_queued_pairs_in_order() {
78        drain_config_locked(); // drain any leftovers from a prior test run
79        PENDING.store(false, Ordering::Release);
80        assert!(!has_pending_config_locked());
81
82        record_config_locked("recall_profile_enabled", "true");
83        record_config_locked("ann_overfetch_max_rounds", "3");
84
85        assert!(has_pending_config_locked());
86        let drained = drain_config_locked();
87        assert_eq!(
88            drained,
89            vec![
90                ("recall_profile_enabled".to_string(), "true".to_string()),
91                ("ann_overfetch_max_rounds".to_string(), "3".to_string()),
92            ]
93        );
94    }
95
96    /// Mirrors the dispatch fast path: rows drain exactly once via the atomic
97    /// swap-and-check, and a second dispatch observes no pending work without
98    /// needing to lock `LEDGER`.
99    #[test]
100    #[serial(config_ledger)]
101    fn dispatch_fast_path_drains_exactly_once_then_reports_no_pending() {
102        drain_config_locked();
103        PENDING.store(false, Ordering::Release);
104
105        record_config_locked("context_profile_enabled", "false");
106
107        assert!(PENDING.swap(false, Ordering::AcqRel));
108        let drained = drain_config_locked();
109        assert_eq!(
110            drained,
111            vec![("context_profile_enabled".to_string(), "false".to_string())]
112        );
113
114        assert!(
115            !PENDING.swap(false, Ordering::AcqRel),
116            "a second dispatch must observe the flag already cleared"
117        );
118        assert!(
119            drain_config_locked().is_empty(),
120            "nothing should be left to re-emit on a later dispatch"
121        );
122    }
123}
khive_runtime/config_ledger.rs

khive_runtime/
config_ledger.rs
