feat(sync/array): add ack sender, catchup persistence, and last-applied tracking

Add ArrayAckSender — a periodic Tokio task that sends ArrayAckMsg frames
to Origin so its GC frontier can advance. The task is driven by a
configurable interval and stopped via a oneshot channel on disconnect.

Extend CatchupTracker with a persistent catchup-needed flag: when Origin
sends ArrayRejectReason::RetentionFloor the flag is written to
Namespace::Meta and survives restarts, ensuring the next connect
triggers a full snapshot catch-up. Expose should_request_catchup,
record_reject_retention_floor, clear_catchup_needed, and
arrays_needing_catchup.

Extend LiteApplyEngine with a per-array last-applied HLC cache persisted
under Namespace::Meta. Advances on every successful apply_write,
apply_delete, and apply_erase call; restored from storage on startup.
Expose last_applied_hlc for use by the ack sender.

Add SchemaRegistry::list_arrays to enumerate registered arrays, required
by the ack sender to iterate over known collections.

Wire CatchupTracker into ArrayInbound::new (dispatcher and test fixtures)
and fix initialization order in NodeDbLite::open so CatchupTracker is
constructed before LiteApplyEngine.

Author: farhan-syah

nodedb-lite/src/nodedb/core.rs

@@ -313,6 +313,10 @@ impl<S: StorageEngine + StorageEngineSync> NodeDbLite<S> {
         ));
 
         // ── Array CRDT inbound receive path ───────────────────────────────────
+        let array_catchup = Arc::new(
+            crate::sync::array::CatchupTracker::load(Arc::clone(&storage))
+                .map_err(NodeDbError::storage)?,
+        );
         let array_apply_engine = Arc::new(crate::sync::array::LiteApplyEngine::new(
             Arc::clone(&storage),
             Arc::clone(&array_state),
@@ -325,11 +329,8 @@ impl<S: StorageEngine + StorageEngineSync> NodeDbLite<S> {
             Arc::clone(&array_replica),
             Arc::clone(array_outbound.pending()),
             Arc::clone(array_outbound.op_log()),
+            Arc::clone(&array_catchup),
         ));
-        let array_catchup = Arc::new(
-            crate::sync::array::CatchupTracker::load(Arc::clone(&storage))
-                .map_err(NodeDbError::storage)?,
-        );
 
         let db = Self {
             storage,

nodedb-lite/src/sync/array/ack_sender.rs

@@ -0,0 +1,109 @@
+//! [`ArrayAckSender`] — periodic ack sender from Lite to Origin.
+//!
+//! A Tokio task (spawned on sync-session start, shut down on disconnect)
+//! that runs every `interval` seconds and sends `ArrayAckMsg` for each known
+//! array to Origin. Origin merges these into its `ArrayAckRegistry` to advance
+//! the GC frontier.
+//!
+//! # What gets acked
+//!
+//! For each registered array the ack carries `last_applied_hlc` — the highest
+//! HLC successfully applied by `LiteApplyEngine`. This is the correct value
+//! because it represents "I have all ops up to this HLC" from Origin's
+//! perspective.
+//!
+//! # Transport
+//!
+//! Frames are sent via a `tokio::sync::mpsc::Sender<Vec<u8>>` that the
+//! session's WebSocket write loop drains. This matches the pattern used by
+//! other Lite sync subsystems.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use nodedb_array::sync::replica_id::ReplicaId;
+use nodedb_types::sync::wire::SyncMessageType;
+use nodedb_types::sync::wire::array::ArrayAckMsg;
+use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
+use tracing::warn;
+
+use crate::storage::engine::StorageEngineSync;
+
+use super::inbound::apply::LiteApplyEngine;
+use super::schema_registry::SchemaRegistry;
+
+/// Default ack interval: 30 seconds.
+pub const DEFAULT_ACK_INTERVAL: Duration = Duration::from_secs(30);
+
+/// Periodic array ack sender task.
+///
+/// Spawned on session connect and cancelled on disconnect. The returned
+/// `JoinHandle` should be stored by the session and aborted (via
+/// `handle.abort()`) on session teardown.
+pub fn spawn<S: StorageEngineSync + Send + Sync + 'static>(
+    schemas: Arc<SchemaRegistry<S>>,
+    engine: Arc<LiteApplyEngine<S>>,
+    replica_id: ReplicaId,
+    tx: mpsc::Sender<Vec<u8>>,
+    interval: Duration,
+    mut stop: tokio::sync::oneshot::Receiver<()>,
+) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(interval);
+        // Skip the immediate first tick so we don't ack before applying anything.
+        ticker.tick().await;
+
+        loop {
+            tokio::select! {
+                _ = ticker.tick() => {
+                    send_acks(&schemas, &engine, replica_id, &tx).await;
+                }
+                _ = &mut stop => {
+                    return;
+                }
+            }
+        }
+    })
+}
+
+/// Build and send `ArrayAckMsg` frames for all known arrays.
+async fn send_acks<S: StorageEngineSync>(
+    schemas: &SchemaRegistry<S>,
+    engine: &LiteApplyEngine<S>,
+    replica_id: ReplicaId,
+    tx: &mpsc::Sender<Vec<u8>>,
+) {
+    let arrays = schemas.list_arrays();
+    for array in &arrays {
+        let Some(ack_hlc) = engine.last_applied_hlc(array) else {
+            // No ops applied for this array yet — nothing to ack.
+            continue;
+        };
+
+        let msg = ArrayAckMsg {
+            array: array.clone(),
+            replica_id: replica_id.as_u64(),
+            ack_hlc_bytes: ack_hlc.to_bytes(),
+        };
+
+        let frame = match nodedb_types::sync::wire::SyncFrame::try_encode(
+            SyncMessageType::ArrayAck,
+            &msg,
+        ) {
+            Some(f) => f.to_bytes(),
+            None => {
+                warn!(
+                    array = %array,
+                    "ack_sender: SyncFrame encode failed for ArrayAck"
+                );
+                continue;
+            }
+        };
+
+        if tx.send(frame).await.is_err() {
+            // Transport closed — session is shutting down.
+            return;
+        }
+    }
+}

