cartel_sqlite/engine/api/tx/save/
mod.rs

1use crate::{api, catalogs, config, sys};
2use api::{exec, stmt, work};
3use dope_fiber::abi;
4use o3::mem::{self, lease};
5use release::Release;
6use std::{cell, marker, result};
7use work::guard::{self, call};
8
9/// A savepoint capability whose queries cannot outlive the savepoint.
10pub struct Savepoint<
11    'transaction,
12    'connection: 'transaction,
13    'd: 'connection,
14    I: catalogs::Catalog = catalogs::Empty,
15> {
16    scope: mem::Lease<Live<'transaction, 'connection, 'd, I>, lease::Open>,
17}
18
19/// Owns one successful savepoint's return to its original work state.
20struct Live<'scope, 'connection: 'scope, 'd: 'connection, I: catalogs::Catalog> {
21    source: &'scope work::Work,
22    depth: work::Count,
23    failed: &'scope cell::Cell<Option<sys::Fault>>,
24    state: marker::PhantomData<&'scope mut api::State<'connection, 'd, I>>,
25}
26
27impl<'transaction, 'connection: 'transaction, 'd: 'connection, I: catalogs::Catalog>
28    Savepoint<'transaction, 'connection, 'd, I>
29{
30    pub(super) fn open(
31        state: &'transaction mut api::State<'connection, 'd, I>,
32        failed: &'transaction cell::Cell<Option<sys::Fault>>,
33    ) -> impl abi::Fiber<'d, Output = result::Result<Self, crate::Error>>
34    + use<'transaction, 'connection, 'd, I> {
35        dope_fiber::fiber!('d => async move {
36            let txn = exec::Tx::new(failed);
37            txn.check()?;
38            let depth = state.control(Open, txn).await?;
39            let live = Live {
40                source: state.work(),
41                depth,
42                failed,
43                state: marker::PhantomData,
44            };
45            Ok(Self {
46                scope: mem::Lease::open(live, state),
47            })
48        })
49    }
50
51    pub fn query(&mut self) -> exec::Query<'_, 'connection, 'd, I> {
52        let (live, state) = self.scope.parts_mut();
53        state.query_in(exec::Tx::new(live.failed))
54    }
55
56    pub fn savepoint<'nested>(
57        &'nested mut self,
58    ) -> impl abi::Fiber<
59        'd,
60        Output = result::Result<Savepoint<'nested, 'connection, 'd, I>, crate::Error>,
61    > + use<'nested, 'transaction, 'connection, 'd, I> {
62        let (live, state) = self.scope.parts_mut();
63        Savepoint::open(state, live.failed)
64    }
65
66    pub fn commit(
67        mut self,
68    ) -> impl abi::Fiber<'d, Output = result::Result<(), crate::Error>>
69    + use<'transaction, 'connection, 'd, I> {
70        dope_fiber::fiber!('d => async move {
71            loop {
72                let (live, state) = self.scope.parts_mut();
73                let txn = exec::Tx::new(live.failed);
74                if state.control(Release::new(live, false), txn).await? {
75                    break;
76                }
77            }
78            let _ = self.scope.into_parts();
79            Ok(())
80        })
81    }
82
83    pub fn rollback<'rollback>(
84        &'rollback mut self,
85    ) -> impl abi::Fiber<'d, Output = result::Result<(), crate::Error>>
86    + use<'rollback, 'transaction, 'connection, 'd, I> {
87        dope_fiber::fiber!('d => async move {
88            loop {
89                let (live, state) = self.scope.parts_mut();
90                let txn = exec::Tx::new(live.failed);
91                if state.control(Release::new(live, true), txn).await? {
92                    break;
93                }
94            }
95            Ok(())
96        })
97    }
98}
99
100impl<'scope, 'connection: 'scope, 'd: 'connection, I: catalogs::Catalog> mem::Return
101    for Live<'scope, 'connection, 'd, I>
102{
103    type Item = &'scope mut api::State<'connection, 'd, I>;
104
105    fn return_item(self, _: Self::Item) {
106        if self.failed.get().is_none() {
107            self.source.revert(self.depth);
108        }
109    }
110}
111
112const _: () = {
113    use std::mem;
114
115    assert!(mem::size_of::<Savepoint<'_, '_, '_>>() == 4 * mem::size_of::<usize>());
116};
117
118struct Open;
119
120impl<I: catalogs::Catalog> stmt::target::Target<I> for Open {
121    type Access = work::Permit;
122
123    fn bind<'query, 'connection, P: api::Bind>(
124        self,
125        engine: &'query mut stmt::Engine<'connection, I>,
126        params: P,
127        config: &config::Config,
128        guard: &guard::Guard<'_, call::Call>,
129    ) -> result::Result<(stmt::Bound<'query, 'connection, P>, Self::Access), crate::Error> {
130        let permit = guard.work().depth.check(config.work.savepoints)?;
131        Ok((
132            engine
133                .controls()
134                .bind(stmt::Control::Savepoint, params, config)?,
135            permit,
136        ))
137    }
138}
139
140mod release;
cartel_sqlite/engine/api/tx/save/mod.rs

cartel_sqlite/engine/api/tx/save/
mod.rs
