nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
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
92
93
94
95
96
97

// SPDX-License-Identifier: BUSL-1.1

//! Rollback driver for the undo log.

use super::UndoEntry;
use crate::data::executor::core_loop::CoreLoop;

impl CoreLoop {
    /// Roll back completed writes in reverse order.
    ///
    /// Returns `Ok(())` if all undo entries were applied successfully.
    ///
    /// Returns `Err((entry_index, detail))` on the first undo failure —
    /// the entry index is the original forward-order position of the failed
    /// entry (before reversal). On failure the caller **must** return a
    /// `RollbackFailed` error to the client; the shard state is unknown
    /// and requires a restart to restore consistency via WAL replay.
    pub(in crate::data::executor::handlers) fn rollback_undo_log(
        &mut self,
        did: u64,
        tid: u64,
        undo_log: Vec<UndoEntry>,
    ) -> Result<(), (usize, String)> {
        self.rollback_undo_log_inner(did, tid, None, undo_log)
    }

    pub(in crate::data::executor::handlers) fn rollback_undo_log_at(
        &mut self,
        did: u64,
        tid: u64,
        vshard_id: crate::types::VShardId,
        undo_log: Vec<UndoEntry>,
    ) -> Result<(), (usize, String)> {
        self.rollback_undo_log_inner(did, tid, Some(vshard_id), undo_log)
    }

    fn rollback_undo_log_inner(
        &mut self,
        did: u64,
        tid: u64,
        vshard_id: Option<crate::types::VShardId>,
        undo_log: Vec<UndoEntry>,
    ) -> Result<(), (usize, String)> {
        let total = undo_log.len();
        for (rev_idx, entry) in undo_log.into_iter().rev().enumerate() {
            // Convert reversed index back to original forward-order index for
            // diagnostics (makes it easier to correlate with the sub-plan that
            // produced this undo entry).
            let original_idx = total.saturating_sub(1 + rev_idx);
            self.apply_undo_entry(did, tid, vshard_id, original_idx, entry)?;
        }
        Ok(())
    }

    /// Apply a single undo entry. Returns `Err((entry_index, detail))` if the
    /// undo cannot be applied — this is a fatal condition: the shard's in-memory
    /// state is now partially rolled back and must not serve writes.
    fn apply_undo_entry(
        &mut self,
        did: u64,
        tid: u64,
        vshard_id: Option<crate::types::VShardId>,
        entry_index: usize,
        entry: UndoEntry,
    ) -> Result<(), (usize, String)> {
        match entry {
            UndoEntry::PutDocument { .. } | UndoEntry::DeleteDocument { .. } => {
                self.apply_undo_document(did, tid, entry_index, entry)
            }
            UndoEntry::InsertVector { .. } | UndoEntry::DeleteVector { .. } => {
                self.apply_undo_vector(tid, entry_index, entry)
            }
            UndoEntry::SpatialInsert { .. } | UndoEntry::SpatialDelete { .. } => {
                self.apply_undo_spatial(entry_index, entry)
            }
            UndoEntry::PutEdge { ref src_id, .. } | UndoEntry::DeleteEdge { ref src_id, .. } => {
                let account_stats = vshard_id.is_none_or(|vshard_id| {
                    vshard_id == crate::types::VShardId::from_key(src_id.as_bytes())
                });
                self.apply_undo_edge_with_stats(did, tid, entry_index, entry, account_stats)
            }
            UndoEntry::KvPut { .. }
            | UndoEntry::KvDelete { .. }
            | UndoEntry::KvBatchPut { .. }
            | UndoEntry::KvTransfer { .. }
            | UndoEntry::KvTransferItem { .. }
            | UndoEntry::KvTtl { .. }
            | UndoEntry::SortedIndexDdl { .. } => self.apply_undo_kv(did, tid, entry_index, entry),
            UndoEntry::ColumnarInsert { .. }
            | UndoEntry::ColumnarUpdate { .. }
            | UndoEntry::ColumnarDelete { .. } => self.apply_undo_columnar(entry_index, entry),
            UndoEntry::TimeseriesIngest { .. } => self.apply_undo_timeseries(entry_index, entry),
            UndoEntry::StatsRestore { .. } => self.apply_undo_stats(entry_index, entry),
            UndoEntry::MarkNodeDeleted { .. } => self.apply_undo_mark_node(entry_index, entry),
        }
    }
}