cartel-sqlite 0.31.0

SQLite driver for the dope runtime
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

use crate::{api, catalogs, config, sys};
use api::{exec, stmt, work};
use dope_fiber::abi;
use o3::mem::{self, lease};
use release::Release;
use std::{cell, marker, result};
use work::guard::{self, call};

/// A savepoint capability whose queries cannot outlive the savepoint.
pub struct Savepoint<
    'transaction,
    'connection: 'transaction,
    'd: 'connection,
    I: catalogs::Catalog = catalogs::Empty,
> {
    scope: mem::Lease<Live<'transaction, 'connection, 'd, I>, lease::Open>,
}

/// Owns one successful savepoint's return to its original work state.
struct Live<'scope, 'connection: 'scope, 'd: 'connection, I: catalogs::Catalog> {
    source: &'scope work::Work,
    depth: work::Count,
    failed: &'scope cell::Cell<Option<sys::Fault>>,
    state: marker::PhantomData<&'scope mut api::State<'connection, 'd, I>>,
}

impl<'transaction, 'connection: 'transaction, 'd: 'connection, I: catalogs::Catalog>
    Savepoint<'transaction, 'connection, 'd, I>
{
    pub(super) fn open(
        state: &'transaction mut api::State<'connection, 'd, I>,
        failed: &'transaction cell::Cell<Option<sys::Fault>>,
    ) -> impl abi::Fiber<'d, Output = result::Result<Self, crate::Error>>
    + use<'transaction, 'connection, 'd, I> {
        dope_fiber::fiber!('d => async move {
            let txn = exec::Tx::new(failed);
            txn.check()?;
            let depth = state.control(Open, txn).await?;
            let live = Live {
                source: state.work(),
                depth,
                failed,
                state: marker::PhantomData,
            };
            Ok(Self {
                scope: mem::Lease::open(live, state),
            })
        })
    }

    pub fn query(&mut self) -> exec::Query<'_, 'connection, 'd, I> {
        let (live, state) = self.scope.parts_mut();
        state.query_in(exec::Tx::new(live.failed))
    }

    pub fn savepoint<'nested>(
        &'nested mut self,
    ) -> impl abi::Fiber<
        'd,
        Output = result::Result<Savepoint<'nested, 'connection, 'd, I>, crate::Error>,
    > + use<'nested, 'transaction, 'connection, 'd, I> {
        let (live, state) = self.scope.parts_mut();
        Savepoint::open(state, live.failed)
    }

    pub fn commit(
        mut self,
    ) -> impl abi::Fiber<'d, Output = result::Result<(), crate::Error>>
    + use<'transaction, 'connection, 'd, I> {
        dope_fiber::fiber!('d => async move {
            loop {
                let (live, state) = self.scope.parts_mut();
                let txn = exec::Tx::new(live.failed);
                if state.control(Release::new(live, false), txn).await? {
                    break;
                }
            }
            let _ = self.scope.into_parts();
            Ok(())
        })
    }

    pub fn rollback<'rollback>(
        &'rollback mut self,
    ) -> impl abi::Fiber<'d, Output = result::Result<(), crate::Error>>
    + use<'rollback, 'transaction, 'connection, 'd, I> {
        dope_fiber::fiber!('d => async move {
            loop {
                let (live, state) = self.scope.parts_mut();
                let txn = exec::Tx::new(live.failed);
                if state.control(Release::new(live, true), txn).await? {
                    break;
                }
            }
            Ok(())
        })
    }
}

impl<'scope, 'connection: 'scope, 'd: 'connection, I: catalogs::Catalog> mem::Return
    for Live<'scope, 'connection, 'd, I>
{
    type Item = &'scope mut api::State<'connection, 'd, I>;

    fn return_item(self, _: Self::Item) {
        if self.failed.get().is_none() {
            self.source.revert(self.depth);
        }
    }
}

const _: () = {
    use std::mem;

    assert!(mem::size_of::<Savepoint<'_, '_, '_>>() == 4 * mem::size_of::<usize>());
};

struct Open;

impl<I: catalogs::Catalog> stmt::target::Target<I> for Open {
    type Access = work::Permit;

    fn bind<'query, 'connection, P: api::Bind>(
        self,
        engine: &'query mut stmt::Engine<'connection, I>,
        params: P,
        config: &config::Config,
        guard: &guard::Guard<'_, call::Call>,
    ) -> result::Result<(stmt::Bound<'query, 'connection, P>, Self::Access), crate::Error> {
        let permit = guard.work().depth.check(config.work.savepoints)?;
        Ok((
            engine
                .controls()
                .bind(stmt::Control::Savepoint, params, config)?,
            permit,
        ))
    }
}

mod release;