feat(sync): replace in-memory outbound queues with durable storage-backed queues

Rewrite all five outbound queues (columnar, vector, FTS, spatial,
timeseries) to persist every entry in their respective storage namespace
before returning from `enqueue`. Entries survive process restarts and
disconnects; the sync transport re-drains them on reconnect while
Origin deduplicates via its idempotent-producer gate.

Key changes:
- Add `DurableOutboundQueue<S>`: generic storage-backed bounded queue
  with configurable cap. Returns `LiteError::Backpressure` at capacity.
- Add `StreamSeqTracker<S>`: per-stream monotonic sequence frontier
  persisted in `Namespace::Meta`. Tracks `last_assigned` and
  `last_acked` per stream so frame numbering resumes after restart.
- Add `constants::PUSH_DRAIN_LIMIT` (256) to bound each push cycle.
- Add `reconcile_outbound_enqueue` helper for uniform backpressure
  error logging at enqueue sites.
- FTS and spatial queues gain a staging buffer that is flushed to
  durable storage by the periodic flush path, keeping synchronous
  write paths free of async storage I/O.
- All outbound queue types are now generic over `S: StorageEngine`.

Author: farhan-syah

nodedb-lite/src/sync/constants.rs

@@ -0,0 +1,5 @@
+//! Sync-layer constants shared across push and drain paths.
+
+/// Maximum number of entries drained from a durable outbound queue per sync
+/// push cycle. Keeps each push loop iteration bounded in time and memory.
+pub const PUSH_DRAIN_LIMIT: usize = 256;

nodedb-lite/src/sync/mod.rs

@@ -2,20 +2,26 @@ pub mod array;
 pub mod client;
 pub mod clock;
 pub mod compensation;
+pub mod constants;
 pub mod flow_control;
 pub mod outbound;
 pub mod shapes;
+pub mod stream_seq;
 pub mod transport;
 
+pub use constants::PUSH_DRAIN_LIMIT;
+
 pub use array::KvOpLogStore;
 pub use client::{SyncClient, SyncConfig, SyncState};
 pub use clock::VectorClock;
 pub use compensation::{CompensationEvent, CompensationHandler, CompensationRegistry};
 pub use flow_control::{FlowControlConfig, FlowController, SyncMetrics, SyncMetricsSnapshot};
+pub(crate) use outbound::reconcile_outbound_enqueue;
 pub use outbound::{
-    ColumnarOutbound, FtsOutbound, PendingColumnarBatch, PendingFtsDelete, PendingFtsIndex,
-    PendingSpatialDelete, PendingSpatialInsert, PendingTimeseriesBatch, PendingVectorDelete,
-    PendingVectorInsert, SpatialOutbound, TimeseriesOutbound, VectorOutbound,
+    ColumnarOutbound, DurableOutboundQueue, FtsOutbound, PendingColumnarBatch, PendingFtsDelete,
+    PendingFtsIndex, PendingSpatialDelete, PendingSpatialInsert, PendingTimeseriesBatch,
+    PendingVectorDelete, PendingVectorInsert, SpatialOutbound, TimeseriesOutbound, VectorOutbound,
 };
 pub use shapes::ShapeManager;
+pub use stream_seq::StreamSeqTracker;
 pub use transport::{SyncDelegate, run_sync_loop};

nodedb-lite/src/sync/outbound/columnar.rs

@@ -1,146 +1,256 @@
 //! Columnar insert outbound queue for Lite sync.
 //!
-//! When `ColumnarEngine::insert` is called on Lite, it enqueues rows here.
+//! When a columnar row is inserted on Lite, it is durably enqueued here.
 //! The sync transport drains this queue and sends `ColumnarInsert` wire
-//! frames to Origin. Each batch gets a monotonic `batch_id` for ACK
-//! correlation.
+//! frames to Origin. Each drained entry carries a monotonic durable key used
+//! to delete it from storage once Origin acknowledges receipt.
 //!
-//! The queue is in-memory only (no durable persistence). If Lite restarts
-//! before a batch is ACKed, the rows are already in the local `ColumnarEngine`
-//! segments; full catch-up replay is future work. PREVIEW targets
-//! live-session replication (device never goes offline between insert and
-//! sync).
+//! # Durability
+//!
+//! The queue is backed by [`DurableOutboundQueue`], which persists every
+//! entry in [`Namespace::ColumnarPending`] before returning from `enqueue`.
+//! Entries survive process restarts; a crash between send and ack causes
+//! at-most-one retry (at-least-once delivery to Origin).
+//!
+//! # Backpressure
+//!
+//! When the queue reaches its cap, `enqueue` returns
+//! [`LiteError::Backpressure`], propagating to the caller so writes pause
+//! until the sync transport drains the backlog.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
 
