cartel_sqlite/engine/api/tx/
mod.rs

1mod save;
2
3pub use save::Savepoint;
4
5use crate::{api, catalogs, sys};
6use api::{exec, stmt};
7use dope_fiber::abi;
8use o3::mem::{self, lease};
9use std::{cell, marker, result};
10
11/// A transaction capability whose queries cannot outlive it.
12/// The transaction cannot expose its underlying engine capability:
13///
14pub struct Transaction<
15    'db,
16    'connection: 'db,
17    'd: 'connection,
18    I: catalogs::Catalog = catalogs::Empty,
19> {
20    scope: mem::Lease<Live<'db, 'connection, 'd, I>, lease::Open>,
21}
22
23struct Live<'db, 'connection: 'db, 'd: 'connection, I: catalogs::Catalog> {
24    failed: cell::Cell<Option<sys::Fault>>,
25    state: marker::PhantomData<&'db mut api::State<'connection, 'd, I>>,
26}
27
28impl<'db, 'connection: 'db, 'd: 'connection, I: catalogs::Catalog>
29    Transaction<'db, 'connection, 'd, I>
30{
31    pub(crate) const fn new(state: &'db mut api::State<'connection, 'd, I>) -> Self {
32        Self {
33            scope: mem::Lease::open(
34                Live {
35                    failed: cell::Cell::new(None),
36                    state: marker::PhantomData,
37                },
38                state,
39            ),
40        }
41    }
42
43    pub fn query(&mut self) -> exec::Query<'_, 'connection, 'd, I> {
44        let (live, state) = self.scope.parts_mut();
45        state.query_in(exec::Tx::new(&live.failed))
46    }
47
48    pub fn savepoint<'transaction>(
49        &'transaction mut self,
50    ) -> impl abi::Fiber<
51        'd,
52        Output = result::Result<Savepoint<'transaction, 'connection, 'd, I>, crate::Error>,
53    > + use<'transaction, 'db, 'connection, 'd, I> {
54        let (live, state) = self.scope.parts_mut();
55        Savepoint::open(state, &live.failed)
56    }
57
58    pub fn commit(
59        mut self,
60    ) -> impl abi::Fiber<'d, Output = result::Result<(), crate::Error>> + use<'db, 'connection, 'd, I>
61    {
62        dope_fiber::fiber!('d => async move {
63            let result = {
64                let (live, state) = self.scope.parts_mut();
65                let txn = exec::Tx::new(&live.failed);
66                txn.check()?;
67                state.control(stmt::Control::Commit, txn).await
68            };
69            if result.is_ok() || matches!(self.scope.item().tx(), sys::Tx::Idle) {
70                self.scope.item().work().depth.reset();
71                let _ = self.scope.into_parts();
72                return result;
73            }
74            result
75        })
76    }
77
78    pub fn rollback(
79        mut self,
80    ) -> impl abi::Fiber<'d, Output = result::Result<(), crate::Error>> + use<'db, 'connection, 'd, I>
81    {
82        dope_fiber::fiber!('d => async move {
83            let result = {
84                let (live, state) = self.scope.parts_mut();
85                exec::Tx::new(&live.failed).rollback()?;
86                state.control(stmt::Control::Rollback, exec::Tx::NONE).await
87            };
88            if result.is_ok() || matches!(self.scope.item().tx(), sys::Tx::Idle) {
89                self.scope.item().work().depth.reset();
90                let _ = self.scope.into_parts();
91                return result;
92            }
93            result
94        })
95    }
96}
97
98impl<'db, 'connection: 'db, 'd: 'connection, I: catalogs::Catalog> mem::Return
99    for Live<'db, 'connection, 'd, I>
100{
101    type Item = &'db mut api::State<'connection, 'd, I>;
102
103    fn return_item(self, state: Self::Item) {
104        let work = state.work();
105        if self.failed.get().is_some_and(sys::Fault::is_aborted) {
106            work.depth.reset();
107        } else {
108            work.pending.transaction();
109        }
110    }
111}
cartel_sqlite/engine/api/tx/mod.rs

cartel_sqlite/engine/api/tx/
mod.rs
