nodedb_lite/sync/array/
outbound.rs

1//! High-level emitter for array CRDT ops on the send path.
2//!
3//! [`ArrayOutbound`] is called by the array engine hooks after each local
4//! mutation succeeds. It:
5//!
6//! 1. Looks up the current `schema_hlc` from the [`SchemaRegistry`].
7//! 2. Mints a fresh HLC via [`ReplicaState::next_hlc`].
8//! 3. Builds the [`ArrayOp`].
9//! 4. Appends to the durable [`KvOpLogStore`] (permanent record for GC).
10//! 5. Enqueues in the durable [`PendingQueue`], which the sync push loop
11//!    drains to Origin.
12//!
13//! The caller must ensure the local engine write has already succeeded before
14//! calling [`ArrayOutbound`] methods. If emit fails after a successful engine
15//! write, the error is returned to the caller and the write is **not** rolled
16//! back. The cell stays local and never reaches Origin.
17
18use std::sync::Arc;
19
20use nodedb_array::sync::hlc::Hlc;
21use nodedb_array::sync::op::{ArrayOp, ArrayOpHeader, ArrayOpKind};
22use nodedb_array::sync::op_log::OpLog;
23use nodedb_array::types::cell_value::value::CellValue;
24use nodedb_array::types::coord::value::CoordValue;
25
26use crate::error::LiteError;
27use crate::storage::engine::StorageEngine;
28use crate::sync::array::op_log_store::KvOpLogStore;
29use crate::sync::array::pending::PendingQueue;
30use crate::sync::array::replica_state::ReplicaState;
31use crate::sync::array::schema_registry::SchemaRegistry;
32
33/// Emitter for locally-originated array CRDT ops.
34///
35/// All fields are `Arc`-wrapped so the struct can be shared across the
36/// `NodeDbLite` struct and any future transport tasks.
37pub struct ArrayOutbound<S: StorageEngine> {
38    pub(crate) op_log: Arc<KvOpLogStore<S>>,
39    pub(crate) pending: Arc<PendingQueue<S>>,
40    pub(crate) schemas: Arc<SchemaRegistry<S>>,
41    pub(crate) replica: Arc<ReplicaState>,
42}
43
44impl<S: StorageEngine> ArrayOutbound<S> {
45    /// Create an [`ArrayOutbound`] from its component parts.
46    pub fn new(
47        op_log: Arc<KvOpLogStore<S>>,
48        pending: Arc<PendingQueue<S>>,
49        schemas: Arc<SchemaRegistry<S>>,
50        replica: Arc<ReplicaState>,
51    ) -> Self {
52        Self {
53            op_log,
54            pending,
55            schemas,
56            replica,
57        }
58    }
59
60    /// Access the underlying op-log (for sharing with inbound handler).
61    pub fn op_log(&self) -> &Arc<KvOpLogStore<S>> {
62        &self.op_log
63    }
64
65    /// Access the underlying pending queue (for sharing with inbound handler).
66    pub fn pending(&self) -> &Arc<PendingQueue<S>> {
67        &self.pending
68    }
69
70    /// Emit a `Put` op for a cell write.
71    ///
72    /// `coord` and `attrs` must be the same values passed to the array engine
73    /// (cloned before the engine call to avoid moves).
74    pub async fn emit_put(
75        &self,
76        array: &str,
77        coord: Vec<CoordValue>,
78        attrs: Vec<CellValue>,
79        valid_from_ms: i64,
80        valid_until_ms: i64,
81    ) -> Result<Hlc, LiteError> {
82        let schema_hlc = self.require_schema_hlc(array)?;
83        let hlc = self.replica.next_hlc()?;
84
85        let op = ArrayOp {
86            header: ArrayOpHeader {
87                array: array.into(),
88                hlc,
89                schema_hlc,
90                valid_from_ms,
91                valid_until_ms,
92                system_from_ms: hlc.physical_ms as i64,
93            },
94            kind: ArrayOpKind::Put,
95            coord,
96            attrs: Some(attrs),
97        };
98
99        self.record(&op).await?;
100        Ok(hlc)
101    }
102
103    /// Emit a `Delete` (soft tombstone) op.
104    ///
105    /// `valid_from_ms` / `valid_until_ms` default to `0` / `i64::MAX` at the
106    /// call sites because the current [`NodeDbLite::array_delete_cell`] API
107    /// does not yet carry valid-time arguments. Phase F will widen the API.
108    pub async fn emit_delete(
109        &self,
110        array: &str,
111        coord: Vec<CoordValue>,
112        valid_from_ms: i64,
113        valid_until_ms: i64,
114    ) -> Result<Hlc, LiteError> {
115        let schema_hlc = self.require_schema_hlc(array)?;
116        let hlc = self.replica.next_hlc()?;
117
118        let op = ArrayOp {
119            header: ArrayOpHeader {
120                array: array.into(),
121                hlc,
122                schema_hlc,
123                valid_from_ms,
124                valid_until_ms,
125                system_from_ms: hlc.physical_ms as i64,
126            },
127            kind: ArrayOpKind::Delete,
128            coord,
129            attrs: None,
130        };
131
132        self.record(&op).await?;
133        Ok(hlc)
134    }
135
136    /// Emit an `Erase` (GDPR hard tombstone) op.
137    ///
138    /// Same valid-time defaulting as [`emit_delete`].
139    pub async fn emit_erase(
140        &self,
141        array: &str,
142        coord: Vec<CoordValue>,
143        valid_from_ms: i64,
144        valid_until_ms: i64,
145    ) -> Result<Hlc, LiteError> {
146        let schema_hlc = self.require_schema_hlc(array)?;
147        let hlc = self.replica.next_hlc()?;
148
149        let op = ArrayOp {
150            header: ArrayOpHeader {
151                array: array.into(),
152                hlc,
153                schema_hlc,
154                valid_from_ms,
155                valid_until_ms,
156                system_from_ms: hlc.physical_ms as i64,
157            },
158            kind: ArrayOpKind::Erase,
159            coord,
160            attrs: None,
161        };
162
163        self.record(&op).await?;
164        Ok(hlc)
165    }
166
167    // ─── Internal helpers ─────────────────────────────────────────────────────
168
169    /// Look up the current `schema_hlc` for `array`, returning an error if the
170    /// array was never registered via [`SchemaRegistry::put_schema`].
171    fn require_schema_hlc(&self, array: &str) -> Result<Hlc, LiteError> {
172        self.schemas
173            .schema_hlc(array)
174            .ok_or_else(|| LiteError::Storage {
175                detail: format!("array '{array}' has no schema CRDT — call create_array first"),
176            })
177    }
178
179    /// Append to op-log then enqueue for transport.
180    async fn record(&self, op: &ArrayOp) -> Result<(), LiteError> {
181        self.op_log.append(op).map_err(|e| LiteError::Storage {
182            detail: format!("array sync op_log: {e}"),
183        })?;
184        self.pending.enqueue(op).await?;
185        Ok(())
186    }
187}
188
189// ─── Tests ────────────────────────────────────────────────────────────────────
190
191#[cfg(test)]
192mod tests {
193    use super::*;
194    use crate::storage::pagedb_storage::PagedbStorageMem;
195    use nodedb_array::schema::array_schema::ArraySchema;
196    use nodedb_array::schema::attr_spec::{AttrSpec, AttrType};
197    use nodedb_array::schema::cell_order::{CellOrder, TileOrder};
198    use nodedb_array::schema::dim_spec::{DimSpec, DimType};
199    use nodedb_array::sync::op_log::OpLog;
200    use nodedb_array::types::domain::{Domain, DomainBound};
201
202    fn simple_schema(name: &str) -> ArraySchema {
203        ArraySchema {
204            name: name.into(),
205            dims: vec![DimSpec::new(
206                "x",
207                DimType::Int64,
208                Domain::new(DomainBound::Int64(0), DomainBound::Int64(99)),
209            )],
210            attrs: vec![AttrSpec::new("v", AttrType::Float64, true)],
211            tile_extents: vec![10],
212            cell_order: CellOrder::RowMajor,
213            tile_order: TileOrder::RowMajor,
214        }
215    }
216
217    // multi_thread flavor needed because emit_put uses pending.enqueue (async)
218    // and op_log.append uses block_in_place.
219
220    async fn make_outbound() -> (ArrayOutbound<PagedbStorageMem>, Arc<PagedbStorageMem>) {
221        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
222        let replica = Arc::new(ReplicaState::load_or_init(&*storage).await.unwrap());
223        let schemas = Arc::new(SchemaRegistry::new(
224            Arc::clone(&storage),
225            Arc::clone(&replica),
226        ));
227        let op_log = Arc::new(KvOpLogStore::new(Arc::clone(&storage)));
228        let pending = Arc::new(PendingQueue::new(Arc::clone(&storage)));
229        let ob = ArrayOutbound::new(op_log, pending, schemas, replica);
230        (ob, storage)
231    }
232
233    #[tokio::test(flavor = "multi_thread")]
234    async fn emit_put_appends_to_log_and_queue() {
235        let (ob, _storage) = make_outbound().await;
236        ob.schemas
237            .put_schema("arr", &simple_schema("arr"))
238            .await
239            .unwrap();
240
241        let coord = vec![CoordValue::Int64(5)];
242        let attrs = vec![CellValue::Null];
243        ob.emit_put("arr", coord, attrs, 0, i64::MAX).await.unwrap();
244
245        assert_eq!(ob.op_log.len().unwrap(), 1);
246        assert_eq!(ob.pending.len().await.unwrap(), 1);
247    }
248
249    #[tokio::test(flavor = "multi_thread")]
250    async fn emit_without_schema_errors() {
251        let (ob, _storage) = make_outbound().await;
252        let err = ob
253            .emit_put(
254                "unknown",
255                vec![CoordValue::Int64(0)],
256                vec![CellValue::Null],
257                0,
258                -1,
259            )
260            .await
261            .unwrap_err();
262        assert!(
263            matches!(err, LiteError::Storage { ref detail } if detail.contains("no schema CRDT")),
264            "unexpected error: {err:?}"
265        );
266    }
267
268    #[tokio::test(flavor = "multi_thread")]
269    async fn emit_delete_carries_no_attrs() {
270        let (ob, _storage) = make_outbound().await;
271        ob.schemas
272            .put_schema("d", &simple_schema("d"))
273            .await
274            .unwrap();
275
276        ob.emit_delete("d", vec![CoordValue::Int64(1)], 0, i64::MAX)
277            .await
278            .unwrap();
279
280        let ops: Vec<_> = ob
281            .op_log
282            .scan_from(Hlc::ZERO)
283            .unwrap()
284            .map(|r| r.unwrap())
285            .collect();
286        assert_eq!(ops.len(), 1);
287        assert!(ops[0].attrs.is_none(), "Delete must carry no attrs");
288    }
289
290    #[tokio::test(flavor = "multi_thread")]
291    async fn emit_erase_carries_no_attrs() {
292        let (ob, _storage) = make_outbound().await;
293        ob.schemas
294            .put_schema("e", &simple_schema("e"))
295            .await
296            .unwrap();
297
298        ob.emit_erase("e", vec![CoordValue::Int64(2)], 0, i64::MAX)
299            .await
300            .unwrap();
301
302        let ops: Vec<_> = ob
303            .op_log
304            .scan_from(Hlc::ZERO)
305            .unwrap()
306            .map(|r| r.unwrap())
307            .collect();
308        assert_eq!(ops.len(), 1);
309        assert!(ops[0].attrs.is_none(), "Erase must carry no attrs");
310    }
311
312    #[tokio::test(flavor = "multi_thread")]
313    async fn emit_advances_hlc() {
314        let (ob, _storage) = make_outbound().await;
315        ob.schemas
316            .put_schema("a", &simple_schema("a"))
317            .await
318            .unwrap();
319
320        let h1 = ob
321            .emit_put(
322                "a",
323                vec![CoordValue::Int64(1)],
324                vec![CellValue::Null],
325                0,
326                i64::MAX,
327            )
328            .await
329            .unwrap();
330        let h2 = ob
331            .emit_put(
332                "a",
333                vec![CoordValue::Int64(2)],
334                vec![CellValue::Null],
335                0,
336                i64::MAX,
337            )
338            .await
339            .unwrap();
340        assert!(h2 > h1, "each emit must mint a strictly greater HLC");
341    }
342}
nodedb_lite/sync/array/outbound.rs

nodedb_lite/sync/array/
outbound.rs
