feat(sync): add array CRDT sync module with outbound and inbound paths

Introduces the sync/array sub-module for replicating the sparse N-dimensional
array engine between Lite and Origin via CRDT op-log deltas.

Outbound path (Lite → Origin):
- RedbOpLog: redb-backed persistent op-log for array cell mutations
- PendingQueue: durable queue of unacknowledged outbound ops
- ArrayOutbound: coordinates emit_put / emit_delete after engine writes
- ReplicaState: stable replica UUID + HLC clock generator
- SchemaRegistry: persists Loro-encoded SchemaDoc snapshots per array

Inbound path (Origin → Lite):
- ArrayInbound: dispatches wire messages to type-specific handlers
- LiteApplyEngine: applies remote deltas to the local array engine
- CatchupTracker: tracks per-array last-seen HLC for catch-up requests

Wire the full send/receive state into NodeDbLite::new and emit ops from
array_put_cell / array_delete_cell after each successful engine write.

Author: farhan-syah

nodedb-lite/src/nodedb/array.rs

@@ -8,6 +8,7 @@ use nodedb_array::schema::ArraySchema;
 use nodedb_array::tile::cell_payload::CellPayload;
 use nodedb_array::types::cell_value::value::CellValue;
 use nodedb_array::types::coord::value::CoordValue;
+use nodedb_types::OPEN_UPPER;
 use nodedb_types::error::{NodeDbError, NodeDbResult};
 
 use crate::storage::engine::{StorageEngine, StorageEngineSync};
