core_crypto_keystore/transaction/
mod.rs

1mod entity_read;
2mod entity_write;
3mod fetch_from_database;
4mod mls;
5#[cfg(feature = "proteus-keystore")]
6pub mod proteus;
7mod specializations;
8
9use std::{borrow::Cow, collections::HashSet, sync::Arc};
10
11use async_lock::{RwLock, SemaphoreGuardArc};
12use itertools::Itertools;
13use ordermap::OrderMap;
14
15use crate::{
16    CryptoKeystoreError, CryptoKeystoreResult, Database, UniqueArc,
17    traits::{Entity, KeyType},
18    transaction::dynamic_dispatch::EntityId,
19};
20
21pub(crate) mod dynamic_dispatch;
22
23/// This is an in-flight transaction: all operations are buffered in memory, and only
24/// applied to the database on [`commit`][UniqueArc<Self>::commit].
25///
26/// Dropping the transaction without committing performs an implicit rollback.
27///
28/// This type is always wrapped in a [`UniqueArc`], which keeps things efficient,
29/// at the cost of prohibiting `Clone`. In case you need to share this around,
30/// there are weak references available via [`UniqueArc::downgrade`].
31/// Alternately, wrap the entire thing in an `Arc<Mutex<Option<UniqueArc<Self>>>>` or similar.
32/// Just be aware that you'll need to take the unique arc out in order to commit.
33pub struct Transaction {
34    cache: RwLock<OrderMap<EntityId, dynamic_dispatch::Entity>>,
35    deleted: RwLock<HashSet<EntityId>>,
36    _semaphore_guard: Arc<SemaphoreGuardArc>,
37    database: Arc<Database>,
38}
39
40impl Transaction {
41    /// Instantiate a new transaction.
42    ///
43    /// Requires a semaphore guard to ensure that only one exists at a time.
44    pub(crate) async fn new(
45        semaphore_guard: SemaphoreGuardArc,
46        database: Arc<Database>,
47    ) -> CryptoKeystoreResult<UniqueArc<Self>> {
48        let transaction = UniqueArc::from(Self {
49            cache: Default::default(),
50            deleted: Default::default(),
51            _semaphore_guard: Arc::new(semaphore_guard),
52            database,
53        });
54
55        let weak = UniqueArc::downgrade(&transaction);
56
57        {
58            let mut transaction_guard = transaction.database.transaction.lock().await;
59            // this transaction guard may be `None` if the database is new or the previous transaction
60            // was committed.
61            // it may be `Some(_)` if the previous transaction was rolled back by means of dropping the
62            // `UniqueArc<Transaction>`. either way, it's correct to simply replace it without checking the
63            // previous value.
64            *transaction_guard = Some(weak);
65        }
66
67        Ok(transaction)
68    }
69
70    /// Build a single list of unique records from two potentially overlapping lists.
71    ///
72    /// In case of overlap, records in `records_a` are prioritized. Identity from the perspective
73    /// of this function is determined by the byte encoding of
74    /// [`primary_key`][crate::traits::PrimaryKey::primary_key].
75    ///
76    /// Records deleted in this transaction are filtered out of the result.
77    async fn merge_records<'a, E>(
78        &self,
79        records_a: impl IntoIterator<Item = Cow<'a, E>>,
80        records_b: impl IntoIterator<Item = Cow<'a, E>>,
81    ) -> Vec<E>
82    where
83        E: 'static + Clone + Entity,
84    {
85        let deleted_records = self.deleted.read().await;
86
87        records_a
88            .into_iter()
89            .chain(records_b)
90            .unique_by(|e| e.primary_key().bytes().into_owned())
91            .filter_map(|record| {
92                let id = EntityId::from_entity(record.as_ref())?;
93                (!deleted_records.contains(&id)).then_some(record.into_owned())
94            })
95            .collect()
96    }
97}
98
99impl UniqueArc<Transaction> {
100    /// Persists all the operations in the database. It will effectively open a transaction
101    /// internally, perform all the buffered operations and commit.
102    pub async fn commit(self) -> Result<(), CryptoKeystoreError> {
103        let Transaction {
104            cache,
105            deleted,
106            database,
107            _semaphore_guard,
108        } = UniqueArc::into_inner(self).await;
109        let cache = cache.into_inner();
110        let deleted_ids = deleted.into_inner();
111
112        if cache.is_empty() && deleted_ids.is_empty() {
113            log::debug!("Empty transaction was committed.");
114            return Ok(());
115        }
116
117        // clear the weak reference to this transaction
118        *database.transaction.lock().await = None;
119
120        // open a database transaction
121        // Because `rusqlite::Transaction: !Send + !Sync`, it's critical that
122        // we don't hold this transaction over any `.await` points.
123        let mut conn = database.conn().await;
124        let tx = conn.transaction()?;
125
126        for entity in cache.values() {
127            entity.execute_save(&tx)?;
128        }
129
130        for deleted_id in deleted_ids.iter() {
131            deleted_id.execute_delete(&tx)?;
132        }
133
134        // and commit everything
135        tx.commit()?;
136
137        Ok(())
138    }
139}
core_crypto_keystore/transaction/mod.rs

core_crypto_keystore/transaction/
mod.rs
