coven_database/
commit_scope.rs

1//! Commits attributed to the work that made them, so an observation can leave
2//! one piece of work's own commits out: an operation attempt's own commits are
3//! never evidence against the wait it ends in (§6.7, §21.2; IO
4//! `WaitingEvidence`).
5
6use std::cell::Cell;
7use std::future::Future;
8use std::sync::atomic::{AtomicU64, Ordering};
9
10/// Commits made by database calls that a [`CommitScope::run`] future starts
11/// belong to that scope. Calls started from other tasks, including tasks the
12/// future spawns, are outside it.
13#[derive(Clone, Copy, Debug, PartialEq, Eq)]
14pub struct CommitScope(u64);
15
16tokio::task_local! {
17    static TASK: CommitScope;
18}
19
20thread_local! {
21    // Clippy asks for a `const` initializer this already has where the
22    // target has no native thread locals, as Android: std builds those
23    // from a function, which the lint reads as the initializer.
24    #[cfg_attr(target_os = "android", allow(clippy::missing_const_for_thread_local))]
25    static THREAD: Cell<Option<CommitScope>> = const { Cell::new(None) };
26}
27
28impl CommitScope {
29    /// A scope no other scope equals.
30    pub fn new() -> Self {
31        static NEXT: AtomicU64 = AtomicU64::new(1);
32        Self(NEXT.fetch_add(1, Ordering::Relaxed))
33    }
34
35    /// Run `work` with its database calls' commits attributed to this scope.
36    /// The work is held once, inside the scope that polls it; an `async fn`
37    /// here would hold a second, moved copy beside it.
38    pub fn run<F: Future>(self, work: F) -> impl Future<Output = F::Output> {
39        TASK.scope(self, work)
40    }
41}
42
43impl Default for CommitScope {
44    fn default() -> Self {
45        Self::new()
46    }
47}
48
49/// Bind a database call to the scope of the task starting it, before it runs
50/// on a blocking thread: commits it makes report that scope.
51pub(crate) fn bind<R>(
52    run: impl FnOnce() -> R + Send + 'static,
53) -> impl FnOnce() -> R + Send + 'static {
54    let scope = TASK.try_with(|scope| *scope).ok();
55    #[cfg(any(test, feature = "test-utils"))]
56    let run = coven_foundation::test_work::bind(run);
57    move || enter(scope, run)
58}
59
60fn enter<R>(scope: Option<CommitScope>, run: impl FnOnce() -> R) -> R {
61    struct Restore(Option<CommitScope>);
62    impl Drop for Restore {
63        fn drop(&mut self) {
64            THREAD.with(|thread| thread.set(self.0));
65        }
66    }
67    let _restore = Restore(THREAD.with(|thread| thread.replace(scope)));
68    run()
69}
70
71/// The scope of the call committing on this thread.
72pub(crate) fn committing() -> Option<CommitScope> {
73    THREAD.with(Cell::get)
74}
coven_database/commit_scope.rs

coven_database/
commit_scope.rs
