reifydb_transaction/
commit.rs1use std::sync::Arc;
5
6use reifydb_core::common::CommitVersion;
7use reifydb_value::Result;
8use tracing::instrument;
9
10use crate::transaction::command::CommandTransaction;
11
12pub type CommitApply = Box<dyn FnOnce(&mut CommandTransaction) -> Result<()> + Send>;
13
14pub type CommitCompletion = Box<dyn FnOnce(Result<CommitVersion>) + Send>;
15
16pub struct CommitSubmission {
17 pub apply: CommitApply,
18 pub completion: CommitCompletion,
19}
20
21pub type CommitBegin = Arc<dyn Fn() -> Result<CommandTransaction> + Send + Sync>;
22
23#[derive(Clone)]
24pub struct CommitHandle {
25 begin: CommitBegin,
26}
27
28impl CommitHandle {
29 pub fn new(begin: CommitBegin) -> Self {
30 Self {
31 begin,
32 }
33 }
34
35 #[instrument(name = "transaction::commit::submit", level = "debug", skip_all)]
36 pub fn submit(&self, submission: CommitSubmission) {
37 let CommitSubmission {
38 apply,
39 completion,
40 } = submission;
41 (completion)(apply_and_commit(&self.begin, apply));
42 }
43}
44
45fn apply_and_commit(begin: &CommitBegin, apply: CommitApply) -> Result<CommitVersion> {
46 let mut transaction = (begin)()?;
47 if let Err(e) = transaction.disable_conflict_tracking() {
48 let _ = transaction.rollback();
49 return Err(e);
50 }
51 if let Err(e) = (apply)(&mut transaction) {
52 let _ = transaction.rollback();
53 return Err(e);
54 }
55 transaction.commit_unchecked()
56}