nodedb-lite/src/sync/array/catchup.rs

@@ -21,21 +21,31 @@ use crate::storage::engine::StorageEngineSync;
 /// Storage key prefix for per-array last-seen HLC entries.
 const LAST_SEEN_PREFIX: &str = "array.last_seen_hlc:";
 
+/// Storage key prefix for per-array catchup-needed flags.
+const CATCHUP_NEEDED_PREFIX: &str = "array.catchup_needed:";
+
 /// Storage key for a named array's last-seen HLC.
 fn last_seen_key(array: &str) -> Vec<u8> {
     format!("{LAST_SEEN_PREFIX}{array}").into_bytes()
 }
 
+/// Storage key for a named array's catchup-needed flag.
+fn catchup_needed_key(array: &str) -> Vec<u8> {
+    format!("{CATCHUP_NEEDED_PREFIX}{array}").into_bytes()
+}
+
 /// Tracks the last successfully applied inbound HLC per array.
 ///
-/// Used by the transport layer (future phases) to populate catchup-request
-/// messages after reconnect or when Origin's log GC horizon advances past the
-/// local log.
+/// Used by the transport layer to populate catchup-request messages after
+/// reconnect or when Origin's log GC horizon advances past the local log.
 ///
 /// Thread-safe via an internal [`Mutex`].
 pub struct CatchupTracker<S: StorageEngineSync> {
     storage: Arc<S>,
     state: Mutex<HashMap<String, Hlc>>,
+    /// Arrays that need a full catchup on next connect.
+    /// Set when Origin sends `ArrayRejectMsg::RetentionFloor`.
+    catchup_needed: Mutex<std::collections::HashSet<String>>,
 }
 
 impl<S: StorageEngineSync> CatchupTracker<S> {
@@ -67,9 +77,24 @@ impl<S: StorageEngineSync> CatchupTracker<S> {
             state.insert(name.to_owned(), Hlc::from_bytes(&hlc_arr));
         }
 
+        // Load catchup-needed flags.
+        let needed_prefix = CATCHUP_NEEDED_PREFIX.as_bytes();
+        let needed_pairs = storage.scan_range_sync(Namespace::Meta, needed_prefix, usize::MAX)?;
+        let mut catchup_needed = std::collections::HashSet::new();
+        for (key, _) in needed_pairs {
+            if !key.starts_with(needed_prefix) {
+                break;
+            }
+            let name_bytes = &key[needed_prefix.len()..];
+            if let Ok(name) = std::str::from_utf8(name_bytes) {
+                catchup_needed.insert(name.to_owned());
+            }
+        }
+
         Ok(Self {
             storage,
             state: Mutex::new(state),
+            catchup_needed: Mutex::new(catchup_needed),
         })
     }
 
