nodedb 0.2.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
98
99
100
101
102
103

// SPDX-License-Identifier: BUSL-1.1

//! Ceiling resolver — reverse range scan with optional valid-time filter.

use super::keys::{
    EdgeRef, edge_version_prefix, is_gdpr_erasure, is_sentinel, is_tombstone,
    parse_versioned_edge_key, versioned_edge_key,
};
use super::payload::EdgeValuePayload;
use crate::engine::graph::edge_store::store::{EDGES, Edge, EdgeStore, redb_err};

impl EdgeStore {
    /// Resolve the Ceiling: the latest version of
    /// `(collection, src, label, dst)` whose `system_from ≤ system_as_of`.
    ///
    /// Returns `Ok(None)` if no version exists at or before the cutoff, or if
    /// the latest qualifying version is a tombstone/GDPR erasure.
    ///
    /// When `valid_at_ms` is supplied, the resolved version must also satisfy
    /// `valid_from_ms ≤ valid_at_ms < valid_until_ms`; otherwise the method
    /// continues scanning to earlier system-time versions.
    pub fn ceiling_resolve_edge(
        &self,
        edge: EdgeRef<'_>,
        system_as_of: i64,
        valid_at_ms: Option<i64>,
    ) -> crate::Result<Option<Vec<u8>>> {
        if system_as_of < 0 {
            return Err(crate::Error::BadRequest {
                detail: format!("ceiling_resolve_edge: negative system_as_of={system_as_of}"),
            });
        }
        let prefix = edge_version_prefix(edge.collection, edge.src, edge.label, edge.dst);
        let upper = versioned_edge_key(
            edge.collection,
            edge.src,
            edge.label,
            edge.dst,
            system_as_of,
        )?;
        let t = edge.tid.as_u64();

        let read_txn = self
            .db
            .begin_read()
            .map_err(|e| redb_err("begin_read", e))?;
        let table = read_txn
            .open_table(EDGES)
            .map_err(|e| redb_err("open edges", e))?;

        // Inclusive upper — the exact key at system_as_of is a valid ceiling.
        let range = table
            .range((t, prefix.as_str())..=(t, upper.as_str()))
            .map_err(|e| redb_err("ceiling range", e))?;

        // Walk newest-first by reversing the iterator.
        for entry in range.rev() {
            let (k, v) = entry.map_err(|e| redb_err("ceiling iter", e))?;
            let (kt, composite) = k.value();
            if kt != t || !composite.starts_with(&prefix) {
                break;
            }
            let bytes = v.value();
            if is_tombstone(bytes) || is_gdpr_erasure(bytes) {
                return Ok(None);
            }
            let payload = EdgeValuePayload::decode(bytes)?;
            match valid_at_ms {
                Some(vt) if !(payload.valid_from_ms <= vt && vt < payload.valid_until_ms) => {
                    // This system-time version didn't assert the fact at `vt` —
                    // scan older versions.
                    continue;
                }
                _ => return Ok(Some(payload.properties)),
            }
        }
        Ok(None)
    }
}

/// Decode a raw edge value to an [`Edge`] projection, treating sentinels as
/// absent. Used by future current-state scanners.
#[allow(dead_code)]
pub(crate) fn edge_from_versioned_entry(
    composite: &str,
    value: &[u8],
) -> Option<(Edge, EdgeValuePayload)> {
    if is_sentinel(value) {
        return None;
    }
    let (collection, src, label, dst, _sys) = parse_versioned_edge_key(composite)?;
    let payload = EdgeValuePayload::decode(value).ok()?;
    Some((
        Edge {
            collection: collection.to_string(),
            src_id: src.to_string(),
            label: label.to_string(),
            dst_id: dst.to_string(),
            properties: payload.properties.clone(),
        },
        payload,
    ))
}