@@ -20,10 +21,20 @@ impl<S: StorageEngine + StorageEngineSync> NodeDbLite<S> {
     ///
     /// Returns an error if an array named `name` already exists.
     pub fn create_array(&self, name: &str, schema: ArraySchema) -> NodeDbResult<()> {
+        // Clone before the engine call to avoid a partial-move of `schema`.
+        let schema_for_crdt = schema.clone();
+
         self.array_state
             .lock_or_recover()
             .create_array(&self.storage, name, schema)
-            .map_err(NodeDbError::storage)
+            .map_err(NodeDbError::storage)?;
+
+        // Register the schema CRDT so subsequent emit_* calls can find schema_hlc.
+        self.array_schemas
+            .put_schema(name, &schema_for_crdt)
+            .map_err(NodeDbError::storage)?;
+
+        Ok(())
     }
 
     /// Write a cell into array `name` at `coord`.
@@ -40,6 +51,10 @@ impl<S: StorageEngine + StorageEngineSync> NodeDbLite<S> {
         valid_from_ms: i64,
         valid_until_ms: i64,
     ) -> NodeDbResult<()> {
+        // Clone before the engine call to keep copies for emit.
+        let coord_emit = coord.clone();
+        let attrs_emit = attrs.clone();
+
         self.array_state
             .lock_or_recover()
             .put_cell(
@@ -51,7 +66,27 @@ impl<S: StorageEngine + StorageEngineSync> NodeDbLite<S> {
                 valid_from_ms,
                 valid_until_ms,
             )
-            .map_err(NodeDbError::storage)
+            .map_err(NodeDbError::storage)?;
+
+        // Emit op after engine succeeds. If emit fails after a successful
+        // engine write, log the error and return it; the engine write is
+        // NOT rolled back — ack reconciliation in later phases will catch
+        // the gap.
+        if let Err(e) = self.array_outbound.emit_put(
+            name,
+            coord_emit,
+            attrs_emit,
+            valid_from_ms,
+            valid_until_ms,
+        ) {
+            tracing::error!(
+                array = name,
+                "array_put_cell: emit failed after engine write — op-log gap: {e}"
+            );
+            return Err(NodeDbError::storage(e));
+        }
+
+        Ok(())
     }
 
     /// Slice query: return all live cells whose coordinates fall within
@@ -94,10 +129,29 @@ impl<S: StorageEngine + StorageEngineSync> NodeDbLite<S> {
         coord: Vec<CoordValue>,
         system_from_ms: i64,
     ) -> NodeDbResult<()> {
+        // Clone before the engine call for emit.
+        let coord_emit = coord.clone();
+
         self.array_state
             .lock_or_recover()
             .delete_cell(name, coord, system_from_ms)
-            .map_err(NodeDbError::storage)
+            .map_err(NodeDbError::storage)?;
+
+        // valid_from_ms / valid_until_ms: the current API does not expose
+        // valid-time arguments on delete. Defaults of 0 / OPEN_UPPER are used
+        // here. Phase F will widen the API to carry the full bitemporal envelope.
+        if let Err(e) = self
+            .array_outbound
+            .emit_delete(name, coord_emit, 0, OPEN_UPPER)
+        {
+            tracing::error!(
+                array = name,
+                "array_delete_cell: emit failed after engine write — op-log gap: {e}"
+            );
+            return Err(NodeDbError::storage(e));
+        }
+
+        Ok(())
     }
 
     /// GDPR erasure: write the `0xFE` sentinel and flush to disk.
@@ -111,10 +165,28 @@ impl<S: StorageEngine + StorageEngineSync> NodeDbLite<S> {
         coord: Vec<CoordValue>,
         system_from_ms: i64,
     ) -> NodeDbResult<()> {
+        // Clone before the engine call for emit.
+        let coord_emit = coord.clone();
+
         self.array_state
             .lock_or_recover()
             .gdpr_erase_cell(&self.storage, name, coord, system_from_ms)
-            .map_err(NodeDbError::storage)
+            .map_err(NodeDbError::storage)?;
+
+        // valid_from_ms / valid_until_ms: same defaulting as array_delete_cell.
+        // Phase F will widen the API.
+        if let Err(e) = self
+            .array_outbound
+            .emit_erase(name, coord_emit, 0, OPEN_UPPER)
+        {
+            tracing::error!(
+                array = name,
+                "array_gdpr_erase_cell: emit failed after engine write — op-log gap: {e}"
+            );
+            return Err(NodeDbError::storage(e));
+        }
+
+        Ok(())
     }
 
     /// Flush any pending memtable data for `name` to a durable segment.

nodedb-lite/src/nodedb/core.rs

@@ -14,7 +14,7 @@ use crate::engine::htap::HtapBridge;
 use crate::engine::strict::StrictEngine;
 use crate::engine::vector::graph::{HnswIndex, HnswParams};
 use crate::memory::{EngineId, MemoryGovernor};
-use crate::storage::engine::{StorageEngine, WriteOp};
+use crate::storage::engine::{StorageEngine, StorageEngineSync, WriteOp};
 
 /// Storage key constants.
 pub(crate) const META_HNSW_COLLECTIONS: &[u8] = b"meta:hnsw_collections";
@@ -31,7 +31,7 @@ pub(crate) const META_LAST_FLUSHED_MID: &[u8] = b"meta:last_flushed_mid";
 ///
 /// Fully capable of vector search, graph traversal, and document CRUD
 /// entirely offline. Optional sync to Origin via WebSocket.
-pub struct NodeDbLite<S: StorageEngine> {
+pub struct NodeDbLite<S: StorageEngine + StorageEngineSync> {
     pub(crate) storage: Arc<S>,
     /// Per-collection HNSW indices.
     pub(crate) hnsw_indices: Mutex<HashMap<String, HnswIndex>>,
@@ -70,7 +70,25 @@ pub struct NodeDbLite<S: StorageEngine> {
     /// Arc-wrapped for sharing with the query engine's DDL handlers.
     pub(crate) timeseries: Arc<Mutex<crate::engine::timeseries::engine::TimeseriesEngine>>,
     /// Array engine in-memory state (storage-agnostic; calls via NodeDbLite methods).
-    pub(crate) array_state: std::sync::Mutex<crate::engine::array::engine::ArrayEngineState>,
+    ///
+    /// `Arc`-wrapped so it can be shared with [`crate::sync::array::LiteApplyEngine`]
+    /// for the inbound receive path without borrowing `NodeDbLite`.
+    pub(crate) array_state: Arc<std::sync::Mutex<crate::engine::array::engine::ArrayEngineState>>,
+    /// Stable per-replica identity + HLC generator for array CRDT sync.
+    /// Used by the transport-layer wiring (Phase F+); held here so that
+    /// the `NodeDbLite` constructor owns the lifetime.
+    #[allow(dead_code)]
+    pub(crate) array_replica: Arc<crate::sync::array::ReplicaState>,
+    /// Per-array [`SchemaDoc`] registry (persisted Loro snapshots).
+    pub(crate) array_schemas: Arc<crate::sync::array::SchemaRegistry<S>>,
+    /// Array CRDT send path: op-log + pending queue emitters.
+    pub(crate) array_outbound: Arc<crate::sync::array::ArrayOutbound<S>>,
+    /// Array CRDT receive path: applies inbound wire messages from Origin.
+    #[allow(dead_code)]
+    pub(crate) array_inbound: Arc<crate::sync::array::ArrayInbound<S>>,
+    /// Per-array last-seen HLC tracker for catch-up requests.
+    #[allow(dead_code)]
+    pub(crate) array_catchup: Arc<crate::sync::array::CatchupTracker<S>>,
     /// When `false`, KV operations go directly to redb, bypassing Loro.
     /// Other engines (vector, graph, document) are unaffected.
     pub(crate) sync_enabled: bool,
@@ -97,7 +115,7 @@ pub(crate) struct KvWriteBuffer {
     pub overlay: HashMap<Vec<u8>, Option<Vec<u8>>>,
 }
 
-impl<S: StorageEngine> NodeDbLite<S> {
+impl<S: StorageEngine + StorageEngineSync> NodeDbLite<S> {
     /// Open or create a Lite database backed by the given storage engine.
     ///
     /// Memory budget and per-engine percentages are resolved from environment
@@ -271,6 +289,47 @@ impl<S: StorageEngine> NodeDbLite<S> {
 
         let array_engine =
             crate::engine::array::ArrayEngineState::open(&storage).map_err(NodeDbError::storage)?;
+        let array_state = Arc::new(Mutex::new(array_engine));
+
+        // ── Array CRDT sync state ──────────────────────────────────────────────
+        let array_replica = Arc::new(
+            crate::sync::array::ReplicaState::load_or_init(&*storage)
+                .map_err(NodeDbError::storage)?,
+        );
+        let array_schemas = Arc::new(
+            crate::sync::array::SchemaRegistry::load(
+                Arc::clone(&storage),
+                Arc::clone(&array_replica),
+            )
+            .map_err(NodeDbError::storage)?,
+        );
+        let array_op_log = Arc::new(crate::sync::array::RedbOpLog::new(Arc::clone(&storage)));
+        let array_pending = Arc::new(crate::sync::array::PendingQueue::new(Arc::clone(&storage)));
+        let array_outbound = Arc::new(crate::sync::array::ArrayOutbound::new(
+            Arc::clone(&array_op_log),
+            Arc::clone(&array_pending),
+            Arc::clone(&array_schemas),
+            Arc::clone(&array_replica),
+        ));
+
+        // ── Array CRDT inbound receive path ───────────────────────────────────
+        let array_apply_engine = Arc::new(crate::sync::array::LiteApplyEngine::new(
+            Arc::clone(&storage),
+            Arc::clone(&array_state),
+            Arc::clone(&array_schemas),
+            Arc::clone(array_outbound.op_log()),
+        ));
+        let array_inbound = Arc::new(crate::sync::array::ArrayInbound::new(
+            array_apply_engine,
+            Arc::clone(&array_schemas),
+            Arc::clone(&array_replica),
+            Arc::clone(array_outbound.pending()),
+            Arc::clone(array_outbound.op_log()),
+        ));
+        let array_catchup = Arc::new(
+            crate::sync::array::CatchupTracker::load(Arc::clone(&storage))
+                .map_err(NodeDbError::storage)?,
+        );
 
         let db = Self {
             storage,
@@ -288,7 +347,12 @@ impl<S: StorageEngine> NodeDbLite<S> {
             columnar,
             htap,
             timeseries,
-            array_state: Mutex::new(array_engine),
+            array_state,
+            array_replica,
+            array_schemas,
+            array_outbound,
+            array_inbound,
+            array_catchup,
             sync_enabled,
             kv_write_buf: Mutex::new(KvWriteBuffer {
                 ops: Vec::with_capacity(1024),