core_api/
subscription.rs

1//! Post-commit subscription API.
2//!
3//! [`GraphDb`] exposes three subscription entry-points:
4//! - [`GraphDb::subscribe_rule`] — edge-fire / retract events for one named rule.
5//! - [`GraphDb::subscribe_all_rules`] — edge events for every rule.
6//! - [`GraphDb::subscribe_writes`] — node and property mutations.
7//!
8//! All three return a [`Subscription`] handle. Dropping it unregisters the
9//! subscriber on the next commit (via [`Weak`] upgrade failure in the distribution
10//! loop).
11//!
12//! # Ordering invariant
13//!
14//! Events are pushed inside `log_then_apply_with` **after** the WAL fsync and
15//! the in-memory `apply` have both completed. A subscriber that queries the db
16//! immediately after receiving an event therefore observes the state that
17//! produced it.
18//!
19//! # Bounded queue / Lagged
20//!
21//! Each subscription has a fixed-capacity queue (default [`DEFAULT_SUB_CAPACITY`]).
22//! When the queue is full, events are dropped and a missed count is incremented.
23//! The next [`Subscription::try_recv`] / [`Subscription::recv_timeout`] call
24//! that finds an empty queue and a non-zero miss count returns
25//! [`DbEvent::Lagged { missed }`] before continuing with queued events.
26//!
27//! # v1 scope
28//!
29//! Rule-edge events and write-mutation events only. Incremental query
30//! subscriptions (differential dataflow) are roadmap (Plan-15 T3+).
31
32use core_storage::Value;
33use serde::Serialize;
34use std::collections::VecDeque;
35use std::sync::{Arc, Condvar, Mutex, Weak};
36use std::time::Duration;
37
38/// Default per-subscriber queue capacity.
39pub const DEFAULT_SUB_CAPACITY: usize = 65_536;
40
41/// A post-commit event delivered to subscribers.
42///
43/// Serialises as internally-tagged JSON (`"type"` discriminant, snake_case).
44///
45/// ```json
46/// {"type":"edge_fired","rule":"skill_fit","src_key":"p1","dst_key":"proj-01",
47///  "edge_type":"FIT","weight":0.87,"commit_seq":42}
48/// {"type":"lagged","missed":3}
49/// ```
50#[derive(Debug, Clone, Serialize, PartialEq)]
51#[serde(tag = "type", rename_all = "snake_case")]
52pub enum DbEvent {
53    /// A rule derived a new edge.
54    EdgeFired {
55        rule: String,
56        src_key: String,
57        dst_key: String,
58        edge_type: String,
59        #[serde(skip_serializing_if = "Option::is_none")]
60        weight: Option<f64>,
61        commit_seq: u64,
62    },
63    /// A rule retracted a previously derived edge.
64    EdgeRetracted {
65        rule: String,
66        src_key: String,
67        dst_key: String,
68        edge_type: String,
69        commit_seq: u64,
70    },
71    /// A node was inserted.
72    NodeInserted {
73        label: String,
74        key: String,
75        commit_seq: u64,
76    },
77    /// A node was deleted.
78    NodeDeleted { key: String, commit_seq: u64 },
79    /// A user-inserted edge was added.
80    EdgeInserted {
81        edge_type: String,
82        src: String,
83        dst: String,
84        commit_seq: u64,
85    },
86    /// A user-inserted edge was deleted.
87    EdgeDeleted {
88        edge_type: String,
89        src: String,
90        dst: String,
91        commit_seq: u64,
92    },
93    /// A property was set on a node.
94    PropSet {
95        key: String,
96        field: String,
97        commit_seq: u64,
98    },
99    /// A property was removed from a node.
100    PropRemoved {
101        key: String,
102        field: String,
103        commit_seq: u64,
104    },
105    /// One or more events were dropped due to a full queue.
106    ///
107    /// The subscriber must re-read graph state to recover consistency for
108    /// lossless consumers.  `missed` is the count of dropped events.
109    Lagged { missed: u64 },
110    /// A row appeared in the result of a `subscribe_query` subscription.
111    ///
112    /// Emitted after each commit when a full re-run of the subscribed Cypher
113    /// query returns a row that was absent in the previous run.
114    ///
115    /// **Full re-run per commit; use LIMIT to bound execution cost.**
116    QueryRowAdded {
117        columns: Vec<String>,
118        row: Vec<Option<Value>>,
119    },
120    /// A row disappeared from the result of a `subscribe_query` subscription.
121    ///
122    /// Emitted after each commit when a row present in the previous run is
123    /// absent from the current run.
124    ///
125    /// **Full re-run per commit; use LIMIT to bound execution cost.**
126    QueryRowRemoved {
127        columns: Vec<String>,
128        row: Vec<Option<Value>>,
129    },
130}
131
132// ---------------------------------------------------------------------------
133// Internal queue
134// ---------------------------------------------------------------------------
135
136pub(crate) struct SubInner {
137    mu: Mutex<SubQueue>,
138    condvar: Condvar,
139}
140
141impl std::fmt::Debug for SubInner {
142    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
143        f.debug_struct("SubInner").finish_non_exhaustive()
144    }
145}
146
147struct SubQueue {
148    items: VecDeque<DbEvent>,
149    missed: u64,
150    capacity: usize,
151}
152
153impl SubInner {
154    pub(crate) fn new(capacity: usize) -> Arc<Self> {
155        Arc::new(SubInner {
156            mu: Mutex::new(SubQueue {
157                items: VecDeque::new(),
158                missed: 0,
159                capacity,
160            }),
161            condvar: Condvar::new(),
162        })
163    }
164
165    /// Push an event, dropping it (incrementing `missed`) if the queue is full.
166    ///
167    /// `notify_one` is only called when an item is actually enqueued.  On
168    /// overflow we increment `missed` but skip the notification: no waiter can
169    /// consume a dropped event, and the spurious wakeup just wastes a syscall.
170    pub(crate) fn push(&self, event: DbEvent) {
171        let mut q = self.mu.lock().unwrap();
172        if q.items.len() >= q.capacity {
173            q.missed += 1;
174            // No notify: the dropped event cannot be consumed.
175            return;
176        }
177        q.items.push_back(event);
178        drop(q);
179        self.condvar.notify_one();
180    }
181
182    fn pop_one(q: &mut SubQueue) -> Option<DbEvent> {
183        if let Some(item) = q.items.pop_front() {
184            return Some(item);
185        }
186        if q.missed > 0 {
187            let missed = std::mem::take(&mut q.missed);
188            return Some(DbEvent::Lagged { missed });
189        }
190        None
191    }
192
193    /// Non-blocking read.
194    pub(crate) fn try_recv(&self) -> Option<DbEvent> {
195        let mut q = self.mu.lock().unwrap();
196        Self::pop_one(&mut q)
197    }
198
199    /// Blocking read with deadline.  Returns `None` on timeout.
200    pub(crate) fn recv_timeout(&self, timeout: Duration) -> Option<DbEvent> {
201        let mut q = self.mu.lock().unwrap();
202        let deadline = std::time::Instant::now() + timeout;
203        loop {
204            if let Some(item) = Self::pop_one(&mut q) {
205                return Some(item);
206            }
207            let now = std::time::Instant::now();
208            if now >= deadline {
209                return None;
210            }
211            let remaining = deadline - now;
212            let (q2, timed_out) = self.condvar.wait_timeout(q, remaining).unwrap();
213            q = q2;
214            if timed_out.timed_out() {
215                // One last try in case a push arrived just as we timed out.
216                return Self::pop_one(&mut q);
217            }
218        }
219    }
220}
221
222// ---------------------------------------------------------------------------
223// Filter
224// ---------------------------------------------------------------------------
225
226/// What events a subscriber receives.
227pub(crate) enum SubFilter {
228    /// Only edge events for the named rule.
229    Rule(String),
230    /// All rule edge events.
231    AllRules,
232    /// Write events only (node/prop mutations; no edge-fire/retract).
233    Writes,
234}
235
236pub(crate) fn event_matches(event: &DbEvent, filter: &SubFilter) -> bool {
237    match filter {
238        SubFilter::Rule(name) => match event {
239            DbEvent::EdgeFired { rule, .. } | DbEvent::EdgeRetracted { rule, .. } => rule == name,
240            _ => false,
241        },
242        SubFilter::AllRules => {
243            matches!(
244                event,
245                DbEvent::EdgeFired { .. } | DbEvent::EdgeRetracted { .. }
246            )
247        }
248        SubFilter::Writes => matches!(
249            event,
250            DbEvent::NodeInserted { .. }
251                | DbEvent::NodeDeleted { .. }
252                | DbEvent::EdgeInserted { .. }
253                | DbEvent::EdgeDeleted { .. }
254                | DbEvent::PropSet { .. }
255                | DbEvent::PropRemoved { .. }
256        ),
257    }
258}
259
260// ---------------------------------------------------------------------------
261// Registry entry
262// ---------------------------------------------------------------------------
263
264pub(crate) struct SubEntry {
265    pub(crate) filter: SubFilter,
266    pub(crate) inner: Weak<SubInner>,
267}
268
269// ---------------------------------------------------------------------------
270// Public handle
271// ---------------------------------------------------------------------------
272
273/// A live subscription handle returned by [`GraphDb::subscribe_rule`] etc.
274///
275/// Dropping this value unregisters the subscriber: the next commit will detect
276/// the dead [`Weak`] reference and prune the entry, so no resources leak.
277#[derive(Clone, Debug)]
278pub struct Subscription(pub(crate) Arc<SubInner>);
279
280impl Subscription {
281    /// Non-blocking read.  Returns `None` if the queue is empty.
282    pub fn try_recv(&self) -> Option<DbEvent> {
283        self.0.try_recv()
284    }
285
286    /// Blocking read with timeout.  Returns `None` on timeout.
287    ///
288    /// Use `tokio::task::spawn_blocking` to bridge into an async context.
289    pub fn recv_timeout(&self, timeout: Duration) -> Option<DbEvent> {
290        self.0.recv_timeout(timeout)
291    }
292}
core_api/subscription.rs

core_api/
subscription.rs
