fix(crdt): stop silent loss from pending and colliding CRDT writes

Loro can accept an import whose causal predecessors are missing and
buffer its operations as pending, leaving the applied state untouched
while reporting success; every apply path along the sync pipeline
(Raft-committed apply, peer validate-and-apply, snapshot restore,
rollback) now treats that as an explicit refusal instead of a no-op, so
callers never advance a high-water-mark past a write that was never
applied. The sync session and dispatch layers propagate this as a
retryable Gap/Accepted status so the client re-pushes rather than
losing the row.

Separately, every collection previously shared the tenant's raw peer
id, so Loro minted identical operation ids for unrelated writes in
different collections; merging two of a tenant's collections into one
document (as an embedded peer does) silently dropped one side of the
collision. Collection state now derives its Loro peer id from
`(base_peer_id, collection)`.

Also splits `tenant_state::core` into `apply`, `constraints`, `rows`,
and `snapshot_io` modules along these behavioural seams.

Author: farhan-syah

Cargo.lock

@@ -35,6 +35,16 @@ pub enum CrdtError {
     #[error("CRDT import has too many operations: {actual} > {limit}")]
     ImportOperationLimitExceeded { limit: usize, actual: usize },
 
+    /// The imported update depends on operations this document has never seen,
+    /// so Loro buffered them as causally pending instead of applying them.
+    ///
+    /// The document state did NOT advance. Reporting such an import as success
+    /// is silent data loss: the caller acknowledges a write that was never
+    /// applied and may never be, since the missing predecessors are not part of
+    /// this document's operation history.
+    #[error("CRDT import depends on operations absent from this document")]
+    ImportPendingDependencies,
+
     /// A delta preview exceeded its encoded byte limit before import.
     #[error("delta preview exceeds byte limit: {actual} > {limit}")]
     PreviewDeltaTooLarge { limit: usize, actual: usize },

nodedb-crdt/src/error.rs

@@ -26,11 +26,22 @@ impl CrdtState {
     ///
     /// The limits are checked before Loro can import the blob, so rejected
     /// bytes never mutate this document or allocate an import graph.
+    ///
+    /// An update whose causal predecessors are absent from this document is
+    /// buffered by Loro as *pending* and leaves the applied state untouched.
+    /// That is reported as [`CrdtError::ImportPendingDependencies`], never as
+    /// success: a caller that took `Ok` here would acknowledge a write that
+    /// was never applied. The buffered operations remain queued inside Loro,
+    /// so a later import carrying the missing predecessors still converges.
     pub fn import_with_limits(&self, data: &[u8], limits: CrdtImportLimits) -> Result<()> {
         admit_import(data, &self.doc.oplog_vv(), limits)?;
-        self.doc
+        let status = self
+            .doc
             .import(data)
             .map_err(|e| CrdtError::DeltaApplyFailed(e.to_string()))?;
+        if status.pending.is_some() {
+            return Err(CrdtError::ImportPendingDependencies);
+        }
         Ok(())
     }
 
@@ -197,6 +208,90 @@ mod tests {
         assert_eq!(receiver.frontier(), before);
     }
 
+    /// Loro accepts a delta whose causal predecessors are missing and buffers
+    /// its operations as *pending*: the import call succeeds, but the applied
+    /// state never advances and the row is never written.
+    ///
+    /// Reporting that as a plain success is what makes the loss silent — the
+    /// caller advances its high-water-mark and acknowledges a write that does
+    /// not exist. An import must never report success while leaving the row it
+    /// carried unwritten.
+    #[test]
+    fn import_does_not_report_success_while_operations_stay_pending() {
+        // One source document, three writes, each exported as an incremental
+        // delta. The middle delta is withheld, so the third depends on
+        // operations the receiver never sees.
+        let source = CrdtState::new(9).expect("source");
+
+        let v0 = source.oplog_version_vector();
+        source
+            .upsert("docs", "first", &[("value", LoroValue::I64(1))])
+            .expect("first write");
+        let delta_first = source.export_updates_since(&v0).expect("first delta");
+
+        source
+            .upsert("docs", "withheld", &[("value", LoroValue::I64(2))])
+            .expect("withheld write");
+
+        let v2 = source.oplog_version_vector();
+        source
+            .upsert("docs", "third", &[("value", LoroValue::I64(3))])
+            .expect("third write");
+        let delta_third = source.export_updates_since(&v2).expect("third delta");
+
+        let receiver = CrdtState::new(1).expect("receiver");
+        receiver.import(&delta_first).expect("first delta applies");
+        assert!(receiver.row_exists("docs", "first"));
+
+        let result = receiver.import(&delta_third);
+
+        // Fix-shape agnostic: the import may refuse the delta outright, or it
+        // may report success — but it MUST NOT do the latter while the row it
+        // carried is absent.
+        assert!(
+            result.is_err() || receiver.row_exists("docs", "third"),
+            "import reported success while its operations stayed causally \
+             pending and the row was never written"
+        );
+    }
+
+    /// The pending-operation buffer must not be observable as a silent
+    /// frontier stall either: if an import reports success, the applied state
+    /// must have advanced past the pre-import frontier.
+    #[test]
+    fn successful_import_advances_the_applied_frontier() {
+        let source = CrdtState::new(21).expect("source");
+
+        let v0 = source.oplog_version_vector();
+        source
+            .upsert("docs", "a", &[("value", LoroValue::I64(1))])
+            .expect("first write");
+        let delta_a = source.export_updates_since(&v0).expect("delta a");
+
+        source
+            .upsert("docs", "gap", &[("value", LoroValue::I64(2))])
+            .expect("withheld write");
+
+        let v2 = source.oplog_version_vector();
+        source
+            .upsert("docs", "b", &[("value", LoroValue::I64(3))])
+            .expect("second write");
+        let delta_b = source.export_updates_since(&v2).expect("delta b");
+
+        let receiver = CrdtState::new(2).expect("receiver");
+        receiver.import(&delta_a).expect("delta a applies");
+        let before = receiver.frontier();
+
+        if receiver.import(&delta_b).is_ok() {
+            assert_ne!(
+                receiver.frontier(),
+                before,
+                "a successful import left the applied frontier unchanged — its \
+                 operations were buffered as pending, not applied"
+            );
+        }
+    }
+
     #[test]
     fn bounded_import_accepts_valid_delta_and_snapshot() {
         let delta = source_delta();

nodedb-crdt/src/state/snapshot.rs

@@ -30,4 +30,11 @@ pub enum AckStatus {
     /// A gap in the sequence was detected; `expected` is the next seq
     /// the server expected from this producer.
     Gap { expected: u64 },
+    /// The message passed envelope validation and was admitted for
+    /// processing, but has NOT been applied yet.
+    ///
+    /// Emitted by the session boundary before the durable apply runs. A sender
+    /// must keep the message queued until a terminal status arrives — treating
+    /// this as success would retire a write that may still be refused.
+    Accepted,
 }

nodedb-types/src/sync/wire/ack_status.rs

@@ -198,21 +198,34 @@ pub(crate) async fn apply_delta_and_finalize(
             // On success, rebuild the DeltaAck with the correct applied_seq and status.
             // The original ack_frame carries mutation_id and clock_skew_warning_ms which
             // we preserve; applied_seq and status come from the gate result.
-            let gate_result = match zerompk::from_msgpack::<nodedb_types::sync::wire::SyncAckResult>(
-                &payload,
-            ) {
-                Ok(r) => r,
-                Err(err) => {
-                    // Payload decode failed: fall back to the original ack frame so
-                    // the client still gets an ack (the delta was applied).
-                    warn!(
-                        collection = %delta_msg.collection,
-                        error = %err,
-                        "sync: failed to decode SyncAckResult from Data Plane; using default ack"
-                    );
-                    return Some(ack_frame);
-                }
-            };
+            let gate_result =
+                match zerompk::from_msgpack::<nodedb_types::sync::wire::SyncAckResult>(&payload) {
+                    Ok(r) => r,
+                    Err(err) => {
+                        // The Data Plane's outcome is unreadable, so whether the
+                        // delta applied is unknown. Returning the provisional ack
+                        // would assert success we cannot verify and let the client
+                        // retire the write. Refuse retryably instead: a re-push is
+                        // idempotent (the gate dedups by seq), whereas a wrongly
+                        // acked write is lost.
+                        warn!(
+                            collection = %delta_msg.collection,
+                            doc = %delta_msg.document_id,
+                            error = %err,
+                            "sync: failed to decode SyncAckResult from Data Plane; \
+                             refusing the delta rather than acking an unverified apply"
+                        );
+                        let reject = DeltaRejectMsg {
+                            mutation_id: delta_msg.mutation_id,
+                            reason: format!("apply outcome undecodable: {err}"),
+                            compensation: Some(CompensationHint::Custom {
+                                constraint: "apply_outcome_unreadable".into(),
+                                detail: "retry: the apply result could not be read".into(),
+                            }),
+                        };
+                        return SyncFrame::try_encode(SyncMessageType::DeltaReject, &reject);
+                    }
+                };
 
             // Applied-then-rejected: the delta committed and imported, but the
             // post-import validator flagged a constraint violation and enqueued a

nodedb/src/control/server/sync/async_dispatch/delta.rs

@@ -165,17 +165,54 @@ impl SyncSession {
             );
         }
 
+        // Provisional: the delta has passed envelope validation and been
+        // admitted, but the durable apply has not run yet and may still refuse
+        // it. Claiming `Applied` here would let a sender retire a write that
+        // never landed, so this reports `Accepted` and the caller overwrites
+        // the status with the real outcome once the Data Plane replies.
         let ack = DeltaAckMsg {
             mutation_id: msg.mutation_id,
             lsn: 0,
             clock_skew_warning_ms,
             applied_seq: 0,
-            status: nodedb_types::sync::wire::AckStatus::Applied,
+            status: nodedb_types::sync::wire::AckStatus::Accepted,
         };
         SyncFrame::try_encode(SyncMessageType::DeltaAck, &ack)
     }
 }
 
+impl SyncSession {
+    /// Record the terminal outcome of a delta whose refusal or success was
+    /// decided downstream of [`Self::handle_delta_push`].
+    ///
+    /// The durable apply runs outside the session, so without this the session
+    /// counters can only ever see admission — which is how a session that
+    /// applied one write out of hundreds still closed reporting
+    /// `rejected=0`. Every terminal frame returned to the client is routed
+    /// through here so the close line reflects what actually happened.
+    pub fn record_delta_outcome(&mut self, frame: &SyncFrame) {
+        match frame.msg_type {
+            SyncMessageType::DeltaReject => {
+                self.mutations_rejected += 1;
+            }
+            SyncMessageType::DeltaAck => {
+                let Some(ack) = frame.decode_body::<DeltaAckMsg>() else {
+                    // An ack we cannot read is not evidence of an apply.
+                    self.mutations_not_applied += 1;
+                    return;
+                };
+                match ack.status {
+                    AckStatus::Applied | AckStatus::Duplicate => self.mutations_applied += 1,
+                    AckStatus::Accepted | AckStatus::Fenced | AckStatus::Gap { .. } => {
+                        self.mutations_not_applied += 1
+                    }
+                }
+            }
+            _ => {}
+        }
+    }
+}
+
 /// Return `Some(skew_ms)` when the device-reported valid-time deviates
 /// from the Origin wall clock by more than 24 hours. Returning `None`
 /// inside tolerance keeps the ack payload small on the common path.

nodedb/src/control/server/sync/session/delta.rs

@@ -29,10 +29,22 @@ pub struct SyncSession {
     pub server_clock: HashMap<String, u64>,
     /// Subscribed shape IDs.
     pub subscribed_shapes: Vec<String>,
-    /// Mutations processed in this session.
+    /// Mutations admitted by envelope validation in this session.
+    ///
+    /// Admission is provisional — it says the frame was well-formed and
+    /// authorized, NOT that it was applied. Compare against
+    /// [`Self::mutations_applied`] to see how many actually landed.
     pub mutations_processed: u64,
-    /// Mutations rejected in this session.
+    /// Mutations the durable apply confirmed as applied.
+    pub mutations_applied: u64,
+    /// Mutations refused after admission — by the constraint validator, the
+    /// Data Plane, or authorization. Counted wherever the refusal is decided,
+    /// including refusals raised downstream of this session's provisional ack.
     pub mutations_rejected: u64,
+    /// Mutations neither applied nor permanently refused: retryable refusals,
+    /// sequence gaps, and fenced epochs. The sender is expected to re-push
+    /// these, so they are neither successes nor dead letters.
+    pub mutations_not_applied: u64,
     /// Mutations silently dropped (security rejections).
     pub mutations_silent_dropped: u64,
     /// Last activity timestamp.
@@ -84,6 +96,8 @@ impl SyncSession {
             server_clock: HashMap::new(),
             subscribed_shapes: Vec::new(),
             mutations_processed: 0,
+            mutations_applied: 0,
+            mutations_not_applied: 0,
             mutations_rejected: 0,
             mutations_silent_dropped: 0,
             last_activity: now,

nodedb/src/control/server/sync/session/state.rs

@@ -375,3 +375,42 @@ fn crc32c_mismatch_rejects_delta() {
     assert_eq!(r2.unwrap().msg_type, SyncMessageType::DeltaReject);
     assert_eq!(session.mutations_rejected, 1);
 }
+
+/// The session emits its `DeltaAck` before the delta has been dispatched to
+/// the Data Plane, and stamps it `AckStatus::Applied`. Nothing has been
+/// applied at that point — the durable apply happens afterwards, and may be
+/// refused. If the connection drops in that window, or the caller forwards the
+/// provisional frame, the client records a write that does not exist.
+///
+/// An acknowledgement must never claim `Applied` before an apply has occurred.
+#[test]
+fn provisional_delta_ack_does_not_claim_applied() {
+    let mut session = make_authenticated_session();
+
+    let data = serde_json::json!({"status": "active"});
+    let msg = DeltaPushMsg {
+        collection: "orders".into(),
+        document_id: "o1".into(),
+        delta: nodedb_types::json_to_msgpack(&data).unwrap(),
+        peer_id: 1,
+        mutation_id: 42,
+        checksum: 0,
+        device_valid_time_ms: None,
+        producer_id: 0,
+        epoch: 0,
+        seq: 0,
+    };
+
+    let frame = session
+        .handle_delta_push(&msg, None, None, None)
+        .expect("an accepted push returns a frame");
+    assert_eq!(frame.msg_type, SyncMessageType::DeltaAck);
+
+    let ack: DeltaAckMsg = frame.decode_body().expect("ack body decodes");
+    assert_ne!(
+        ack.status,
+        nodedb_types::sync::wire::AckStatus::Applied,
+        "the session acknowledged a delta as Applied before it was dispatched \
+         to the Data Plane, let alone applied"
+    );
+}

nodedb/src/control/server/sync/session/tests.rs

@@ -295,10 +295,18 @@ pub(in crate::control::server::sync) async fn handle_sync_session(
                             Some(response)
                         };
 
-                        if let Some(r) = final_response
-                            && ws.send(Message::Binary(r.to_bytes().into())).await.is_err()
-                        {
-                            break;
+                        if let Some(r) = final_response {
+                            // Account for the terminal outcome before the frame
+                            // leaves: refusals decided downstream of the
+                            // session's provisional ack are invisible to the
+                            // session otherwise, which is what let a lossy
+                            // session close reporting zero rejections.
+                            if delta_msg.is_some() {
+                                session.record_delta_outcome(&r);
+                            }
+                            if ws.send(Message::Binary(r.to_bytes().into())).await.is_err() {
+                                break;
+                            }
                         }
                     }
                 }
@@ -491,8 +499,10 @@ pub(in crate::control::server::sync) async fn handle_sync_session(
 
     info!(
         session = %session_id,
-        mutations = session.mutations_processed,
+        admitted = session.mutations_processed,
+        applied = session.mutations_applied,
         rejected = session.mutations_rejected,
+        not_applied = session.mutations_not_applied,
         silent_dropped = session.mutations_silent_dropped,
         uptime_secs = session.uptime_secs(),
         "sync: session closed"