+use nodedb_types::Namespace;
 use nodedb_types::value::Value;
+use tokio::sync::Mutex;
 
-use super::queue::{BatchIdGen, PendingQueue};
+use super::durable_queue::DurableOutboundQueue;
+use crate::error::LiteError;
+use crate::storage::engine::StorageEngine;
 
 /// A single pending batch of columnar rows awaiting sync to Origin.
-#[derive(Debug, Clone)]
+#[derive(
+    Debug,
+    Clone,
+    serde::Serialize,
+    serde::Deserialize,
+    zerompk::ToMessagePack,
+    zerompk::FromMessagePack,
+)]
 pub struct PendingColumnarBatch {
-    /// Monotonic batch ID (per-collection, Lite-assigned).
+    /// Monotonic batch ID (per-collection, Lite-assigned) for ACK correlation.
     pub batch_id: u64,
     /// Collection name.
     pub collection: String,
     /// Rows in schema column order, one `Vec<Value>` per row.
     pub rows: Vec<Vec<Value>>,
     /// MessagePack-serialized `ColumnarSchema` hint. May be empty.
     pub schema_bytes: Vec<u8>,
+    /// Stable idempotent-producer seq for this entry. 0 = not yet assigned;
+    /// assigned at first drain and persisted so re-sends after reconnect reuse
+    /// the same seq (Origin dedups instead of double-applying).
+    #[serde(default)]
+    pub seq: u64,
 }
 
-/// Thread-safe outbound queue for columnar inserts.
+/// Durable outbound queue for columnar inserts.
 ///
