linkrs-transaction 0.1.0

Transaction management for Linkrs
Documentation 
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

//! Transaction savepoint management

use super::TransactionManager;
use crate::error::TransactionError;
use crate::types::*;
use crate::undo_log::UndoTarget;
use linkrs_core::types::StagedWriteMark;

impl TransactionManager {
    /// Create savepoint
    ///
    /// `staged_mark` is the storage-staged write boundary the caller peeked
    /// from the undo target before this call (`None` when the target holds
    /// no staged writes). It is required explicitly so a staging target can
    /// never silently skip the savepoint mark.
    pub fn create_savepoint(
        &self,
        txn_id: TransactionId,
        name: Option<String>,
        staged_mark: Option<StagedWriteMark>,
    ) -> Result<SavepointId, TransactionError> {
        let context = self.get_context(txn_id)?;
        context.can_execute()?;
        // Savepoint names resolve to the latest match, so a duplicate
        // would silently shadow the earlier one; reject it instead.
        if let Some(ref wanted) = name {
            if context.find_savepoint_by_name(wanted).is_some() {
                return Err(TransactionError::savepoint_failed(format!(
                    "Savepoint '{}' already exists",
                    wanted
                )));
            }
        }
        let sync_sequence = self
            .sync_manager
            .as_ref()
            .map(|manager| manager.pending_transaction_intent_sequence(txn_id))
            .unwrap_or(0);
        let id = context.create_savepoint(name, sync_sequence);
        context.set_savepoint_staging_mark(id, staged_mark)?;
        Ok(id)
    }

    /// Get savepoint info
    pub fn get_savepoint(&self, txn_id: TransactionId, id: SavepointId) -> Option<SavepointInfo> {
        let context = self.get_context(txn_id).ok()?;
        context.get_savepoint(id)
    }

    /// Release savepoint
    pub fn release_savepoint(
        &self,
        txn_id: TransactionId,
        id: SavepointId,
    ) -> Result<(), TransactionError> {
        let context = self.get_context(txn_id)?;
        context.release_savepoint(id)
    }

    /// Rollback to savepoint
    pub fn rollback_to_savepoint<T: UndoTarget + ?Sized>(
        &self,
        txn_id: TransactionId,
        id: SavepointId,
        target: &T,
    ) -> Result<(), TransactionError> {
        let context = self.get_context(txn_id)?;
        let savepoint = context
            .get_savepoint(id)
            .ok_or(TransactionError::savepoint_not_found(id))?;

        if let Some(sync_manager) = self.sync_manager.as_ref() {
            sync_manager
                .rollback_transaction_to_sequence_sync(txn_id, savepoint.sync_sequence)
                .map_err(|e| TransactionError::sync_failed(e.to_string()))?;
        }

        context
            .rollback_to_savepoint(id, target)
            .map_err(|e| TransactionError::rollback_failed(e.to_string()))?;

        Ok(())
    }

    /// Get all active savepoints for transaction
    pub fn get_active_savepoints(&self, txn_id: TransactionId) -> Vec<SavepointInfo> {
        self.get_context(txn_id)
            .map(|ctx| ctx.get_all_savepoints())
            .unwrap_or_default()
    }
}