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//!
7//! Writes in this process still publish immediately through the store hook.
8//! Writes from another process reach the same SQLite file; [`super::session_wal_watch`]
9//! notices those via WAL + `PRAGMA data_version` and publishes through the
10//! same fanout.
11
12use std::path::Path;
13use std::sync::atomic::{AtomicU64, Ordering};
14use std::sync::{Arc, RwLock, RwLockWriteGuard};
15
16use harn_session_store::{SessionMeta, SharedSessionChangeObserver};
17
18use super::session_wal_watch;
19
20/// Process-wide observers notified when session metadata is committed.
21///
22/// Deliberately process-scoped, not thread-local like the redaction policy
23/// beside it: a store handle is opened on whatever thread needs it, and an
24/// update commits on whatever executor thread the caller happens to be on. A
25/// thread-local sink would be installed on one thread and silently miss every
26/// write made from another, which reads as "notifications do not work" rather
27/// than as a wiring mistake.
28///
29/// A list rather than one slot, because more than one surface can care about
30/// the same store and a single slot makes the second registration silently
31/// evict the first.
32static OBSERVERS: RwLock<Vec<(u64, SharedSessionChangeObserver)>> = RwLock::new(Vec::new());
33static NEXT_SUBSCRIPTION: AtomicU64 = AtomicU64::new(1);
34type TitleFingerprint = (Option<String>, bool);
35type RememberedTitles = Vec<(String, TitleFingerprint)>;
36static TITLES: RwLock<RememberedTitles> = RwLock::new(Vec::new());
37
38fn observers() -> RwLockWriteGuard<'static, Vec<(u64, SharedSessionChangeObserver)>> {
39    OBSERVERS
40        .write()
41        .unwrap_or_else(|poisoned| poisoned.into_inner())
42}
43
44fn titles() -> RwLockWriteGuard<'static, RememberedTitles> {
45    TITLES
46        .write()
47        .unwrap_or_else(|poisoned| poisoned.into_inner())
48}
49
50/// Live registration for [`subscribe`]. Unregisters on drop so
51/// a surface that goes away cannot keep receiving, and so a test cannot leak a
52/// sink into the next one sharing the process.
53#[must_use = "dropping the subscription immediately unregisters the observer"]
54pub struct SessionChangeSubscription {
55    id: u64,
56}
57
58impl Drop for SessionChangeSubscription {
59    fn drop(&mut self) {
60        observers().retain(|(id, _)| *id != self.id);
61        let live = subscriber_count() > 0;
62        if !live {
63            titles().clear();
64        }
65        session_wal_watch::sync_watchers(live);
66    }
67}
68
69/// Register an observer notified after any session metadata update commits
70/// through a store this VM opens, or after a WAL watcher sees another
71/// process rename a session in the same file.
72///
73/// Every canonical store opened here carries the registered observers, so a
74/// surface does not have to be handed the specific handle a writer happened to
75/// use.
76pub fn subscribe(observer: SharedSessionChangeObserver) -> SessionChangeSubscription {
77    let id = NEXT_SUBSCRIPTION.fetch_add(1, Ordering::Relaxed);
78    observers().push((id, observer));
79    session_wal_watch::sync_watchers(true);
80    SessionChangeSubscription { id }
81}
82
83/// Remember a canonical store file so a live subscription can watch it.
84pub(crate) fn watch_store(path: &Path) {
85    session_wal_watch::register_store_path(path);
86    if subscriber_count() > 0 {
87        session_wal_watch::sync_watchers(true);
88    }
89}
90
91fn subscriber_count() -> usize {
92    OBSERVERS
93        .read()
94        .unwrap_or_else(|poisoned| poisoned.into_inner())
95        .len()
96}
97
98/// Whether a title/pin pair is new to this process, unchanged, or moved.
99#[derive(Clone, Copy, Debug, PartialEq, Eq)]
100pub(super) enum TitleMemory {
101    New,
102    Unchanged,
103    Changed,
104}
105
106pub(super) fn remember_title(
107    session_id: &str,
108    title: Option<&str>,
109    title_pinned: bool,
110) -> TitleMemory {
111    let next = (title.map(str::to_string), title_pinned);
112    let mut titles = titles();
113    if let Some((_, previous)) = titles.iter_mut().find(|(id, _)| id == session_id) {
114        if *previous == next {
115            return TitleMemory::Unchanged;
116        }
117        *previous = next;
118        return TitleMemory::Changed;
119    }
120    titles.push((session_id.to_string(), next));
121    TitleMemory::New
122}
123
124/// Fans one committed change out to every live subscriber.
125pub(super) fn dispatch(meta: &SessionMeta) {
126    let observers: Vec<SharedSessionChangeObserver> = OBSERVERS
127        .read()
128        .unwrap_or_else(|poisoned| poisoned.into_inner())
129        .iter()
130        .map(|(_, observer)| Arc::clone(observer))
131        .collect();
132    // Copy out before dispatching: an observer is allowed to subscribe or
133    // unsubscribe in response, which would deadlock against a held guard.
134    for observer in observers {
135        observer.session_updated(meta);
136    }
137}
138
139struct SessionChangeFanout;
140
141impl harn_session_store::SessionChangeObserver for SessionChangeFanout {
142    fn session_updated(&self, meta: &SessionMeta) {
143        remember_title(&meta.id, meta.title.as_deref(), meta.title_pinned);
144        dispatch(meta);
145    }
146}
147
148pub(crate) fn current_observer() -> Option<SharedSessionChangeObserver> {
149    if subscriber_count() == 0 {
150        return None;
151    }
152    Some(Arc::new(SessionChangeFanout))
153}
harn_vm/stdlib/session_change.rs

harn_vm/stdlib/
session_change.rs
