dora_node_api/event_stream/
extensions.rs

1//! Drop notifications for the daemon's opaque extension table.
2//!
3//! When any node drops an extension key, the daemon sends
4//! [`NodeEvent::ExtensionDropped`](dora_message::daemon_to_node::NodeEvent::ExtensionDropped)
5//! to every node that stored or loaded it. That event is out-of-band — it is
6//! not a dataflow input, and surfacing it to user code would mean every node
7//! author had to match on an event they did not ask for. So the event-stream
8//! thread consumes it into this process-global queue, and the extension's own
9//! code drains it whenever it next runs.
10//!
11//! Process-global rather than per-node because a language binding holds its
12//! caches the same way: one process, one set of mappings, regardless of how
13//! many `DoraNode`s live in it.
14
15use std::collections::VecDeque;
16use std::sync::{LazyLock, Mutex};
17
18/// Bound on the queue. A consumer that never drains must not grow it without
19/// limit; dropping the oldest entry is safe because a missed notification only
20/// means the extension releases that resource later (on its own `drop`) rather
21/// than promptly.
22const MAX_PENDING: usize = 4096;
23
24type Dropped = (String, String);
25
26static DROPPED: LazyLock<Mutex<VecDeque<Dropped>>> = LazyLock::new(|| Mutex::new(VecDeque::new()));
27
28/// Record a dropped `(namespace, key)`. Called by the event-stream thread.
29pub(crate) fn push_dropped(namespace: String, key: String) {
30    let mut queue = DROPPED.lock().unwrap_or_else(|e| e.into_inner());
31    if queue.len() >= MAX_PENDING {
32        queue.pop_front();
33    }
34    queue.push_back((namespace, key));
35}
36
37/// Take every pending drop notification for `namespace`, leaving the rest.
38///
39/// Scoped by namespace so two extensions in one process cannot swallow each
40/// other's notifications.
41pub fn drain_dropped_keys(namespace: &str) -> Vec<String> {
42    let mut queue = DROPPED.lock().unwrap_or_else(|e| e.into_inner());
43    let mut taken = Vec::new();
44    queue.retain(|(ns, key)| {
45        if ns == namespace {
46            taken.push(key.clone());
47            false
48        } else {
49            true
50        }
51    });
52    taken
53}
54
55#[cfg(test)]
56mod tests {
57    use super::*;
58
59    /// The queue is process-global by design, so these tests would clobber
60    /// each other under cargo's default thread-per-test. Serialize them and
61    /// start each from empty.
62    static TEST_LOCK: Mutex<()> = Mutex::new(());
63
64    fn guard() -> std::sync::MutexGuard<'static, ()> {
65        let g = TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner());
66        DROPPED.lock().unwrap_or_else(|e| e.into_inner()).clear();
67        g
68    }
69
70    #[test]
71    fn drain_returns_only_the_requested_namespace() {
72        let _g = guard();
73        push_dropped("pool".into(), "a".into());
74        push_dropped("other".into(), "b".into());
75        push_dropped("pool".into(), "c".into());
76
77        assert_eq!(drain_dropped_keys("pool"), vec!["a", "c"]);
78        // The other namespace's entry survived rather than being consumed.
79        assert_eq!(drain_dropped_keys("other"), vec!["b"]);
80    }
81
82    #[test]
83    fn drain_is_exhaustive() {
84        let _g = guard();
85        push_dropped("pool".into(), "a".into());
86        assert_eq!(drain_dropped_keys("pool"), vec!["a"]);
87        assert!(drain_dropped_keys("pool").is_empty());
88    }
89
90    #[test]
91    fn queue_is_bounded_and_drops_oldest() {
92        let _g = guard();
93        for i in 0..MAX_PENDING + 10 {
94            push_dropped("pool".into(), i.to_string());
95        }
96        let drained = drain_dropped_keys("pool");
97        assert_eq!(drained.len(), MAX_PENDING);
98        // The oldest ten were evicted, so the window starts at 10.
99        assert_eq!(drained.first().map(String::as_str), Some("10"));
100    }
101}
dora_node_api/event_stream/extensions.rs

dora_node_api/event_stream/
extensions.rs