-/// Held by `NodeDbLite` and shared with `ColumnarEngine` via `Arc`.
-#[derive(Debug, Default)]
-pub struct ColumnarOutbound {
-    queue: PendingQueue<PendingColumnarBatch>,
-    ids: BatchIdGen,
+/// Held as `Arc<ColumnarOutbound<S>>` by `NodeDbLite` and shared with
+/// `ColumnarEngine` via `Arc`. The inner storage is accessed only for
+/// `enqueue` (from the sync insert path) and `drain_batch`/`ack_keys`
+/// (from the async sync transport path).
+pub struct ColumnarOutbound<S: StorageEngine> {
+    queue: DurableOutboundQueue<S>,
+    ids: AtomicU64,
+    /// batch_id → durable_key for entries that have been sent but not yet
+    /// acked by Origin. Cleared on reconnect so entries are re-drained.
+    in_flight: Mutex<HashMap<u64, Vec<u8>>>,
 }
 
-impl ColumnarOutbound {
-    pub const fn new() -> Self {
-        Self {
-            queue: PendingQueue::new(),
-            ids: BatchIdGen::new(),
-        }
+impl<S: StorageEngine> ColumnarOutbound<S> {
+    /// Open the durable queue backed by [`Namespace::ColumnarPending`].
+    pub async fn open(storage: Arc<S>) -> Result<Self, LiteError> {
+        Self::open_with_cap(storage, DurableOutboundQueue::<S>::DEFAULT_CAP).await
+    }
+
+    /// Open with a custom cap.
+    pub async fn open_with_cap(storage: Arc<S>, cap: usize) -> Result<Self, LiteError> {
+        let queue =
+            DurableOutboundQueue::open_with_cap(storage, Namespace::ColumnarPending, cap).await?;
+        Ok(Self {
+            queue,
+            ids: AtomicU64::new(1),
+            in_flight: Mutex::new(HashMap::new()),
+        })
     }
 
-    /// Enqueue a single row for a collection.
+    /// Durably enqueue a single row for a collection.
     ///
-    /// Rows for the same collection are coalesced into a single batch if a
-    /// pending batch for that collection already exists; otherwise a new
-    /// batch is created with a fresh `batch_id`.
-    pub fn enqueue_row(&self, collection: &str, row: Vec<Value>, schema_bytes: Vec<u8>) {
-        // `with_first_mut` consumes `row` only on the matched path, so we use
-        // an `Option` shuttle to recover ownership when no open batch exists.
-        let mut row_slot = Some(row);
-        let appended = self.queue.with_first_mut(
-            |b| b.collection == collection,
-            |b| {
-                if let Some(r) = row_slot.take() {
-                    b.rows.push(r);
-                }
-            },
-        );
-        if appended.is_some() {
-            return;
-        }
-        let row = row_slot.expect("row preserved when no open batch matched");
-        self.queue.push(PendingColumnarBatch {
-            batch_id: self.ids.next(),
+    /// Rows for the same collection are **not** coalesced here — each call
+    /// produces one durable entry. The sync transport batches by collection
+    /// when building wire frames.
+    ///
+    /// Returns [`LiteError::Backpressure`] when the queue is at cap.
+    pub async fn enqueue_row(
+        &self,
+        collection: &str,
+        row: Vec<Value>,
+        schema_bytes: Vec<u8>,
+    ) -> Result<(), LiteError> {
+        let batch_id = self.ids.fetch_add(1, Ordering::Relaxed);
+        let batch = PendingColumnarBatch {
+            batch_id,
             collection: collection.to_string(),
             rows: vec![row],
             schema_bytes,
-        });
+            seq: 0,
+        };
+        let payload = zerompk::to_msgpack_vec(&batch).map_err(|e| LiteError::Serialization {
+            detail: format!("columnar outbound encode: {e}"),
+        })?;
+        self.queue.enqueue(&payload).await
+    }
+
+    /// Drain up to `limit` pending batches in FIFO order, skipping any entries
+    /// currently in-flight (sent but not yet acked by Origin).
+    ///
+    /// Returns `(durable_key, batch)` pairs. On send success, call
+    /// [`mark_in_flight`] with the batch_id and key. The durable entry is
+    /// deleted only when Origin's ack arrives via [`ack_in_flight`].
+    ///
+    /// Does **not** remove entries from storage.
+    pub async fn drain_batch(
+        &self,
+        limit: usize,
+    ) -> Result<Vec<(Vec<u8>, PendingColumnarBatch)>, LiteError> {
+        let in_flight = self.in_flight.lock().await;
+        let pairs = self.queue.drain_batch(limit).await?;
+        let mut out = Vec::with_capacity(pairs.len());
+        for (key, payload) in pairs {
+            if in_flight.values().any(|k| k == &key) {
+                continue;
+            }
+            let batch: PendingColumnarBatch =
+                zerompk::from_msgpack(&payload).map_err(|e| LiteError::Serialization {
+                    detail: format!("columnar outbound decode: {e}"),
+                })?;
+            out.push((key, batch));
+        }
+        Ok(out)
     }
 
-    /// Drain all pending batches for sending.
-    pub fn drain_pending(&self) -> Vec<PendingColumnarBatch> {
-        self.queue.drain()
+    /// Record that a batch has been sent to Origin and is awaiting its ack.
+    ///
+    /// The durable entry is kept in storage until [`ack_in_flight`] is called.
+    pub async fn mark_in_flight(&self, batch_id: u64, durable_key: Vec<u8>) {
+        self.in_flight.lock().await.insert(batch_id, durable_key);
     }
 
-    /// Remove the batch with the given `batch_id` (ACK path; no-op if absent).
-    pub fn acknowledge_batch(&self, batch_id: u64) {
-        self.queue.retain(|b| b.batch_id != batch_id);
+    /// Remove the in-flight record for `batch_id` and return its durable key.
+    ///
+    /// Returns `Some(key)` if the entry was in-flight; `None` if already acked
+    /// or not tracked (e.g. the entry was un-encodable and dropped at send time).
+    pub async fn ack_in_flight(&self, batch_id: u64) -> Option<Vec<u8>> {
+        self.in_flight.lock().await.remove(&batch_id)
     }
 
-    /// Re-queue a rejected batch at the head for retry on the next drain.
-    pub fn requeue_batch(&self, batch: PendingColumnarBatch) {
-        self.queue.requeue(batch);
+    /// Clear all in-flight records on reconnect.
+    ///
+    /// The durable entries are still in storage and will be re-drained on the
+    /// next push tick. Origin's idempotent gate deduplicates re-sent batches.
+    pub async fn clear_in_flight(&self) {
+        self.in_flight.lock().await.clear();
     }
 
-    /// Number of pending batches (diagnostics).
-    pub fn pending_count(&self) -> usize {
-        self.queue.len()
+    /// Delete the durable entries identified by `keys` (Origin ack path).
+    pub async fn ack_keys(&self, keys: &[Vec<u8>]) -> Result<(), LiteError> {
+        self.queue.ack_keys(keys).await
+    }
+
+    /// Update the durable payload for `key` with the new seq stamped into `batch`.
+    pub async fn update_entry(
+        &self,
+        key: &[u8],
+        batch: &PendingColumnarBatch,
+    ) -> Result<(), LiteError> {
+        let payload = zerompk::to_msgpack_vec(batch).map_err(|e| LiteError::Serialization {
+            detail: format!("columnar outbound update encode: {e}"),
+        })?;
+        self.queue.update_entry(key, &payload).await
+    }
+
+    /// Number of pending entries in durable storage.
+    pub async fn len(&self) -> Result<u64, LiteError> {
+        self.queue.len().await
+    }
+
+    /// Returns `true` if no pending entries remain.
+    pub async fn is_empty(&self) -> Result<bool, LiteError> {
+        self.queue.is_empty().await
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::storage::pagedb_storage::PagedbStorageMem;
 
-    #[test]
-    fn enqueue_and_drain() {
-        let q = ColumnarOutbound::new();
-        q.enqueue_row("metrics", vec![Value::Integer(1)], Vec::new());
-        q.enqueue_row("metrics", vec![Value::Integer(2)], Vec::new());
-
-        let batches = q.drain_pending();
-        assert_eq!(batches.len(), 1);
-        assert_eq!(batches[0].collection, "metrics");
-        assert_eq!(batches[0].rows.len(), 2);
-        assert!(q.drain_pending().is_empty());
+    async fn make_queue() -> ColumnarOutbound<PagedbStorageMem> {
+        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
+        ColumnarOutbound::open(storage).await.unwrap()
     }
 
-    #[test]
-    fn separate_collections_separate_batches() {
-        let q = ColumnarOutbound::new();
-        q.enqueue_row("a", vec![Value::Integer(1)], Vec::new());
-        q.enqueue_row("b", vec![Value::Integer(2)], Vec::new());
+    #[tokio::test]
+    async fn enqueue_and_drain() {
+        let q = make_queue().await;
+        q.enqueue_row("metrics", vec![Value::Integer(1)], Vec::new())
+            .await
+            .unwrap();
+        q.enqueue_row("metrics", vec![Value::Integer(2)], Vec::new())
+            .await
+            .unwrap();
 
-        let batches = q.drain_pending();
-        assert_eq!(batches.len(), 2);
+        let pairs = q.drain_batch(usize::MAX).await.unwrap();
+        assert_eq!(pairs.len(), 2);
+        assert_eq!(pairs[0].1.collection, "metrics");
+        assert_eq!(pairs[0].1.rows[0][0], Value::Integer(1));
+        assert_eq!(pairs[1].1.rows[0][0], Value::Integer(2));
     }
 
-    #[test]
-    fn acknowledge_removes_batch() {
-        let q = ColumnarOutbound::new();
-        q.enqueue_row("m", vec![Value::Integer(1)], Vec::new());
-        let batches = q.drain_pending();
-        let id = batches[0].batch_id;
-        q.acknowledge_batch(id);
-        assert!(q.drain_pending().is_empty());
-    }
+    #[tokio::test]
+    async fn ack_keys_removes_entries() {
+        let q = make_queue().await;
+        q.enqueue_row("m", vec![Value::Integer(1)], Vec::new())
+            .await
+            .unwrap();
+        q.enqueue_row("m", vec![Value::Integer(2)], Vec::new())
+            .await
+            .unwrap();
+
+        let pairs = q.drain_batch(1).await.unwrap();
+        assert_eq!(pairs.len(), 1);
+        let keys: Vec<Vec<u8>> = pairs.into_iter().map(|(k, _)| k).collect();
+        q.ack_keys(&keys).await.unwrap();
 
-    #[test]
-    fn requeue_retries_on_next_drain() {
-        let q = ColumnarOutbound::new();
-        q.enqueue_row("m", vec![Value::Integer(1)], Vec::new());
-        let batches = q.drain_pending();
-        q.requeue_batch(batches.into_iter().next().unwrap());
+        assert_eq!(q.len().await.unwrap(), 1);
+    }
 
-        let retried = q.drain_pending();
-        assert_eq!(retried.len(), 1);
-        assert_eq!(retried[0].rows.len(), 1);
+    #[tokio::test]
+    async fn cap_returns_backpressure() {
+        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
+        let q = ColumnarOutbound::open_with_cap(storage, 2).await.unwrap();
+        q.enqueue_row("m", vec![Value::Integer(1)], Vec::new())
+            .await
+            .unwrap();
+        q.enqueue_row("m", vec![Value::Integer(2)], Vec::new())
+            .await
+            .unwrap();
+        let err = q
+            .enqueue_row("m", vec![Value::Integer(3)], Vec::new())
+            .await
+            .unwrap_err();
+        assert!(matches!(err, LiteError::Backpressure { .. }));
     }
 }