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

// SPDX-License-Identifier: BUSL-1.1

//! Dispatch loop for replaying `SurrogateAlloc` / `SurrogateBind` WAL records.

use nodedb_types::{DatabaseId, TenantId};
use nodedb_wal::WalRecord;
use nodedb_wal::record::RecordType;

use crate::control::security::catalog::SystemCatalog;
use crate::control::surrogate::SurrogateRegistryHandle;

use super::{apply_surrogate_alloc, apply_surrogate_bind};

/// Replay every `SurrogateAlloc` and `SurrogateBind` record in `records`
/// into the live `SurrogateRegistry` + `SystemCatalog`. Run once at
/// startup, after the registry has been seeded from the catalog hwm row
/// and after the catalog has been opened — replay then advances both
/// past the WAL's tail so any binding emitted before the crash is
/// durable on the next allocation.
pub fn replay_surrogate_records(
    records: &[WalRecord],
    catalog: &SystemCatalog,
    registry: &SurrogateRegistryHandle,
) -> crate::Result<ReplayStats> {
    let mut stats = ReplayStats::default();
    for record in records {
        let raw = record.logical_record_type();
        let Some(rt) = RecordType::from_raw(raw) else {
            continue;
        };
        match rt {
            RecordType::SurrogateAlloc => {
                apply_surrogate_alloc(&record.payload, registry)?;
                stats.allocs += 1;
            }
            RecordType::SurrogateBind => {
                let db = DatabaseId::new(record.header.database_id);
                let tenant = TenantId::new(record.header.tenant_id);
                apply_surrogate_bind(&record.payload, db, tenant, catalog, registry)?;
                stats.binds += 1;
            }
            RecordType::Noop
            | RecordType::Put
            | RecordType::Delete
            | RecordType::VectorPut
            | RecordType::VectorDelete
            | RecordType::VectorParams
            | RecordType::VectorDirectUpsert
            | RecordType::SparseVectorPut
            | RecordType::SparseVectorDelete
            | RecordType::MultiVectorPut
            | RecordType::MultiVectorDelete
            | RecordType::CrdtDelta
            // CrdtListOp carries no surrogate; the list ops it replays never
            // read `surrogate` (see `CrdtOp::ListInsert`/`ListDelete`/
            // `ListMove`'s dispatch arms, all `surrogate: _`).
            | RecordType::CrdtListOp
            // CrdtDocOp carries the row's own surrogate, but that surrogate was
            // allocated upstream and made durable via SurrogateAlloc/Bind — this
            // record never allocates one, so it does not advance the hwm here.
            | RecordType::CrdtDocOp
            | RecordType::TimeseriesBatch
            | RecordType::LogBatch
            | RecordType::ArrayPut
            | RecordType::ArrayDelete
            | RecordType::ArrayFlush
            | RecordType::Transaction
            | RecordType::TransactionRedo
            | RecordType::Checkpoint
            | RecordType::CollectionTombstoned
            | RecordType::LsnMsAnchor
            | RecordType::TemporalPurge
            | RecordType::CalvinApplied
            // SyncSeqAdvance: not relevant to surrogate replay; the sync
            // idempotency replay pass handles HWM reconstruction.
            | RecordType::SyncSeqAdvance
            | RecordType::FtsIndex
            | RecordType::FtsDelete
            | RecordType::SpatialPut
            | RecordType::SpatialDelete
            | RecordType::GraphNodeLabelSet
            | RecordType::GraphNodeLabelRemove => {}
        }
    }
    Ok(stats)
}

#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct ReplayStats {
    pub allocs: usize,
    pub binds: usize,
}