harn_vm/stdlib/
session_change.rs

1//! Process-wide registry of observers for committed session-metadata changes.
2//!
3//! Split out of `session_store` rather than living beside the store opener: it
4//! is a notification concern, not a storage one, and the two have no shared
5//! state beyond the hook the opener attaches.
6
7use std::sync::atomic::{AtomicU64, Ordering};
8use std::sync::{Arc, RwLock, RwLockWriteGuard};
9
10use harn_session_store::SharedSessionChangeObserver;
11
12/// Process-wide observers notified when session metadata is committed.
13///
14/// Deliberately process-scoped, not thread-local like the redaction policy
15/// beside it: a store handle is opened on whatever thread needs it, and an
16/// update commits on whatever executor thread the caller happens to be on. A
17/// thread-local sink would be installed on one thread and silently miss every
18/// write made from another, which reads as "notifications do not work" rather
19/// than as a wiring mistake.
20///
21/// A list rather than one slot, because more than one surface can care about
22/// the same store and a single slot makes the second registration silently
23/// evict the first.
24static OBSERVERS: RwLock<Vec<(u64, SharedSessionChangeObserver)>> = RwLock::new(Vec::new());
25static NEXT_SUBSCRIPTION: AtomicU64 = AtomicU64::new(1);
26
27fn observers() -> RwLockWriteGuard<'static, Vec<(u64, SharedSessionChangeObserver)>> {
28    OBSERVERS
29        .write()
30        .unwrap_or_else(|poisoned| poisoned.into_inner())
31}
32
33/// Live registration for [`subscribe`]. Unregisters on drop so
34/// a surface that goes away cannot keep receiving, and so a test cannot leak a
35/// sink into the next one sharing the process.
36#[must_use = "dropping the subscription immediately unregisters the observer"]
37pub struct SessionChangeSubscription {
38    id: u64,
39}
40
41impl Drop for SessionChangeSubscription {
42    fn drop(&mut self) {
43        observers().retain(|(id, _)| *id != self.id);
44    }
45}
46
47/// Register an observer notified after any session metadata update commits
48/// through a store this VM opens.
49///
50/// Every canonical store opened here carries the registered observers, so a
51/// surface does not have to be handed the specific handle a writer happened to
52/// use. Scope is this process only: a write from another process reaches the
53/// same database file but no in-process sink.
54pub fn subscribe(observer: SharedSessionChangeObserver) -> SessionChangeSubscription {
55    let id = NEXT_SUBSCRIPTION.fetch_add(1, Ordering::Relaxed);
56    observers().push((id, observer));
57    SessionChangeSubscription { id }
58}
59
60/// Fans one committed change out to every live subscriber.
61struct SessionChangeFanout;
62
63impl harn_session_store::SessionChangeObserver for SessionChangeFanout {
64    fn session_updated(&self, meta: &harn_session_store::SessionMeta) {
65        let observers: Vec<SharedSessionChangeObserver> = OBSERVERS
66            .read()
67            .unwrap_or_else(|poisoned| poisoned.into_inner())
68            .iter()
69            .map(|(_, observer)| Arc::clone(observer))
70            .collect();
71        // Copy out before dispatching: an observer is allowed to subscribe or
72        // unsubscribe in response, which would deadlock against a held guard.
73        for observer in observers {
74            observer.session_updated(meta);
75        }
76    }
77}
78
79pub(crate) fn current_observer() -> Option<SharedSessionChangeObserver> {
80    let empty = OBSERVERS
81        .read()
82        .unwrap_or_else(|poisoned| poisoned.into_inner())
83        .is_empty();
84    if empty {
85        return None;
86    }
87    Some(Arc::new(SessionChangeFanout))
88}
harn_vm/stdlib/session_change.rs

harn_vm/stdlib/
session_change.rs
