rullst_orm/
post_commit.rs

1//! Transaction-aware post-commit callbacks.
2
3use std::future::Future;
4use std::pin::Pin;
5use std::sync::Arc;
6
7use tokio::sync::Mutex;
8
9use crate::Error;
10
11type CallbackFuture = Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'static>>;
12type Callback = Box<dyn FnOnce() -> CallbackFuture + Send + 'static>;
13type CallbackQueue = Arc<Mutex<Vec<Callback>>>;
14
15tokio::task_local! {
16    static CALLBACKS: CallbackQueue;
17}
18
19/// Database mutation represented by a committed model event.
20#[derive(Clone, Copy, Debug, Eq, PartialEq)]
21pub enum ModelOperation {
22    /// A new model row was created.
23    Created,
24    /// An existing model row was updated.
25    Updated,
26    /// A model row was deleted or soft-deleted.
27    Deleted,
28}
29
30impl ModelOperation {
31    /// Stable lowercase name used by event transports.
32    pub const fn as_str(self) -> &'static str {
33        match self {
34            Self::Created => "created",
35            Self::Updated => "updated",
36            Self::Deleted => "deleted",
37        }
38    }
39}
40
41/// Owned model snapshot delivered only after a managed transaction commits.
42#[derive(Clone, Debug, Eq, PartialEq)]
43pub struct ModelCommittedEvent {
44    /// Persisted table name generated from the model metadata.
45    pub table: &'static str,
46    /// Model primary key.
47    pub id: i32,
48    /// Mutation that was committed.
49    pub operation: ModelOperation,
50    /// Model JSON generated with the model's hidden-field policy.
51    pub payload: String,
52}
53
54impl ModelCommittedEvent {
55    /// Creates an owned committed-event snapshot.
56    pub fn new(
57        table: &'static str,
58        id: i32,
59        operation: ModelOperation,
60        payload: impl Into<String>,
61    ) -> Self {
62        Self {
63            table,
64            id,
65            operation,
66            payload: payload.into(),
67        }
68    }
69}
70
71/// Owns callbacks collected by one managed database transaction.
72///
73/// This type is public only so generated ORM code can share the same commit
74/// boundary. Applications should normally use [`crate::Orm::transaction`] and
75/// [`after_commit`]. Dropping a scope discards every pending callback.
76#[doc(hidden)]
77pub struct PostCommitScope {
78    callbacks: CallbackQueue,
79}
80
81impl PostCommitScope {
82    /// Creates an empty callback scope.
83    #[doc(hidden)]
84    pub fn new() -> Self {
85        Self {
86            callbacks: Arc::new(Mutex::new(Vec::new())),
87        }
88    }
89
90    /// Runs a future with this scope as the current commit boundary.
91    #[doc(hidden)]
92    pub async fn run<F>(&self, future: F) -> F::Output
93    where
94        F: Future,
95    {
96        CALLBACKS.scope(self.callbacks.clone(), future).await
97    }
98
99    /// Like [`Self::run`], but the returned future owns its callback queue
100    /// instead of borrowing `self`. Generic callers need this: a generic
101    /// future holding a borrowed scope is rejected by rustc as not `Send`.
102    pub(crate) fn scope<F>(
103        &self,
104        future: F,
105    ) -> tokio::task::futures::TaskLocalFuture<CallbackQueue, F>
106    where
107        F: Future,
108    {
109        CALLBACKS.scope(self.callbacks.clone(), future)
110    }
111
112    /// Runs all callbacks after the caller has confirmed its database commit.
113    #[doc(hidden)]
114    pub async fn commit(self) -> Result<(), Error> {
115        let callbacks = {
116            let mut callbacks = self.callbacks.lock().await;
117            std::mem::take(&mut *callbacks)
118        };
119        run_callbacks(callbacks).await
120    }
121
122    /// Promotes callbacks into an enclosing commit boundary after a nested
123    /// savepoint succeeds. Without an enclosing managed transaction, the
124    /// callbacks run immediately, matching the raw-transaction limitation.
125    #[doc(hidden)]
126    pub async fn promote_to_parent(self) -> Result<(), Error> {
127        let parent = CALLBACKS.try_with(Clone::clone).ok();
128        if parent
129            .as_ref()
130            .is_some_and(|parent| Arc::ptr_eq(parent, &self.callbacks))
131        {
132            return Err(Error::Internal(
133                "post-commit scope cannot promote into itself".to_string(),
134            ));
135        }
136        let callbacks = {
137            let mut callbacks = self.callbacks.lock().await;
138            std::mem::take(&mut *callbacks)
139        };
140        if let Some(parent) = parent {
141            parent.lock().await.extend(callbacks);
142            Ok(())
143        } else {
144            run_callbacks(callbacks).await
145        }
146    }
147}
148
149impl Default for PostCommitScope {
150    fn default() -> Self {
151        Self::new()
152    }
153}
154
155/// Runs a callback after the current [`crate::Orm::transaction`] commits.
156///
157/// If no managed transaction is active, the callback runs immediately. This
158/// matches an already committed/autocommit operation. A callback failure is
159/// returned as [`Error::PostCommit`], making it explicit that the database
160/// write may already be durable.
161pub async fn after_commit<F, Fut>(callback: F) -> Result<(), Error>
162where
163    F: FnOnce() -> Fut + Send + 'static,
164    Fut: Future<Output = Result<(), Error>> + Send + 'static,
165{
166    if let Ok(queue) = CALLBACKS.try_with(Clone::clone) {
167        queue
168            .lock()
169            .await
170            .push(Box::new(move || Box::pin(callback())));
171        return Ok(());
172    }
173
174    callback()
175        .await
176        .map_err(|error| Error::PostCommit(error.to_string()))
177}
178
179async fn run_callbacks(callbacks: Vec<Callback>) -> Result<(), Error> {
180    let mut failures = Vec::new();
181    for callback in callbacks {
182        if let Err(error) = callback().await {
183            failures.push(error.to_string());
184        }
185    }
186
187    if failures.is_empty() {
188        Ok(())
189    } else {
190        Err(Error::PostCommit(failures.join("; ")))
191    }
192}
193
194#[cfg(test)]
195mod tests {
196    use std::sync::Arc;
197    use std::sync::atomic::{AtomicUsize, Ordering};
198
199    use super::{PostCommitScope, after_commit};
200    use crate::Error;
201
202    #[tokio::test]
203    async fn callback_without_scope_runs_immediately() {
204        let calls = Arc::new(AtomicUsize::new(0));
205        let callback_calls = calls.clone();
206        after_commit(move || async move {
207            callback_calls.fetch_add(1, Ordering::SeqCst);
208            Ok(())
209        })
210        .await
211        .expect("run immediate callback");
212
213        assert_eq!(calls.load(Ordering::SeqCst), 1);
214    }
215
216    #[tokio::test]
217    async fn scope_defers_callbacks_and_drop_discards_them() {
218        let calls = Arc::new(AtomicUsize::new(0));
219        let scope = PostCommitScope::new();
220        let callback_calls = calls.clone();
221        scope
222            .run(after_commit(move || async move {
223                callback_calls.fetch_add(1, Ordering::SeqCst);
224                Ok(())
225            }))
226            .await
227            .expect("register callback");
228        assert_eq!(calls.load(Ordering::SeqCst), 0);
229
230        drop(scope);
231        assert_eq!(calls.load(Ordering::SeqCst), 0);
232    }
233
234    #[tokio::test]
235    async fn commit_runs_every_callback_and_reports_failures() {
236        let calls = Arc::new(AtomicUsize::new(0));
237        let scope = PostCommitScope::new();
238        scope
239            .run(async {
240                after_commit(|| async { Err(Error::CacheError("cache unavailable".to_string())) })
241                    .await?;
242                let callback_calls = calls.clone();
243                after_commit(move || async move {
244                    callback_calls.fetch_add(1, Ordering::SeqCst);
245                    Ok(())
246                })
247                .await
248            })
249            .await
250            .expect("register callbacks");
251
252        let error = scope
253            .commit()
254            .await
255            .expect_err("failed callback must be reported");
256        assert!(matches!(error, Error::PostCommit(_)));
257        assert_eq!(calls.load(Ordering::SeqCst), 1);
258    }
259
260    #[tokio::test]
261    async fn nested_scope_promotes_only_after_its_operation_succeeds() {
262        let calls = Arc::new(AtomicUsize::new(0));
263        let outer = PostCommitScope::new();
264        outer
265            .run(async {
266                let inner = PostCommitScope::new();
267                let callback_calls = calls.clone();
268                inner
269                    .run(after_commit(move || async move {
270                        callback_calls.fetch_add(1, Ordering::SeqCst);
271                        Ok(())
272                    }))
273                    .await?;
274                inner.promote_to_parent().await
275            })
276            .await
277            .expect("promote nested callbacks");
278        assert_eq!(calls.load(Ordering::SeqCst), 0);
279        outer.commit().await.expect("commit outer callbacks");
280        assert_eq!(calls.load(Ordering::SeqCst), 1);
281    }
282}
rullst_orm/post_commit.rs

rullst_orm/
post_commit.rs
