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