nodedb_lite/sync/array/
pending.rs

1//! Durable FIFO queue of outbound array ops waiting for transport to Origin.
2//!
3//! Backed by [`Namespace::ArrayDelta`]. Each entry's key is the 18-byte
4//! big-endian HLC encoding from [`Hlc::to_bytes`], giving natural FIFO order
5//! because HLCs are strictly monotonic per replica and `to_bytes` is
6//! byte-comparable.
7//!
8//! Each value holds the encoded op and its stream seq. The seq is `0` until
9//! the push loop first sends the op. The push loop persists the seq before
10//! that send, so a re-send after a reconnect carries the same seq.
11//!
12//! An op leaves the queue when Origin confirms it ([`PendingQueue::retire`])
13//! or refuses it permanently ([`PendingQueue::remove`]).
14//!
15//! # Backpressure
16//!
17//! When [`PendingQueue::len`] reaches [`PendingQueue::DEFAULT_CAP`] (or the
18//! cap supplied to [`PendingQueue::with_cap`]), [`PendingQueue::enqueue`]
19//! returns [`LiteError::Backpressure`] instead of writing to storage.
20
21use std::sync::Arc;
22
23use nodedb_array::sync::hlc::Hlc;
24use nodedb_array::sync::op::ArrayOp;
25use nodedb_array::sync::op_codec;
26use nodedb_types::Namespace;
27
28use crate::error::LiteError;
29use crate::storage::engine::{StorageEngine, WriteOp};
30
31/// Value stored per queued op.
32#[derive(zerompk::ToMessagePack, zerompk::FromMessagePack)]
33struct StoredOp {
34    /// Stream seq, `0` until the op is first sent.
35    seq: u64,
36    /// `op_codec` encoding of the op.
37    op: Vec<u8>,
38}
39
40/// One queued op, in the shape the push loop sends it.
41#[derive(Debug, Clone, PartialEq)]
42pub struct PendingArrayOp {
43    /// Target array.
44    pub array: String,
45    /// The op's HLC. It is also the op's queue key.
46    pub hlc: Hlc,
47    /// The `op_codec` encoding, sent verbatim as the delta payload.
48    pub op_payload: Vec<u8>,
49    /// Stream seq, `0` until the op is first sent.
50    pub seq: u64,
51}
52
53fn decode_stored(value: &[u8]) -> Result<StoredOp, LiteError> {
54    zerompk::from_msgpack(value).map_err(|e| LiteError::Storage {
55        detail: format!("pending queue entry decode: {e}"),
56    })
57}
58
59fn encode_stored(stored: &StoredOp) -> Result<Vec<u8>, LiteError> {
60    zerompk::to_msgpack_vec(stored).map_err(|e| LiteError::Serialization {
61        detail: format!("pending queue entry encode: {e}"),
62    })
63}
64
65fn decode_payload(payload: &[u8]) -> Result<ArrayOp, LiteError> {
66    op_codec::decode_op(payload).map_err(|e| LiteError::Storage {
67        detail: format!("pending queue decode: {e}"),
68    })
69}
70
71/// The smallest key strictly greater than `hlc`'s key.
72///
73/// Every key is exactly 18 bytes, so appending one zero byte orders directly
74/// after `hlc` and before every larger 18-byte key.
75fn key_after(hlc: Hlc) -> Vec<u8> {
76    let mut key = hlc.to_bytes().to_vec();
77    key.push(0);
78    key
79}
80
81/// Durable outbound pending-op queue backed by [`Namespace::ArrayDelta`].
82pub struct PendingQueue<S: StorageEngine> {
83    storage: Arc<S>,
84    cap: usize,
85}
86
87impl<S: StorageEngine> PendingQueue<S> {
88    /// Default maximum number of pending ops before backpressure kicks in.
89    pub const DEFAULT_CAP: usize = 100_000;
90
91    /// Create a queue with [`DEFAULT_CAP`].
92    pub fn new(storage: Arc<S>) -> Self {
93        Self {
94            storage,
95            cap: Self::DEFAULT_CAP,
96        }
97    }
98
99    /// Create a queue with a custom cap (useful for tests with small limits).
100    pub fn with_cap(storage: Arc<S>, cap: usize) -> Self {
101        Self { storage, cap }
102    }
103
104    /// Enqueue a single op.
105    ///
106    /// Returns [`LiteError::Backpressure`] when `len() >= cap` to prevent
107    /// unbounded queue growth during prolonged offline operation.
108    pub async fn enqueue(&self, op: &ArrayOp) -> Result<(), LiteError> {
109        let current = self.len().await?;
110        if current >= self.cap as u64 {
111            return Err(LiteError::Backpressure {
112                detail: format!(
113                    "array pending queue full ({current} >= {}); local writes paused until Origin sync drains the queue",
114                    self.cap
115                ),
116            });
117        }
118        let key = op.header.hlc.to_bytes().to_vec();
119        let payload = op_codec::encode_op(op).map_err(|e| LiteError::Storage {
120            detail: format!("pending queue encode: {e}"),
121        })?;
122        let value = encode_stored(&StoredOp {
123            seq: 0,
124            op: payload,
125        })?;
126        self.storage.put(Namespace::ArrayDelta, &key, &value).await
127    }
128
129    /// Return up to `limit` queued ops with an HLC strictly greater than
130    /// `after`, in FIFO order. `None` starts at the head of the queue.
131    pub async fn scan_after(
132        &self,
133        after: Option<Hlc>,
134        limit: usize,
135    ) -> Result<Vec<PendingArrayOp>, LiteError> {
136        let start = after.map(key_after).unwrap_or_default();
137        let pairs = self
138            .storage
139            .scan_range(Namespace::ArrayDelta, &start, limit)
140            .await?;
141
142        let mut ops = Vec::with_capacity(pairs.len());
143        for (_key, value) in pairs {
144            let stored = decode_stored(&value)?;
145            let op = decode_payload(&stored.op)?;
146            ops.push(PendingArrayOp {
147                array: op.header.array,
148                hlc: op.header.hlc,
149                op_payload: stored.op,
150                seq: stored.seq,
151            });
152        }
153        Ok(ops)
154    }
155
156    /// Persist `seq` as the stream seq of the op queued under `hlc`.
157    ///
158    /// Returns `false` when the op is no longer queued. The push loop calls
159    /// this only for an op it has not sent yet, so no ack or reject can
160    /// remove the op between the read and the write.
161    pub async fn set_seq(&self, hlc: Hlc, seq: u64) -> Result<bool, LiteError> {
162        let key = hlc.to_bytes();
163        let Some(value) = self.storage.get(Namespace::ArrayDelta, &key).await? else {
164            return Ok(false);
165        };
166        let mut stored = decode_stored(&value)?;
167        stored.seq = seq;
168        self.storage
169            .put(Namespace::ArrayDelta, &key, &encode_stored(&stored)?)
170            .await?;
171        Ok(true)
172    }
173
174    /// Remove every op named in `hlcs` in one atomic write.
175    ///
176    /// Ops no longer queued are skipped. Returns the number removed.
177    pub async fn retire(&self, hlcs: &[Hlc]) -> Result<u64, LiteError> {
178        let mut deletes = Vec::with_capacity(hlcs.len());
179        for hlc in hlcs {
180            let key = hlc.to_bytes().to_vec();
181            if self
182                .storage
183                .get(Namespace::ArrayDelta, &key)
184                .await?
185                .is_some()
186            {
187                deletes.push(WriteOp::Delete {
188                    ns: Namespace::ArrayDelta,
189                    key,
190                });
191            }
192        }
193        let count = deletes.len() as u64;
194        if !deletes.is_empty() {
195            self.storage.batch_write(&deletes).await?;
196        }
197        Ok(count)
198    }
199
200    /// Return up to `limit` ops in FIFO order (lowest HLC first).
201    ///
202    /// Does not remove the returned ops — call [`ack_through`] after they are
203    /// confirmed by Origin.
204    pub async fn drain_batch(&self, limit: usize) -> Result<Vec<ArrayOp>, LiteError> {
205        let pairs = self
206            .storage
207            .scan_range(Namespace::ArrayDelta, &[], limit)
208            .await?;
209
210        let mut ops = Vec::with_capacity(pairs.len());
211        for (_key, value) in pairs {
212            let stored = decode_stored(&value)?;
213            ops.push(decode_payload(&stored.op)?);
214        }
215        Ok(ops)
216    }
217
218    /// Remove all ops whose `header.hlc <= ack_hlc`.
219    ///
220    /// Called after Origin acknowledges a batch up to `ack_hlc`. Returns the
221    /// number of ops removed.
222    pub async fn ack_through(&self, ack_hlc: Hlc) -> Result<u64, LiteError> {
223        // HLC bytes are byte-comparable; scan from the start and stop when we
224        // hit an entry with a key > ack_hlc bytes.
225        let ack_key = ack_hlc.to_bytes();
226        let pairs = self
227            .storage
228            .scan_range(Namespace::ArrayDelta, &[], usize::MAX)
229            .await?;
230
231        let to_delete: Vec<WriteOp> = pairs
232            .into_iter()
233            .take_while(|(key, _)| {
234                if key.len() != 18 {
235                    return false;
236                }
237                // key <= ack_key (bytes are big-endian comparable)
238                key.as_slice() <= ack_key.as_slice()
239            })
240            .map(|(key, _)| WriteOp::Delete {
241                ns: Namespace::ArrayDelta,
242                key,
243            })
244            .collect();
245
246        let count = to_delete.len() as u64;
247        if !to_delete.is_empty() {
248            self.storage.batch_write(&to_delete).await?;
249        }
250        Ok(count)
251    }
252
253    /// Total count of pending ops in storage.
254    pub async fn len(&self) -> Result<u64, LiteError> {
255        self.storage.count(Namespace::ArrayDelta).await
256    }
257
258    /// Returns `true` if the queue contains no pending ops.
259    pub async fn is_empty(&self) -> Result<bool, LiteError> {
260        self.len().await.map(|n| n == 0)
261    }
262
263    /// Remove a single op identified by `hlc` from the queue.
264    ///
265    /// Returns `true` if the op existed and was deleted, `false` if it was
266    /// not present (idempotent — safe to call on already-acked ops).
267    ///
268    /// Used by the inbound reject handler to roll back a single optimistic
269    /// local write without touching the rest of the queue.
270    pub async fn remove(&self, hlc: Hlc) -> Result<bool, LiteError> {
271        let key = hlc.to_bytes().to_vec();
272        let exists = self
273            .storage
274            .get(Namespace::ArrayDelta, &key)
275            .await?
276            .is_some();
277        if exists {
278            self.storage.delete(Namespace::ArrayDelta, &key).await?;
279        }
280        Ok(exists)
281    }
282}
283
284// ─── Tests ────────────────────────────────────────────────────────────────────
285
286#[cfg(test)]
287mod tests {
288    use super::*;
289    use crate::storage::pagedb_storage::{PagedbStorageDefault, PagedbStorageMem};
290    use nodedb_array::sync::op::{ArrayOpHeader, ArrayOpKind};
291    use nodedb_array::sync::replica_id::ReplicaId;
292    use nodedb_array::types::cell_value::value::CellValue;
293    use nodedb_array::types::coord::value::CoordValue;
294
295    fn replica() -> ReplicaId {
296        ReplicaId::new(1)
297    }
298
299    fn hlc(ms: u64) -> Hlc {
300        Hlc::new(ms, 0, replica()).unwrap()
301    }
302
303    fn make_op(ms: u64) -> ArrayOp {
304        ArrayOp {
305            header: ArrayOpHeader {
306                array: "arr".into(),
307                hlc: hlc(ms),
308                schema_hlc: hlc(1),
309                valid_from_ms: 0,
310                valid_until_ms: -1,
311                system_from_ms: ms as i64,
312            },
313            kind: ArrayOpKind::Put,
314            coord: vec![CoordValue::Int64(ms as i64)],
315            attrs: Some(vec![CellValue::Null]),
316        }
317    }
318
319    async fn make_queue() -> PendingQueue<PagedbStorageMem> {
320        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
321        PendingQueue::new(storage)
322    }
323
324    #[tokio::test]
325    async fn enqueue_drain_fifo_order() {
326        let q = make_queue().await;
327        q.enqueue(&make_op(10)).await.unwrap();
328        q.enqueue(&make_op(20)).await.unwrap();
329        q.enqueue(&make_op(30)).await.unwrap();
330
331        let ops = q.drain_batch(usize::MAX).await.unwrap();
332        assert_eq!(ops.len(), 3);
333        let ms: Vec<u64> = ops.iter().map(|o| o.header.hlc.physical_ms).collect();
334        assert_eq!(ms, vec![10, 20, 30], "must be FIFO (ascending HLC) order");
335    }
336
337    #[tokio::test]
338    async fn enqueue_full_returns_backpressure() {
339        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
340        let q = PendingQueue::with_cap(Arc::clone(&storage), 2);
341
342        q.enqueue(&make_op(1)).await.unwrap();
343        q.enqueue(&make_op(2)).await.unwrap();
344
345        let err = q.enqueue(&make_op(3)).await.unwrap_err();
346        assert!(
347            matches!(err, LiteError::Backpressure { .. }),
348            "expected Backpressure, got: {err:?}"
349        );
350    }
351
352    #[tokio::test]
353    async fn ack_through_removes_lower() {
354        let q = make_queue().await;
355        q.enqueue(&make_op(10)).await.unwrap();
356        q.enqueue(&make_op(20)).await.unwrap();
357        q.enqueue(&make_op(30)).await.unwrap();
358
359        // Ack through ms=20 (inclusive).
360        let removed = q.ack_through(hlc(20)).await.unwrap();
361        assert_eq!(removed, 2);
362        assert_eq!(q.len().await.unwrap(), 1);
363
364        let remaining = q.drain_batch(usize::MAX).await.unwrap();
365        assert_eq!(remaining.len(), 1);
366        assert_eq!(remaining[0].header.hlc.physical_ms, 30);
367    }
368
369    #[tokio::test]
370    async fn remove_existing_returns_true() {
371        let q = make_queue().await;
372        q.enqueue(&make_op(10)).await.unwrap();
373        q.enqueue(&make_op(20)).await.unwrap();
374
375        let removed = q.remove(hlc(10)).await.unwrap();
376        assert!(removed, "remove of existing op must return true");
377        assert_eq!(q.len().await.unwrap(), 1);
378
379        let remaining = q.drain_batch(usize::MAX).await.unwrap();
380        assert_eq!(remaining[0].header.hlc.physical_ms, 20);
381    }
382
383    #[tokio::test]
384    async fn remove_missing_returns_false() {
385        let q = make_queue().await;
386        q.enqueue(&make_op(10)).await.unwrap();
387
388        let removed = q.remove(hlc(99)).await.unwrap();
389        assert!(!removed, "remove of absent op must return false");
390        assert_eq!(q.len().await.unwrap(), 1, "queue length must be unchanged");
391    }
392
393    #[tokio::test]
394    async fn survives_storage_restart() {
395        let dir = tempfile::tempdir().unwrap();
396        let path = dir.path().join("pending_test.pagedb");
397
398        {
399            let storage = Arc::new(
400                PagedbStorageDefault::open(
401                    &path,
402                    crate::storage::encryption::Encryption::Plaintext,
403                )
404                .await
405                .unwrap(),
406            );
407            let q = PendingQueue::new(storage);
408            q.enqueue(&make_op(5)).await.unwrap();
409            q.enqueue(&make_op(15)).await.unwrap();
410        }
411
412        {
413            let storage = Arc::new(
414                PagedbStorageDefault::open(
415                    &path,
416                    crate::storage::encryption::Encryption::Plaintext,
417                )
418                .await
419                .unwrap(),
420            );
421            let q = PendingQueue::new(storage);
422            assert_eq!(q.len().await.unwrap(), 2);
423            let ops = q.drain_batch(usize::MAX).await.unwrap();
424            let ms: Vec<u64> = ops.iter().map(|o| o.header.hlc.physical_ms).collect();
425            assert_eq!(ms, vec![5, 15]);
426        }
427    }
428
429    fn physical(ops: &[PendingArrayOp]) -> Vec<u64> {
430        ops.iter().map(|o| o.hlc.physical_ms).collect()
431    }
432
433    #[tokio::test]
434    async fn scan_after_starts_strictly_after_the_cursor() {
435        let q = make_queue().await;
436        for ms in [10, 20, 30] {
437            q.enqueue(&make_op(ms)).await.unwrap();
438        }
439
440        let all = q.scan_after(None, usize::MAX).await.unwrap();
441        assert_eq!(physical(&all), vec![10, 20, 30]);
442        assert!(all.iter().all(|o| o.seq == 0 && o.array == "arr"));
443        let decoded = op_codec::decode_op(&all[0].op_payload).unwrap();
444        assert_eq!(decoded.header.hlc, hlc(10), "payload is the encoded op");
445
446        let tail = q.scan_after(Some(hlc(10)), usize::MAX).await.unwrap();
447        assert_eq!(physical(&tail), vec![20, 30]);
448        let bounded = q.scan_after(None, 2).await.unwrap();
449        assert_eq!(physical(&bounded), vec![10, 20]);
450    }
451
452    #[tokio::test]
453    async fn set_seq_persists_the_seq_and_reports_absent_ops() {
454        let q = make_queue().await;
455        q.enqueue(&make_op(10)).await.unwrap();
456
457        assert!(q.set_seq(hlc(10), 7).await.unwrap());
458        assert!(!q.set_seq(hlc(99), 7).await.unwrap());
459
460        let ops = q.scan_after(None, usize::MAX).await.unwrap();
461        assert_eq!(ops[0].seq, 7);
462        let drained = q.drain_batch(usize::MAX).await.unwrap();
463        assert_eq!(
464            drained[0].header.hlc,
465            hlc(10),
466            "the op survives the seq write"
467        );
468    }
469
470    #[tokio::test]
471    async fn retire_removes_exactly_the_named_ops() {
472        let q = make_queue().await;
473        for ms in [10, 20, 30] {
474            q.enqueue(&make_op(ms)).await.unwrap();
475        }
476
477        let removed = q.retire(&[hlc(10), hlc(30), hlc(99)]).await.unwrap();
478        assert_eq!(removed, 2, "an absent op is skipped");
479        let remaining = q.scan_after(None, usize::MAX).await.unwrap();
480        assert_eq!(physical(&remaining), vec![20]);
481    }
482}
nodedb_lite/sync/array/pending.rs

nodedb_lite/sync/array/
pending.rs
