coven_database/
live_query.rs

1//! Request revisions and cancellation-safe, demand-driven live results.
2
3#[cfg(any(test, feature = "test-utils"))]
4use crate::live_query_runs::LiveQueryRuns;
5use crate::observation::CommitSubscription;
6use crate::DbError;
7use coven_foundation::locks::lock_unpoisoned;
8use coven_foundation::task::{BugSinkRef, CarriesBug};
9use std::{
10    future::Future,
11    pin::Pin,
12    sync::{Arc, Mutex},
13};
14use tokio::sync::Notify;
15
16type Run<T, E> = Pin<Box<dyn Future<Output = Result<T, E>> + Send>>;
17type Query<Q, T, E> = Arc<dyn Fn(Q) -> Run<T, E> + Send + Sync>;
18
19/// A request revision assigned by one live query.
20#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
21pub struct LiveQueryRevision(pub u64);
22
23/// The query was dropped before its request could be replaced.
24#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
25#[error("live query is closed")]
26pub struct LiveQueryClosed;
27
28/// Why this run was requested; the first run answers the initial request.
29#[derive(Clone, Copy, Debug, PartialEq, Eq)]
30pub enum LiveQueryCause {
31    /// A new request, including the initial one.
32    Request,
33    /// A commit changed a column the query read, in a row its conditions
34    /// allow, or a view it read changed with no change to its tables.
35    Write,
36    /// A new request and a relevant commit arrived before this run.
37    RequestAndWrite,
38}
39
40/// One result with the exact request and revision that produced it.
41#[derive(Debug)]
42pub struct ReconfigurableLiveQueryEvent<Q, T, E> {
43    /// The request used by this run.
44    pub request: Q,
45    /// The revision assigned to that request.
46    pub revision: LiveQueryRevision,
47    /// The reason the query ran.
48    pub cause: LiveQueryCause,
49    /// A value or a failure, of the query's own error type; failures do
50    /// not end the query.
51    pub result: Result<T, E>,
52}
53
54struct Requests<Q> {
55    open: bool,
56    latest: Q,
57    revision: LiveQueryRevision,
58    pending: Option<(Q, LiveQueryRevision)>,
59}
60
61struct RequestState<Q> {
62    requests: Mutex<Requests<Q>>,
63    changed: Notify,
64}
65
66/// Shared access to replacing a live query's request.
67pub struct LiveQueryRequests<Q> {
68    state: Arc<RequestState<Q>>,
69}
70
71impl<Q> Clone for LiveQueryRequests<Q> {
72    fn clone(&self) -> Self {
73        Self { state: Arc::clone(&self.state) }
74    }
75}
76
77impl<Q: Clone + PartialEq> LiveQueryRequests<Q> {
78    /// Replace the request and return the revision whose result will answer it.
79    /// Equal consecutive requests share their revision. A replacement before
80    /// a run supersedes the pending revision; that revision is never answered.
81    pub fn set(&self, request: Q) -> Result<LiveQueryRevision, LiveQueryClosed> {
82        let mut state = lock_unpoisoned(&self.state.requests);
83        if !state.open {
84            return Err(LiveQueryClosed);
85        }
86        if request == state.latest {
87            return Ok(state.revision);
88        }
89        let revision =
90            LiveQueryRevision(state.revision.0.checked_add(1).expect("request revision exhausted"));
91        state.latest = request.clone();
92        state.revision = revision;
93        state.pending = Some((request, revision));
94        self.state.changed.notify_one();
95        Ok(revision)
96    }
97}
98
99// The subscription owns this guard; request handles do not prolong its life.
100struct QueryLifetime<Q>(Arc<RequestState<Q>>);
101
102impl<Q> Drop for QueryLifetime<Q> {
103    fn drop(&mut self) {
104        let mut state = lock_unpoisoned(&self.0.requests);
105        state.open = false;
106        state.pending = None;
107    }
108}
109
110struct Running<Q, T, E> {
111    request: Q,
112    revision: LiveQueryRevision,
113    cause: LiveQueryCause,
114    run: Run<T, E>,
115}
116
117/// A subscription whose request may be replaced while it runs. Its results
118/// fail with `E`, the query closure's own error type.
119pub struct ReconfigurableLiveQuery<Q, T, E> {
120    lifetime: QueryLifetime<Q>,
121    query: Query<Q, T, E>,
122    commits: CommitSubscription,
123    /// Where a bug in a run nobody waits for any more goes (E15).
124    bugs: BugSinkRef,
125    current: Option<(Q, LiveQueryRevision)>,
126    running: Option<Running<Q, T, E>>,
127    previous: Option<T>,
128    /// The store closed, its handle failed or a run's error carried a bug,
129    /// as a panic's does, and a result said so: the query has ended, and
130    /// answers nothing more.
131    ended: bool,
132    #[cfg(any(test, feature = "test-utils"))]
133    runs: LiveQueryRuns,
134}
135
136impl<Q, T, E> ReconfigurableLiveQuery<Q, T, E>
137where
138    Q: Clone + PartialEq + Send + Sync + 'static,
139    T: Send + 'static,
140    E: From<DbError> + CarriesBug + Send + 'static,
141{
142    pub(crate) fn new(
143        request: Q,
144        query: Query<Q, T, E>,
145        commits: CommitSubscription,
146        bugs: BugSinkRef,
147    ) -> Self {
148        Self {
149            lifetime: QueryLifetime(Arc::new(RequestState {
150                requests: Mutex::new(Requests {
151                    open: true,
152                    latest: request.clone(),
153                    revision: LiveQueryRevision(0),
154                    pending: Some((request, LiveQueryRevision(0))),
155                }),
156                changed: Notify::new(),
157            })),
158            query,
159            commits,
160            bugs,
161            current: None,
162            running: None,
163            previous: None,
164            ended: false,
165            #[cfg(any(test, feature = "test-utils"))]
166            runs: LiveQueryRuns::default(),
167        }
168    }
169
170    /// How often this query has run so far, its processed form's runs too.
171    #[cfg(any(test, feature = "test-utils"))]
172    pub fn runs(&self) -> LiveQueryRuns {
173        self.runs
174    }
175
176    /// Whether a commit since this idle query's last result changes what it
177    /// reads, so asking for its next result runs it again. A commit marks the
178    /// queries it changes as it commits, so this answers with no waiting.
179    #[cfg(any(test, feature = "test-utils"))]
180    pub fn is_marked_for_rerun(&self) -> bool {
181        assert!(
182            self.current.is_some() && self.running.is_none(),
183            "probe requires a completed, idle query"
184        );
185        self.commits.state().changed
186    }
187
188    /// Obtain a cloneable handle for replacing this query's request.
189    pub fn requests(&self) -> LiveQueryRequests<Q> {
190        LiveQueryRequests { state: Arc::clone(&self.lifetime.0) }
191    }
192
193    /// Process each value on a worker after releasing the connection.
194    /// Equality compares processed values; request handles and event metadata remain.
195    pub fn process<P, U>(self, process: P) -> ReconfigurableLiveQuery<Q, U, E>
196    where
197        T: Clone + PartialEq,
198        P: Fn(T) -> Result<U, E> + Send + Sync + 'static,
199        U: Send + 'static,
200    {
201        let process = Arc::new(process);
202        let ReconfigurableLiveQuery {
203            lifetime,
204            query,
205            commits,
206            bugs,
207            current,
208            running,
209            previous: _,
210            ended,
211            #[cfg(any(test, feature = "test-utils"))]
212            runs,
213        } = self;
214        let next_process = Arc::clone(&process);
215        let next_bugs = bugs.clone();
216        let mapped: Query<Q, U, E> = Arc::new(move |request| {
217            process_run(query(request), Arc::clone(&next_process), next_bugs.clone())
218        });
219        let running = running.map(|Running { request, revision, cause, run }| Running {
220            request,
221            revision,
222            cause,
223            run: process_run(run, Arc::clone(&process), bugs.clone()),
224        });
225        // A new output type has no previous result to compare. Rerun the current
226        // request if it has already completed so the transformed query answers it.
227        if running.is_none() {
228            if let Some(request) = &current {
229                let mut requests = lock_unpoisoned(&lifetime.0.requests);
230                if requests.pending.is_none() {
231                    requests.pending = Some(request.clone());
232                }
233            }
234        }
235        ReconfigurableLiveQuery {
236            lifetime,
237            query: mapped,
238            commits,
239            bugs,
240            current,
241            running,
242            previous: None,
243            ended,
244            #[cfg(any(test, feature = "test-utils"))]
245            runs,
246        }
247    }
248
249    /// Return the next result and its request. Cancelling this wait retains the
250    /// in-flight run, so its request revision is not lost. Once the store is
251    /// closed, or the handle fails every call, one result reports why and
252    /// every later call returns `None`. So does a run whose error carries a
253    /// bug, as one that panicked does with `DbError::Bug`: what it read is
254    /// unknown, so no commit could tell whether to run it again (E15).
255    pub async fn next(&mut self) -> Option<ReconfigurableLiveQueryEvent<Q, T, E>>
256    where
257        T: Clone + PartialEq,
258    {
259        loop {
260            if self.ended {
261                return None;
262            }
263            if self.running.is_none() {
264                let commits = self.commits.state();
265                let failure = commits.error();
266                let changed = commits.changed;
267                let request = lock_unpoisoned(&self.lifetime.0.requests).pending.take();
268                let next = match (request, changed) {
269                    (Some(request), true) => Some((request, LiveQueryCause::RequestAndWrite)),
270                    (Some(request), false) => Some((request, LiveQueryCause::Request)),
271                    (None, true) => Some((
272                        self.current.as_ref().expect("a previous run read columns").clone(),
273                        LiveQueryCause::Write,
274                    )),
275                    (None, false) if failure.is_some() => Some((
276                        self.current.as_ref().expect("initial request was answered").clone(),
277                        LiveQueryCause::Write,
278                    )),
279                    (None, false) => None,
280                };
281                if let Some(((request, revision), cause)) = next {
282                    let run: Run<T, E> = if let Some(error) = failure {
283                        Box::pin(async move { Err(error.into()) })
284                    } else {
285                        self.commits.begin();
286                        caught_run((self.query)(request.clone()))
287                    };
288                    self.running = Some(Running { request, revision, cause, run });
289                } else {
290                    tokio::select! {
291                        _ = self.lifetime.0.changed.notified() => {},
292                        _ = self.commits.changed() => {},
293                    }
294                    continue;
295                }
296            }
297            let running = self.running.as_mut().expect("run selected");
298            let result = running.run.as_mut().await;
299            let running = self.running.take().expect("finished run");
300            self.current = Some((running.request.clone(), running.revision));
301            let equal = match &result {
302                Ok(value) => self.previous.as_ref() == Some(value),
303                Err(_) => false,
304            };
305            // An error keeps its typed sources, which need have no value
306            // equality or Clone. A failed run is delivered and breaks the chain
307            // of equal successful results; it never terminates the subscription.
308            self.previous = match &result {
309                Ok(value) => Some(value.clone()),
310                Err(_) => None,
311            };
312            #[cfg(any(test, feature = "test-utils"))]
313            self.runs.count(running.cause, equal);
314            if running.cause == LiveQueryCause::Write && equal {
315                continue;
316            }
317            // A failure while the store is closed or the handle failed is the
318            // last result: answering again would only repeat it. So is a bug,
319            // after which the query watches nothing more.
320            let bug = result.as_ref().err().and_then(CarriesBug::bug).is_some();
321            if bug {
322                self.commits.stop();
323            }
324            self.ended = bug || (result.is_err() && self.commits.state().error().is_some());
325            return Some(ReconfigurableLiveQueryEvent {
326                request: running.request,
327                revision: running.revision,
328                cause: running.cause,
329                result,
330            });
331        }
332    }
333}
334
335/// `run`, which ends with its `DbError::Bug` if it panics, so `next` never
336/// polls it again after the panic.
337fn caught_run<T: Send + 'static, E: From<DbError> + Send + 'static>(run: Run<T, E>) -> Run<T, E> {
338    Box::pin(async move {
339        coven_foundation::task::caught("live query", run)
340            .await
341            .unwrap_or_else(|bug| Err(E::from(DbError::Bug(bug))))
342    })
343}
344
345/// A subscription's results and lifetime; dropping it ends future runs. Its
346/// results fail with `E`, the query closure's own error type.
347pub struct LiveQuery<T, E> {
348    inner: ReconfigurableLiveQuery<(), T, E>,
349}
350
351impl<T: Send + 'static, E: From<DbError> + CarriesBug + Send + 'static> LiveQuery<T, E> {
352    pub(crate) fn new(inner: ReconfigurableLiveQuery<(), T, E>) -> Self {
353        Self { inner }
354    }
355
356    /// The first result at once; then only changed results. Errors remain
357    /// results and don't end the query, except the one reporting that the
358    /// store closed, the handle failed or a run's error carries a bug, as a
359    /// panic's does: after it, `None`.
360    pub async fn next(&mut self) -> Option<Result<T, E>>
361    where
362        T: Clone + PartialEq,
363    {
364        Some(self.inner.next().await?.result)
365    }
366
367    /// How often this query has run so far.
368    #[cfg(any(test, feature = "test-utils"))]
369    pub fn runs(&self) -> LiveQueryRuns {
370        self.inner.runs()
371    }
372
373    /// Whether a commit since this idle query's last result changes what it
374    /// reads, so asking for its next result runs it again, with no waiting.
375    #[cfg(any(test, feature = "test-utils"))]
376    pub fn is_marked_for_rerun(&self) -> bool {
377        self.inner.is_marked_for_rerun()
378    }
379
380    /// Process each value on a worker after its connection has been released.
381    /// Equality is tested on the processed value.
382    pub fn process<P, U>(self, process: P) -> LiveQuery<U, E>
383    where
384        T: Clone + PartialEq,
385        P: Fn(T) -> Result<U, E> + Send + Sync + 'static,
386        U: Send + 'static,
387    {
388        LiveQuery { inner: self.inner.process(process) }
389    }
390}
391
392fn process_run<T: Send + 'static, U: Send + 'static, E, P>(
393    run: Run<T, E>,
394    process: Arc<P>,
395    bugs: BugSinkRef,
396) -> Run<U, E>
397where
398    E: From<DbError> + CarriesBug + Send + 'static,
399    P: Fn(T) -> Result<U, E> + Send + Sync + 'static,
400{
401    Box::pin(async move {
402        let result = run.await;
403        match result {
404            Ok(value) => crate::database::process(&bugs, move || process(value)).await,
405            Err(error) => Err(error),
406        }
407    })
408}
409
410#[cfg(test)]
411#[path = "live_query_tests.rs"]
412mod tests;
coven_database/live_query.rs

coven_database/
live_query.rs
