rindle_replica/
query.rs

1//! The `Query` handle + its subscription bookkeeping.
2
3use std::rc::Rc;
4
5use rindle::graph::NodeId;
6use rindle::Ast;
7
8use crate::{ChangeEvent, Inner, QueryId, TxId, Update};
9
10/// A query's subscriber callbacks. Factored into an alias (clippy::type_complexity) and
11/// reused by the writer, which takes them out of the entry before firing to avoid a
12/// re-borrow deadlock on `subs`.
13pub(crate) type Callbacks = Vec<Box<dyn FnMut(&Update)>>;
14
15/// The `Db`-internal registration id `subs` is keyed by — STABLE across a delta-overflow
16/// shed rebuild (design 306 D4), unlike the engine's generational sink `NodeId`, which a
17/// rebuild replaces. The handle → entry association must survive the rebuild so a
18/// pre-shed [`Query`] handle keeps working.
19#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Debug)]
20pub(crate) struct RegId(pub(crate) u64);
21
22/// Per-query subscription state, keyed by [`RegId`] in `Inner`.
23pub(crate) struct QueryEntry {
24    /// The query's current change-sink `NodeId` in the engine — updated in place when a
25    /// shed rebuild re-registers the query on a fresh engine.
26    pub sink: NodeId,
27    /// The caller's opaque tag, carried so a shed rebuild re-registers under it.
28    pub query_id: QueryId,
29    /// The registered AST, retained so a shed rebuild can re-register + re-hydrate the
30    /// query without the caller's involvement.
31    pub ast: Ast,
32    /// The hydration change set (all `Add`s) captured at registration — or at the most
33    /// recent shed re-hydration. Ordinary writes do not update this cache.
34    pub hydrated: Vec<ChangeEvent>,
35    /// The committed tx the hydration reflects.
36    pub hydrated_tx: u64,
37    /// Subscriber callbacks. Fired with `Hydrated` on subscribe, then `Changed` per
38    /// committed write that touches this query — and with a fresh `Hydrated` again
39    /// after a delta-overflow shed (see [`Update`]).
40    pub callbacks: Callbacks,
41}
42
43/// A registered live query. Subscribe to receive its raw change events.
44///
45/// Tear it down explicitly with [`Query::destroy`]. Dropping the handle is still a
46/// no-op (the query keeps running and any callbacks keep firing), so an accidental
47/// drop never silently stops delivery — teardown is always a deliberate call.
48pub struct Query {
49    pub(crate) inner: Rc<Inner>,
50    pub(crate) reg: RegId,
51    pub(crate) query_id: QueryId,
52}
53
54impl Query {
55    /// The caller-supplied [`QueryId`] tag this query was registered under (echoed
56    /// verbatim; the engine never interprets it).
57    pub fn id(&self) -> QueryId {
58        self.query_id
59    }
60
61    /// Tear this query down: stop delivering its events and reclaim its pipeline.
62    ///
63    /// Removes its subscription registry entry (so no further `Changed` callbacks fire
64    /// and the hydration buffer is dropped), then disconnects it from the shared
65    /// sources and frees its operator/storage slots in the engine graph
66    /// ([`rindle::graph::Graph::destroy_pipeline`]). The shared sources — and every *other*
67    /// query — are untouched. Consumes the handle; the freed slots are recycled by a
68    /// later [`crate::Db::query`].
69    ///
70    /// Like registration, this borrows the engine mutably, so it must not be called
71    /// re-entrantly from inside a subscriber callback.
72    pub fn destroy(self) {
73        // Drop the subscription first (no callback can fire for a half-torn query),
74        // then tear the pipeline out of the shared graph.
75        let entry = self.inner.subs.borrow_mut().remove(&self.reg);
76        if let Some(entry) = entry {
77            self.inner.engine.borrow_mut().deregister_query(entry.sink);
78        }
79    }
80    /// Register a callback and immediately call it with the cached `Update::Hydrated`.
81    /// This baseline is from registration or the latest shed recovery; ordinary writes
82    /// do not update it. Subscribe before later writes. A late consumer that needs the
83    /// current result should create a new query registration.
84    ///
85    /// Later `Changed` callbacks run synchronously on the writer thread after commit.
86    /// A callback panic cannot roll back that SQL commit. Keep callbacks short, avoid
87    /// re-entering the `Db`, and forward events to a channel for other threads.
88    /// A shed recovery delivers a replacing `Hydrated`; see [`Update`]. Call
89    /// [`Query::destroy`] to stop the registration and all its callbacks.
90    ///
91    /// A no-op (the callback is dropped without firing) in one edge: the query was torn
92    /// down because its post-shed re-hydration failed (see [`crate::CommitInfo::shed`]).
93    pub fn subscribe(&self, mut cb: impl FnMut(&Update) + 'static) {
94        // Build the Hydrated payload without holding the `subs` borrow across the
95        // user callback. Callback execution must still avoid re-entering this Db.
96        let hydrated = {
97            let subs = self.inner.subs.borrow();
98            let Some(entry) = subs.get(&self.reg) else {
99                return;
100            };
101            Update::Hydrated {
102                tx_id: TxId(entry.hydrated_tx),
103                changes: entry.hydrated.clone(),
104            }
105        };
106        cb(&hydrated);
107        self.inner
108            .subs
109            .borrow_mut()
110            .get_mut(&self.reg)
111            .expect("query entry exists")
112            .callbacks
113            .push(Box::new(cb));
114    }
115}
rindle_replica/query.rs

rindle_replica/
query.rs
