feat(nodedb-crdt): add bitemporal CRDT sync with live-scoped constraints

farhan-syah · farhan-syah · commit f95e995a3214 · 2026-04-24T21:07:29.000+08:00
Bitemporal collections carry valid-time (`_ts_valid_from`/`_ts_valid_until`)
and system-time (`_ts_system`) metadata. Three behaviours had to change:

UNIQUE scoping — a UNIQUE collision between a superseded row (finite
`_ts_valid_until`) and a new live row is not a violation; they represent
the same logical entity at different valid-times. `constraint_checks` now
calls the new `field_value_exists_live` on bitemporal collections instead
of the global scan.

Receiver-stamped system time — `_ts_system` is always overwritten with
the receiving node's wall clock during `validate_and_apply`, regardless
of what the sender supplied. This keeps system-time receiver-authoritative
so convergence does not depend on clock agreement between peers.

New CrdtState helpers — `field_value_exists_live` and `live_row_ids`
filter out superseded rows using the `row_is_live` predicate
(`_ts_valid_until` absent, null, or `i64::MAX`).

TenantCrdtEngine surface — `mark_bitemporal`, `is_bitemporal`, and
`set_collection_policy_typed` allow in-process callers to configure
collections without going through DDL JSON.

Integration tests in `nodedb/tests/bitemporal_crdt_sync.rs` verify:
- superseded email does not collide with a new live row
- two live rows with the same UNIQUE value still collide
- sender-supplied `_ts_system` is overwritten by the receiver
diff --git a/nodedb-crdt/src/constraint_checks.rs b/nodedb-crdt/src/constraint_checks.rs
@@ -41,8 +41,16 @@ impl Validator {
 
         let value = &field_value.1;
 
-        // Check if this value already exists in another row.
-        if state.field_value_exists(&change.collection, &constraint.field, value) {
+        // Bitemporal collections: only consider live (non-superseded) rows,
+        // so a new version of the same logical row with the same value does
+        // not spuriously collide with its prior version.
+        let exists = if self.is_bitemporal(&change.collection) {
+            state.field_value_exists_live(&change.collection, &constraint.field, value)
+        } else {
+            state.field_value_exists(&change.collection, &constraint.field, value)
+        };
+
+        if exists {
             let value_str = format!("{:?}", value);
             Some(Violation {
                 constraint_name: constraint.name.clone(),
diff --git a/nodedb-crdt/src/state/core.rs b/nodedb-crdt/src/state/core.rs
@@ -3,6 +3,20 @@
 use loro::{LoroDoc, LoroMap, LoroValue, ValueOrContainer};
 
 use crate::error::{CrdtError, Result};
+use crate::validator::bitemporal::{VALID_UNTIL, VALID_UNTIL_OPEN};
+
+/// A row is live when its `_ts_valid_until` field is absent, null, or the
+/// open sentinel (`i64::MAX`). Rows with any finite `_ts_valid_until` are
+/// treated as superseded, independent of wall-clock time — the write path
+/// sets finite `_ts_valid_until` only when explicitly terminating a version.
+fn row_is_live(row: &LoroMap) -> bool {
+    match row.get(VALID_UNTIL) {
+        None => true,
+        Some(ValueOrContainer::Value(LoroValue::Null)) => true,
+        Some(ValueOrContainer::Value(LoroValue::I64(n))) => n == VALID_UNTIL_OPEN,
+        _ => true,
+    }
+}
 
 /// A CRDT state for a single tenant/namespace.
 pub struct CrdtState {
@@ -137,6 +151,56 @@ impl CrdtState {
         false
     }
 
+    /// Bitemporal variant of [`field_value_exists`]: only considers rows
+    /// whose `_ts_valid_until` is open (absent or `i64::MAX`).
+    ///
+    /// A UNIQUE collision between a superseded version and a new live row
+    /// is not a violation — both may share the same value because they
+    /// represent the same logical entity at different valid-times.
+    pub fn field_value_exists_live(
+        &self,
+        collection: &str,
+        field: &str,
+        value: &LoroValue,
+    ) -> bool {
+        let coll = self.doc.get_map(collection);
+        for key in coll.keys() {
+            let row_map = match coll.get(&key) {
+                Some(ValueOrContainer::Container(loro::Container::Map(m))) => m,
+                _ => continue,
+            };
+            if !row_is_live(&row_map) {
+                continue;
+            }
+            let field_val = match row_map.get(field) {
+                Some(ValueOrContainer::Value(v)) => v,
+                _ => continue,
+            };
+            if &field_val == value {
+                return true;
+            }
+        }
+        false
+    }
+
+    /// Return row IDs currently "live" in a bitemporal collection
+    /// (rows whose `_ts_valid_until` is open). For non-bitemporal
+    /// collections every row is returned.
+    pub fn live_row_ids(&self, collection: &str) -> Vec<String> {
+        let coll = self.doc.get_map(collection);
+        let mut out = Vec::new();
+        for key in coll.keys() {
+            let row_map = match coll.get(&key) {
+                Some(ValueOrContainer::Container(loro::Container::Map(m))) => m,
+                _ => continue,
+            };
+            if row_is_live(&row_map) {
+                out.push(key.to_string());
+            }
+        }
+        out
+    }
+
     /// Get the underlying LoroDoc for advanced operations.
     pub fn doc(&self) -> &LoroDoc {
         &self.doc
diff --git a/nodedb/src/engine/crdt/tenant_state/core.rs b/nodedb/src/engine/crdt/tenant_state/core.rs
@@ -90,6 +90,11 @@ impl TenantCrdtEngine {
     ///
     /// If constraints are violated, the delta is routed to the DLQ.
     /// Returns `Ok(())` on success, or the constraint violation error.
+    ///
+    /// For bitemporal collections, `_ts_system` is always stamped with the
+    /// receiving node's clock, overwriting any value the sender supplied.
+    /// This keeps system-time receiver-authoritative so convergence does
+    /// not depend on clock agreement between peers.
     pub fn validate_and_apply(
         &mut self,
         peer_id: u64,
@@ -101,20 +106,53 @@ impl TenantCrdtEngine {
             .validate_or_reject(&self.state, peer_id, auth, change, delta_bytes)
             .map_err(crate::Error::Crdt)?;
 
-        // Validation passed — apply to state.
-        // In production this would apply the delta bytes, but for now
-        // we upsert the fields directly since we have the ProposedChange.
-        let fields: Vec<(&str, LoroValue)> = change
+        let is_bitemporal = self.validator.is_bitemporal(&change.collection);
+        let now_ms = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_millis() as i64;
+
+        let mut fields: Vec<(&str, LoroValue)> = change
             .fields
             .iter()
+            .filter(|(k, _)| !(is_bitemporal && k == "_ts_system"))
             .map(|(k, v)| (k.as_str(), v.clone()))
             .collect();
 
+        if is_bitemporal {
+            fields.push(("_ts_system", LoroValue::I64(now_ms)));
+        }
+
         self.state
             .upsert(&change.collection, &change.row_id, &fields)
             .map_err(crate::Error::Crdt)
     }
 
+    /// Set the conflict-resolution policy for a collection from a typed
+    /// `CollectionPolicy`. The JSON-accepting variant in `policy.rs` is the
+    /// DDL-facing path; this one is for in-process callers (tests, engine
+    /// setup).
+    pub fn set_collection_policy_typed(
+        &mut self,
+        collection: &str,
+        policy: nodedb_crdt::policy::CollectionPolicy,
+    ) {
+        self.validator.policies_mut().set(collection, policy);
+    }
+
+    /// Register a collection as bitemporal on this tenant's validator.
+    ///
+    /// Bitemporal collections get (a) UNIQUE constraints scoped to live
+    /// rows only and (b) receiver-stamped `_ts_system` on apply.
+    pub fn mark_bitemporal(&mut self, collection: impl Into<String>) {
+        self.validator.mark_bitemporal(collection);
+    }
+
+    /// Is the named collection bitemporal?
+    pub fn is_bitemporal(&self, collection: &str) -> bool {
+        self.validator.is_bitemporal(collection)
+    }
+
     /// Number of entries in the dead-letter queue.
     pub fn dlq_len(&self) -> usize {
         self.validator.dlq().len()
diff --git a/nodedb/tests/bitemporal_crdt_sync.rs b/nodedb/tests/bitemporal_crdt_sync.rs
@@ -0,0 +1,141 @@
+//! Bitemporal CRDT sync: version preservation + live-scoped UNIQUE +
+//! receiver-stamped system time.
+//!
+//! The CRDT engine must:
+//! - Scope UNIQUE checks to currently-live rows (`_ts_valid_until == MAX`)
+//!   so a superseded version does not falsely collide with a new live row.
+//! - Stamp `_ts_system` with the receiving node's clock on `validate_and_apply`,
+//!   ignoring whatever the sender supplied — keeping system-time receiver-
+//!   authoritative for convergence.
+
+use loro::LoroValue;
+
+use nodedb::engine::crdt::tenant_state::TenantCrdtEngine;
+use nodedb::types::TenantId;
+use nodedb_crdt::CrdtAuthContext;
+use nodedb_crdt::constraint::ConstraintSet;
+use nodedb_crdt::policy::CollectionPolicy;
+use nodedb_crdt::validator::ProposedChange;
+use nodedb_crdt::validator::bitemporal::VALID_UNTIL_OPEN;
+
+const USERS: &str = "users";
+
+fn tenant() -> TenantCrdtEngine {
+    let mut cs = ConstraintSet::new();
+    cs.add_unique("users_email_unique", USERS, "email");
+    let mut eng = TenantCrdtEngine::new(TenantId::new(1), 1, cs).unwrap();
+    eng.mark_bitemporal(USERS);
+    eng.set_collection_policy_typed(USERS, CollectionPolicy::strict());
+    eng
+}
+
+fn change(
+    row_id: &str,
+    email: &str,
+    valid_from: i64,
+    valid_until: i64,
+    sender_system_ts: i64,
+) -> ProposedChange {
+    ProposedChange {
+        collection: USERS.into(),
+        row_id: row_id.into(),
+        fields: vec![
+            ("name".into(), LoroValue::String("Alice".into())),
+            ("email".into(), LoroValue::String(email.into())),
+            ("_ts_valid_from".into(), LoroValue::I64(valid_from)),
+            ("_ts_valid_until".into(), LoroValue::I64(valid_until)),
+            ("_ts_system".into(), LoroValue::I64(sender_system_ts)),
+        ],
+    }
+}
+
+/// A superseded version with the same email as a new live row should not
+/// trigger a UNIQUE collision: the old version is no longer live.
+#[test]
+fn bitemporal_unique_scoped_to_current() {
+    let mut eng = tenant();
+
+    // Version 1: Alice at alice@old.com, terminated at t=1000.
+    eng.validate_and_apply(
+        1,
+        CrdtAuthContext::default(),
+        &change("u1-v1", "alice@old.com", 0, 1_000, 100),
+        b"delta1".to_vec(),
+    )
+    .expect("v1 should accept");
+
+    // A second user insert at the SAME email (old one) — but the old row is
+    // no longer live, so this must succeed, not collide.
+    eng.validate_and_apply(
+        1,
+        CrdtAuthContext::default(),
+        &change("u2-v1", "alice@old.com", 2_000, VALID_UNTIL_OPEN, 200),
+        b"delta2".to_vec(),
+    )
+    .expect("reusing superseded email should not collide");
+}
+
+/// Two currently-live rows with the same UNIQUE value MUST collide even in
+/// bitemporal mode — bitemporal relaxes only the old-vs-new axis.
+#[test]
+fn bitemporal_unique_still_collides_for_live_rows() {
+    let mut eng = tenant();
+
+    eng.validate_and_apply(
+        1,
+        CrdtAuthContext::default(),
+        &change("u1", "bob@example.com", 0, VALID_UNTIL_OPEN, 100),
+        b"delta1".to_vec(),
+    )
+    .expect("first live insert accepted");
+
+    let err = eng.validate_and_apply(
+        1,
+        CrdtAuthContext::default(),
+        &change("u2", "bob@example.com", 0, VALID_UNTIL_OPEN, 200),
+        b"delta2".to_vec(),
+    );
+    assert!(
+        err.is_err(),
+        "two live rows with same UNIQUE value must collide"
+    );
+}
+
+/// The sender's `_ts_system` must be overwritten with receiver's clock on
+/// apply — receiver-authoritative system time.
+#[test]
+fn bitemporal_crdt_system_ts_receiver_stamped() {
+    let mut eng = tenant();
+
+    let sender_ts = 100_000i64; // some arbitrary sender-supplied stamp
+    let before_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_millis() as i64;
+
+    eng.validate_and_apply(
+        1,
+        CrdtAuthContext::default(),
+        &change("u1", "c@example.com", 0, VALID_UNTIL_OPEN, sender_ts),
+        b"delta".to_vec(),
+    )
+    .unwrap();
+
+    let row = eng.read_row(USERS, "u1").expect("row present");
+    let map = match row {
+        LoroValue::Map(m) => m,
+        other => panic!("expected map, got {other:?}"),
+    };
+    let stamped = match map.get("_ts_system") {
+        Some(LoroValue::I64(n)) => *n,
+        other => panic!("expected i64 _ts_system, got {other:?}"),
+    };
+    assert_ne!(
+        stamped, sender_ts,
+        "receiver must overwrite sender-supplied _ts_system"
+    );
+    assert!(
+        stamped >= before_ms,
+        "stamped ts ({stamped}) must be >= receiver clock at apply ({before_ms})"
+    );
+}
diff --git a/nodedb/tests/checkpoint_durability.rs b/nodedb/tests/checkpoint_durability.rs
@@ -131,7 +131,9 @@ fn timeseries_partition_registry_persist_is_durable() {
     // `PartitionRegistry::persist` writes the authoritative partition map via
     // the same tmp+rename pattern — recovery reads it on startup, so the
     // zero-file failure mode is identical to checkpoint writers.
-    assert_durable_checkpoint_writer("nodedb/src/engine/timeseries/partition_registry/persistence.rs");
+    assert_durable_checkpoint_writer(
+        "nodedb/src/engine/timeseries/partition_registry/persistence.rs",
+    );
 }
 
 #[test]
