nodedb_array/sync/
op_log.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Append-only operation log abstraction for array CRDT sync.
4//!
5//! [`OpLog`] is a storage-agnostic trait. Concrete implementations live in
6//! `nodedb-lite` (redb-backed) and `nodedb` (WAL-backed); [`InMemoryOpLog`]
7//! is a `BTreeMap`-backed implementation suitable for tests in any crate
8//! that depends on `nodedb-array`.
9
10use std::collections::BTreeMap;
11use std::sync::Mutex;
12
13use crate::error::{ArrayError, ArrayResult};
14use crate::sync::hlc::Hlc;
15use crate::sync::op::ArrayOp;
16
17/// Boxed fallible iterator over array ops, returned by [`OpLog`] scans.
18pub type OpIter<'a> = Box<dyn Iterator<Item = ArrayResult<ArrayOp>> + 'a>;
19
20/// Storage-agnostic interface for an append-only array operation log.
21///
22/// Implementations are expected to be `Send + Sync` and durable. An
23/// in-memory implementation suitable for tests is provided by
24/// [`InMemoryOpLog`] below.
25pub trait OpLog: Send + Sync {
26    /// Append an operation to the log.
27    ///
28    /// Must be idempotent: re-appending an op with the same `(array, hlc)`
29    /// is a no-op.
30    fn append(&self, op: &ArrayOp) -> ArrayResult<()>;
31
32    /// Iterate all ops with `hlc >= from`, in HLC order, across all arrays.
33    fn scan_from<'a>(&'a self, from: Hlc) -> ArrayResult<OpIter<'a>>;
34
35    /// Iterate ops for `array` with `from <= hlc <= to`, in HLC order.
36    fn scan_range<'a>(&'a self, array: &str, from: Hlc, to: Hlc) -> ArrayResult<OpIter<'a>>;
37
38    /// Return the total number of ops in the log (across all arrays).
39    fn len(&self) -> ArrayResult<u64>;
40
41    /// Return `true` if the log contains no ops in any array.
42    fn is_empty(&self) -> ArrayResult<bool> {
43        Ok(self.len()? == 0)
44    }
45
46    /// Drop all ops whose `hlc < hlc` and return the count dropped.
47    ///
48    /// Used by GC after snapshotting ops below the min-ack frontier.
49    fn drop_below(&self, hlc: Hlc) -> ArrayResult<u64>;
50}
51
52// ─── In-memory implementation ───────────────────────────────────────────────
53
54type OpMap = BTreeMap<(String, [u8; 18]), ArrayOp>;
55
56/// `BTreeMap`-backed [`OpLog`] implementation.
57///
58/// Key: `(array_name, hlc_bytes)` so iteration is in HLC order per array.
59/// Operations with the same key (same array + same HLC) are idempotent.
60pub struct InMemoryOpLog {
61    ops: Mutex<OpMap>,
62}
63
64impl InMemoryOpLog {
65    /// Create a new empty log.
66    pub fn new() -> Self {
67        Self {
68            ops: Mutex::new(BTreeMap::new()),
69        }
70    }
71
72    fn lock(&self) -> ArrayResult<std::sync::MutexGuard<'_, OpMap>> {
73        self.ops.lock().map_err(|_| ArrayError::HlcLockPoisoned)
74    }
75}
76
77impl Default for InMemoryOpLog {
78    fn default() -> Self {
79        Self::new()
80    }
81}
82
83impl OpLog for InMemoryOpLog {
84    fn append(&self, op: &ArrayOp) -> ArrayResult<()> {
85        let key = (op.header.array.clone(), op.header.hlc.to_bytes());
86        self.lock()?.entry(key).or_insert_with(|| op.clone());
87        Ok(())
88    }
89
90    fn scan_from<'a>(&'a self, from: Hlc) -> ArrayResult<OpIter<'a>> {
91        let guard = self.lock()?;
92        let results: Vec<ArrayOp> = guard
93            .iter()
94            .filter(|((_, hlc_bytes), _)| {
95                let hlc = Hlc::from_bytes(hlc_bytes);
96                hlc >= from
97            })
98            .map(|(_, op)| op.clone())
99            .collect();
100        Ok(Box::new(results.into_iter().map(Ok)))
101    }
102
103    fn scan_range<'a>(&'a self, array: &str, from: Hlc, to: Hlc) -> ArrayResult<OpIter<'a>> {
104        let guard = self.lock()?;
105        let results: Vec<ArrayOp> = guard
106            .iter()
107            .filter(|((arr, hlc_bytes), _)| {
108                if arr != array {
109                    return false;
110                }
111                let hlc = Hlc::from_bytes(hlc_bytes);
112                hlc >= from && hlc <= to
113            })
114            .map(|(_, op)| op.clone())
115            .collect();
116        Ok(Box::new(results.into_iter().map(Ok)))
117    }
118
119    fn len(&self) -> ArrayResult<u64> {
120        Ok(self.lock()?.len() as u64)
121    }
122
123    fn drop_below(&self, hlc: Hlc) -> ArrayResult<u64> {
124        let mut guard = self.lock()?;
125        let before = guard.len() as u64;
126        guard.retain(|(_, hlc_bytes), _| Hlc::from_bytes(hlc_bytes) >= hlc);
127        let after = guard.len() as u64;
128        Ok(before - after)
129    }
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135    use crate::sync::hlc::Hlc;
136    use crate::sync::op::{ArrayOpHeader, ArrayOpKind};
137    use crate::sync::replica_id::ReplicaId;
138    use crate::types::cell_value::value::CellValue;
139    use crate::types::coord::value::CoordValue;
140
141    fn replica() -> ReplicaId {
142        ReplicaId::new(1)
143    }
144
145    fn hlc(ms: u64, logical: u16) -> Hlc {
146        Hlc::new(ms, logical, replica()).unwrap()
147    }
148
149    fn make_op(array: &str, ms: u64, logical: u16) -> ArrayOp {
150        ArrayOp {
151            header: ArrayOpHeader {
152                array: array.into(),
153                hlc: hlc(ms, logical),
154                schema_hlc: hlc(1, 0),
155                valid_from_ms: 0,
156                valid_until_ms: -1,
157                system_from_ms: ms as i64,
158            },
159            kind: ArrayOpKind::Put,
160            coord: vec![CoordValue::Int64(ms as i64)],
161            attrs: Some(vec![CellValue::Null]),
162        }
163    }
164
165    #[test]
166    fn append_then_scan() {
167        let log = InMemoryOpLog::new();
168        log.append(&make_op("arr", 10, 0)).unwrap();
169        log.append(&make_op("arr", 20, 0)).unwrap();
170        assert_eq!(log.len().unwrap(), 2);
171
172        let ops: Vec<_> = log
173            .scan_from(Hlc::ZERO)
174            .unwrap()
175            .map(|r| r.unwrap())
176            .collect();
177        assert_eq!(ops.len(), 2);
178    }
179
180    #[test]
181    fn scan_from_skips_below() {
182        let log = InMemoryOpLog::new();
183        for ms in [10, 20, 30, 40] {
184            log.append(&make_op("arr", ms, 0)).unwrap();
185        }
186
187        let ops: Vec<_> = log
188            .scan_from(hlc(25, 0))
189            .unwrap()
190            .map(|r| r.unwrap())
191            .collect();
192        // Only ms=30 and ms=40 should appear.
193        assert_eq!(ops.len(), 2);
194        assert!(ops.iter().all(|op| op.header.hlc.physical_ms >= 25));
195    }
196
197    #[test]
198    fn drop_below_drops_correctly() {
199        let log = InMemoryOpLog::new();
200        for ms in [10, 20, 30] {
201            log.append(&make_op("arr", ms, 0)).unwrap();
202        }
203        let dropped = log.drop_below(hlc(20, 0)).unwrap();
204        assert_eq!(dropped, 1); // ms=10 dropped
205        assert_eq!(log.len().unwrap(), 2);
206    }
207
208    #[test]
209    fn scan_range_filters_array() {
210        let log = InMemoryOpLog::new();
211        log.append(&make_op("a", 10, 0)).unwrap();
212        log.append(&make_op("b", 20, 0)).unwrap();
213        log.append(&make_op("a", 30, 0)).unwrap();
214
215        let ops: Vec<_> = log
216            .scan_range("a", Hlc::ZERO, hlc(u64::MAX >> 16, 0))
217            .unwrap()
218            .map(|r| r.unwrap())
219            .collect();
220        assert_eq!(ops.len(), 2);
221        assert!(ops.iter().all(|op| op.header.array == "a"));
222    }
223
224    #[test]
225    fn len_counts_all() {
226        let log = InMemoryOpLog::new();
227        assert_eq!(log.len().unwrap(), 0);
228        log.append(&make_op("x", 1, 0)).unwrap();
229        log.append(&make_op("y", 2, 0)).unwrap();
230        assert_eq!(log.len().unwrap(), 2);
231        // Idempotent re-append.
232        log.append(&make_op("x", 1, 0)).unwrap();
233        assert_eq!(log.len().unwrap(), 2);
234    }
235}
nodedb_array/sync/op_log.rs

nodedb_array/sync/
op_log.rs