@@ -110,6 +135,60 @@ impl<S: StorageEngineSync> CatchupTracker<S> {
             from_hlc_bytes: self.last_seen(array).to_bytes(),
         }
     }
+
+    /// Returns `true` if `array` should request a full catch-up on connect.
+    ///
+    /// True when:
+    /// - No `last_seen_hlc` is recorded (first connect), OR
+    /// - `record_reject_retention_floor` was called (Origin GC advanced past our log).
+    pub fn should_request_catchup(&self, array: &str, _current_local_hlc: Hlc) -> bool {
+        let needs_catchup = self
+            .catchup_needed
+            .lock()
+            .ok()
+            .map(|s| s.contains(array))
+            .unwrap_or(false);
+        if needs_catchup {
+            return true;
+        }
+        // First connect: no last-seen HLC recorded.
+        let state = self.state.lock().ok();
+        state.map(|s| !s.contains_key(array)).unwrap_or(false)
+    }
+
+    /// Mark `array` as needing a full catch-up on the next connect.
+    ///
+    /// Called when Origin sends `ArrayRejectMsg::RetentionFloor`.
+    /// Persisted under `Namespace::Meta` `"array.catchup_needed:{array}"`.
+    pub fn record_reject_retention_floor(&self, array: &str) -> Result<(), LiteError> {
+        if let Ok(mut needed) = self.catchup_needed.lock() {
+            needed.insert(array.to_owned());
+        }
+        // Persist flag (value = 1 byte sentinel).
+        self.storage
+            .put_sync(Namespace::Meta, &catchup_needed_key(array), &[1u8])
+    }
+
+    /// Clear the catchup-needed flag for `array` after a successful catch-up.
+    ///
+    /// Called after the snapshot stream has been fully applied.
+    pub fn clear_catchup_needed(&self, array: &str) -> Result<(), LiteError> {
+        if let Ok(mut needed) = self.catchup_needed.lock() {
+            needed.remove(array);
+        }
+        // Remove the persisted flag.
+        self.storage
+            .delete_sync(Namespace::Meta, &catchup_needed_key(array))
+    }
+
+    /// Return all arrays that are marked as needing catch-up.
+    pub fn arrays_needing_catchup(&self) -> Vec<String> {
+        self.catchup_needed
+            .lock()
+            .ok()
+            .map(|s| s.iter().cloned().collect())
+            .unwrap_or_default()
+    }
 }
 
 // ─── Tests ────────────────────────────────────────────────────────────────────

nodedb-lite/src/sync/array/inbound/apply.rs

@@ -2,6 +2,7 @@
 //! [`ApplyEngine`] trait so [`nodedb_array::sync::apply::apply_op`] can drive
 //! local state from inbound wire messages.
 
+use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
 
 use nodedb_array::error::ArrayError;
@@ -10,12 +11,16 @@ use nodedb_array::sync::hlc::Hlc;
 use nodedb_array::sync::op::ArrayOp;
 use nodedb_array::sync::op_log::OpLog;
 use nodedb_array::types::coord::value::CoordValue;
+use nodedb_types::Namespace;
 
 use crate::engine::array::engine::ArrayEngineState;
 use crate::storage::engine::StorageEngineSync;
 use crate::sync::array::op_log_redb::RedbOpLog;
 use crate::sync::array::schema_registry::SchemaRegistry;
 
+/// Key prefix for `last_applied_hlc` entries persisted under `Namespace::Meta`.
+const LAST_APPLIED_PREFIX: &str = "array.last_applied:";
+
 /// Adapts NodeDB-Lite's array engine state to the [`ApplyEngine`] trait.
 ///
 /// All fields are `Arc`-wrapped so their interior mutability can satisfy the
