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