fix(crdt): materialize synced deltas into sparse document store (closes #146)

nodedb/src/data/executor/handlers/control/crdt.rs

@@ -11,10 +11,14 @@ use nodedb_types::sync::wire::{AckStatus, SyncProvenance};
 
 use crate::bridge::envelope::{ErrorCode, Response};
 use crate::data::executor::sync_gate::SyncAdmit;
-use crate::engine::crdt::tenant_state::ValidatedApplyOutcome;
+use crate::engine::crdt::tenant_state::{TenantCrdtEngine, ValidatedApplyOutcome};
 
 use crate::data::executor::core_loop::CoreLoop;
+use crate::data::executor::doc_format::canonicalize_document_for_storage;
 use crate::data::executor::task::ExecutionTask;
+use crate::engine::document::crdt_store::loro_value_to_json;
+use crate::engine::document::store::surrogate_to_doc_id;
+use crate::engine::sparse::btree_versioned::VersionedPut;
 
 impl CoreLoop {
     pub(in crate::data::executor) fn execute_crdt_read(
@@ -225,33 +229,58 @@ impl CoreLoop {
             // Non-sync path (SQL / native client): validate + apply, no gate.
             // There is no client to reject here, so the validated outcome is
             // only observed for its DLQ side effect and logged.
-            let engine = match self.get_crdt_engine(tenant_id) {
-                Ok(e) => e,
-                Err(e) => {
-                    warn!(core = self.core_id, error = %e, "failed to create CRDT engine");
-                    return self.response_error(
-                        task,
-                        ErrorCode::Internal {
-                            detail: e.to_string(),
-                        },
-                    );
+            // Borrow the engine in a nested block so the &mut borrow is dropped
+            // before the sparse write below takes &self. On a Clean apply we
+            // read the merged row back and encode it while the borrow is live,
+            // carrying the materialized bytes out.
+            let materialized = {
+                let engine = match self.get_crdt_engine(tenant_id) {
+                    Ok(e) => e,
+                    Err(e) => {
+                        warn!(core = self.core_id, error = %e, "failed to create CRDT engine");
+                        return self.response_error(
+                            task,
+                            ErrorCode::Internal {
+                                detail: e.to_string(),
+                            },
+                        );
+                    }
+                };
+                let outcome = engine.apply_committed_delta_validated(
+                    collection,
+                    delta,
+                    surrogate,
+                    document_id,
+                    peer_id,
+                );
+                match outcome {
+                    ValidatedApplyOutcome::Clean => {
+                        if surrogate != Surrogate::ZERO {
+                            Self::encode_crdt_row(engine, collection, document_id)
+                        } else {
+                            None
+                        }
+                    }
+                    ValidatedApplyOutcome::Rejected(vt) => {
+                        debug!(core = self.core_id, %collection, reason = %vt, "crdt apply violated constraint (DLQ)");
+                        None
+                    }
+                    ValidatedApplyOutcome::Malformed => {
+                        warn!(core = self.core_id, %collection, "crdt apply skipped malformed delta");
+                        None
+                    }
                 }
             };
-            let outcome = engine.apply_committed_delta_validated(
-                collection,
-                delta,
-                surrogate,
-                document_id,
-                peer_id,
-            );
-            match outcome {
-                ValidatedApplyOutcome::Clean => {}
-                ValidatedApplyOutcome::Rejected(vt) => {
-                    debug!(core = self.core_id, %collection, reason = %vt, "crdt apply violated constraint (DLQ)");
-                }
-                ValidatedApplyOutcome::Malformed => {
-                    warn!(core = self.core_id, %collection, "crdt apply skipped malformed delta");
-                }
+            // engine borrow dropped here; materialize into the sparse document
+            // store so DocumentScan / ShapeSnapshot see the synced document.
+            if let Some(bytes) = materialized {
+                self.materialize_synced_document(
+                    task.request.database_id.as_u64(),
+                    tenant_id.as_u64(),
+                    collection,
+                    surrogate,
+                    &bytes,
+                );
             }
             self.checkpoint_coordinator.mark_dirty("crdt", 1);
             return self.response_ok(task);
@@ -284,7 +313,7 @@ impl CoreLoop {
                     Pending { installed: u64 },
                     Applied(ValidatedApplyOutcome),
                 }
-                let outcome = {
+                let (outcome, materialized) = {
                     let engine = match self.get_crdt_engine(tenant_id) {
                         Ok(e) => e,
                         Err(e) => {
@@ -299,19 +328,30 @@ impl CoreLoop {
                     };
                     let installed = engine.installed_constraint_version(collection);
                     if constraint_version_required > installed {
-                        GateOutcome::Pending { installed }
+                        (GateOutcome::Pending { installed }, None)
                     } else {
-                        GateOutcome::Applied(engine.apply_committed_delta_validated(
+                        let applied = engine.apply_committed_delta_validated(
                             collection,
                             delta,
                             surrogate,
                             document_id,
                             peer_id,
-                        ))
+                        );
+                        // On a Clean apply, read the merged row back and encode
+                        // it while the engine borrow is still live so the bytes
+                        // can be materialized into the sparse store below.
+                        let mat = if matches!(applied, ValidatedApplyOutcome::Clean)
+                            && surrogate != Surrogate::ZERO
+                        {
+                            Self::encode_crdt_row(engine, collection, document_id)
+                        } else {
+                            None
+                        };
+                        (GateOutcome::Applied(applied), mat)
                     }
                 };
                 // engine borrow is dropped here; mark_dirty / sync_commit take
-                // &mut self.
+                // &mut self, and the sparse materialize takes &self.
                 let reject = match outcome {
                     GateOutcome::Pending { installed } => {
                         // Create-race: the constraints this delta was admitted
@@ -348,6 +388,18 @@ impl CoreLoop {
                         None
                     }
                 };
+                // Materialize the merged document into the sparse store so
+                // DocumentScan / ShapeSnapshot see the synced write. `materialized`
+                // is Some only on a Clean apply with an assigned surrogate.
+                if let Some(bytes) = materialized {
+                    self.materialize_synced_document(
+                        task.request.database_id.as_u64(),
+                        tenant_id.as_u64(),
+                        collection,
+                        surrogate,
+                        &bytes,
+                    );
+                }
                 // Advance the HWM unconditionally after apply — a rejected,
                 // fenced, or malformed delta must not wedge the sync stream.
                 self.sync_commit(prov);
@@ -416,4 +468,76 @@ impl CoreLoop {
     ) -> u64 {
         *self.sync_hwm.get(&(producer_id, stream_id)).unwrap_or(&0)
     }
+
+    /// Read the merged Loro row back and encode it into the canonical
+    /// schemaless storage bytes the native put path writes.
+    ///
+    /// Called while the CRDT engine `&mut` borrow is still live (the borrow
+    /// checker forbids touching `self.sparse` here), so it is an associated
+    /// function over the borrowed engine rather than a method. Returns `None`
+    /// when the row is absent or cannot be converted — the caller then skips
+    /// the sparse write. A materialization miss must never fail the delta
+    /// apply: the Loro merge has already succeeded and the sync stream must
+    /// not wedge.
+    fn encode_crdt_row(
+        engine: &TenantCrdtEngine,
+        collection: &str,
+        document_id: &str,
+    ) -> Option<Vec<u8>> {
+        let loro_val = engine.read_row(collection, document_id)?;
+        let json = loro_value_to_json(&loro_val);
+        let msgpack = match nodedb_types::json_to_msgpack(&json) {
+            Ok(bytes) => bytes,
+            Err(_) => return None,
+        };
+        Some(canonicalize_document_for_storage(&msgpack))
+    }
+
+    /// Write the merged CRDT document into the sparse document store so
+    /// `DocumentScan` / `ShapeSnapshot` observe the synced write, matching the
+    /// key and bytes the native schemaless put path produces.
+    ///
+    /// The storage key is the hex-encoded surrogate (identical to the native
+    /// path), NOT the CRDT `document_id` (which is the user-facing Loro row
+    /// id). Bitemporal collections append a version per applied delta;
+    /// non-bitemporal collections overwrite by key (idempotent under replay).
+    /// FTS is intentionally NOT indexed here — the sync path delivers a
+    /// separate `FtsIndex` frame, and re-indexing would double-index. A write
+    /// failure is logged and swallowed so a materialization miss never wedges
+    /// the sync stream.
+    fn materialize_synced_document(
+        &self,
+        database_id: u64,
+        tid: u64,
+        collection: &str,
+        surrogate: Surrogate,
+        stored: &[u8],
+    ) {
+        let storage_key = surrogate_to_doc_id(surrogate);
+        let result = if self.is_bitemporal(tid, collection) {
+            self.sparse.versioned_put(VersionedPut {
+                database_id,
+                tenant: tid,
+                coll: collection,
+                doc_id: storage_key.as_str(),
+                sys_from_ms: self.bitemporal_now_ms(),
+                valid_from_ms: i64::MIN,
+                valid_until_ms: i64::MAX,
+                body: stored,
+            })
+        } else {
+            self.sparse
+                .put(database_id, tid, collection, storage_key.as_str(), stored)
+                .map(|_| ())
+        };
+        if let Err(e) = result {
+            warn!(
+                core = self.core_id,
+                %collection,
+                document_id = %storage_key,
+                error = %e,
+                "crdt sync materialize into sparse document store failed"
+            );
+        }
+    }
 }