@@ -35,6 +40,9 @@ pub struct LiteApplyEngine<S: StorageEngineSync> {
     pub(super) array_state: Arc<Mutex<ArrayEngineState>>,
     pub(super) schemas: Arc<SchemaRegistry<S>>,
     pub(super) op_log: Arc<RedbOpLog<S>>,
+    /// In-memory cache of the last applied HLC per array.
+    /// Persisted under `Namespace::Meta` `"array.last_applied:{name}"`.
+    last_applied: Mutex<HashMap<String, Hlc>>,
 }
 
 impl<S: StorageEngineSync> LiteApplyEngine<S> {
@@ -45,11 +53,69 @@ impl<S: StorageEngineSync> LiteApplyEngine<S> {
         schemas: Arc<SchemaRegistry<S>>,
         op_log: Arc<RedbOpLog<S>>,
     ) -> Self {
+        // Restore any persisted last-applied HLCs from storage on startup.
+        let last_applied = Self::load_last_applied(&storage);
         Self {
             storage,
             array_state,
             schemas,
             op_log,
+            last_applied: Mutex::new(last_applied),
+        }
+    }
+
+    fn load_last_applied(storage: &Arc<S>) -> HashMap<String, Hlc> {
+        let prefix = LAST_APPLIED_PREFIX.as_bytes();
+        let Ok(pairs) = storage.scan_range_sync(Namespace::Meta, prefix, usize::MAX) else {
+            return HashMap::new();
+        };
+        let mut map = HashMap::new();
+        for (key, value) in pairs {
+            if !key.starts_with(prefix) {
+                break;
+            }
+            let name_bytes = &key[prefix.len()..];
+            let Ok(name) = std::str::from_utf8(name_bytes) else {
+                continue;
+            };
+            if value.len() == 18 {
+                let bytes: [u8; 18] = match value.try_into() {
+                    Ok(b) => b,
+                    Err(_) => continue,
+                };
+                map.insert(name.to_owned(), Hlc::from_bytes(&bytes));
+            }
+        }
+        map
+    }
+
+    /// Return the highest HLC applied for `array`, or `None` if no ops applied.
+    pub fn last_applied_hlc(&self, array: &str) -> Option<Hlc> {
+        self.last_applied.lock().ok()?.get(array).copied()
+    }
+
+    /// Record that `hlc` has been successfully applied for `array`.
+    ///
+    /// Advances the in-memory record and persists under `Namespace::Meta`.
+    fn record_applied_hlc(&self, array: &str, hlc: Hlc) {
+        let should_persist = {
+            let mut map = match self.last_applied.lock() {
+                Ok(m) => m,
+                Err(_) => return,
+            };
+            let current = map.get(array).copied().unwrap_or(Hlc::ZERO);
+            if hlc > current {
+                map.insert(array.to_owned(), hlc);
+                true
+            } else {
+                false
+            }
+        };
+        if should_persist {
+            let key = format!("{LAST_APPLIED_PREFIX}{array}").into_bytes();
+            let _ = self
+                .storage
+                .put_sync(Namespace::Meta, &key, &hlc.to_bytes());
         }
     }
 }
@@ -92,7 +158,9 @@ impl<S: StorageEngineSync> ApplyEngine for &LiteApplyEngine<S> {
             })?;
         // Record in op-log so that subsequent `already_seen` returns true.
         // `append` is idempotent on duplicate (array, hlc) pairs.
-        self.op_log.append(op)
+        self.op_log.append(op)?;
+        self.record_applied_hlc(&op.header.array, op.header.hlc);
+        Ok(())
     }
 
     fn apply_delete(&mut self, op: &ArrayOp) -> nodedb_array::error::ArrayResult<()> {
@@ -105,7 +173,9 @@ impl<S: StorageEngineSync> ApplyEngine for &LiteApplyEngine<S> {
             .map_err(|e| ArrayError::SegmentCorruption {
                 detail: format!("apply_delete: {e}"),
             })?;
-        self.op_log.append(op)
+        self.op_log.append(op)?;
+        self.record_applied_hlc(&op.header.array, op.header.hlc);
+        Ok(())
     }
 
     fn apply_erase(&mut self, op: &ArrayOp) -> nodedb_array::error::ArrayResult<()> {
@@ -123,7 +193,9 @@ impl<S: StorageEngineSync> ApplyEngine for &LiteApplyEngine<S> {
             .map_err(|e| ArrayError::SegmentCorruption {
                 detail: format!("apply_erase: {e}"),
             })?;
-        self.op_log.append(op)
+        self.op_log.append(op)?;
+        self.record_applied_hlc(&op.header.array, op.header.hlc);
+        Ok(())
     }
 
     /// No-op: the Lite array engine reads from the durable memtable + segments