fix(crdt): scope constraint checkpoints and snapshots by database

CRDT constraint checkpoint filenames and tombstone lookups were keyed
by tenant and collection only, so a collection with the same name in
a different database within the same tenant could collide. Include
the database id in checkpoint filenames and replace the crdt_constraints
snapshot tuple with a named CrdtConstraintEntry struct carrying the
database id alongside tenant, collection, version, and constraints.

Author: farhan-syah

nodedb-cluster-tests/tests/cluster_collection_hard_delete.rs

@@ -32,8 +32,14 @@ fn has_tombstone(node: &TestClusterNode, name: &str) -> bool {
     let cat = node.shared.credentials.catalog();
     cat.load_wal_tombstones()
         .map(|set| {
-            set.iter()
-                .any(|(tenant, n, lsn)| tenant == 1 && n == name && lsn > 0)
+            // Tombstones are keyed by database as well as tenant; this helper
+            // only ever asks about collections in the default database.
+            set.iter().any(|(database, tenant, n, lsn)| {
+                database == nodedb_types::DatabaseId::DEFAULT.as_u64()
+                    && tenant == 1
+                    && n == name
+                    && lsn > 0
+            })
         })
         .unwrap_or(false)
 }

nodedb-cluster-tests/tests/cluster_post_apply_follower_dispatch.rs

@@ -45,13 +45,18 @@ fn pick_follower_index(cluster: &TestCluster) -> usize {
 
 /// Tombstone tuples `(tenant_id, collection, purge_lsn)` currently
 /// persisted in this node's `_system.wal_tombstones`.
+///
+/// Tombstones are keyed by database as well as tenant; this test only
+/// creates and purges collections in the default database, so entries from
+/// any other database are filtered out rather than flattened away.
 fn follower_tombstones(node: &common::cluster_harness::TestClusterNode) -> Vec<(u64, String, u64)> {
     let catalog = node.shared.credentials.catalog();
     catalog
         .load_wal_tombstones()
         .expect("load_wal_tombstones")
         .iter()
-        .map(|(t, n, l)| (t, n.to_string(), l))
+        .filter(|(database, _, _, _)| *database == nodedb_types::DatabaseId::DEFAULT.as_u64())
+        .map(|(_, t, n, l)| (t, n.to_string(), l))
         .collect()
 }
 

nodedb-cluster-tests/tests/install_snapshot_crdt_constraints_cluster.rs

@@ -128,11 +128,12 @@ async fn crdt_constraint_survives_snapshot_capture_and_restore() {
          validator's installed constraint set"
     );
     assert!(
-        snap.crdt_constraints
-            .iter()
-            .any(|(tid, coll, version, encoded)| {
-                *tid == TENANT && coll == COLLECTION && *version == 1 && !encoded.is_empty()
-            }),
+        snap.crdt_constraints.iter().any(|entry| {
+            entry.tenant_id == TENANT
+                && entry.collection == COLLECTION
+                && entry.version == 1
+                && !entry.constraints.is_empty()
+        }),
         "captured snapshot did not contain an entry for (tenant={TENANT}, collection={COLLECTION}, \
          version=1) with at least one constraint blob; observed: {:?}",
         snap.crdt_constraints

nodedb/src/control/cluster/snapshot_builder.rs

@@ -225,11 +225,9 @@ impl DataPlaneSnapshotBuilder {
 
         // CRDT constraints: same per-collection vshard filter as `crdt_state`
         // — each entry is routed by its single collection's vshard.
-        for (database_id, tid, collection, version, encoded) in snap.crdt_constraints {
-            if group_vshards.contains(&Self::vshard_of(&collection)) {
-                merged
-                    .crdt_constraints
-                    .push((database_id, tid, collection, version, encoded));
+        for entry in snap.crdt_constraints {
+            if group_vshards.contains(&Self::vshard_of(&entry.collection)) {
+                merged.crdt_constraints.push(entry);
             }
         }
 

nodedb/src/data/executor/crdt_checkpoint.rs

@@ -4,7 +4,7 @@
 //!
 //! The matching write path lives in `handlers/control/checkpoint_crdt.rs`
 //! (`checkpoint_crdt_engines`). Checkpoints are written per-core to
-//! `{data_dir}/crdt-ckpt/core-{core_id}/tenant-{tid}-coll-{hex(collection)}.ckpt` because
+//! `{data_dir}/crdt-ckpt/core-{core_id}/db-{dbid}-tenant-{tid}-coll-{hex(collection)}.ckpt` because
 //! `data_dir` is shared across cores and each core only owns the CRDT
 //! fragments routed to its vShards.
 
@@ -21,7 +21,8 @@ pub(crate) fn crdt_ckpt_dir(data_dir: &std::path::Path, core_id: usize) -> std::
     data_dir.join("crdt-ckpt").join(format!("core-{core_id}"))
 }
 
-/// Per-collection checkpoint filename: `tenant-{tid}-coll-{hex(collection)}.ckpt`.
+/// Per-collection checkpoint filename:
+/// `db-{dbid}-tenant-{tid}-coll-{hex(collection)}.ckpt`.
 ///
 /// The collection is hex-encoded so the filename is filesystem-safe (collection
 /// names may contain `/`, `:` or `-`) and unambiguously parseable: hex contains
@@ -101,7 +102,8 @@ impl CoreLoop {
                 continue;
             }
 
-            // Checkpoint filenames are `"tenant-{tid}-coll-{hex(collection)}.ckpt"`.
+            // Checkpoint filenames are
+            // `"db-{dbid}-tenant-{tid}-coll-{hex(collection)}.ckpt"`.
             let stem = path
                 .file_stem()
                 .and_then(|s| s.to_str())
@@ -199,7 +201,7 @@ mod tests {
             .expect("an absent checkpoint dir must not be treated as corruption");
     }
 
-    /// A `.ckpt` file with a valid, parseable `tenant-{tid}-coll-{hex}` stem
+    /// A `.ckpt` file with a valid, parseable `db-{dbid}-tenant-{tid}-coll-{hex}` stem
     /// but bytes that are not a real Loro snapshot must fail the load, not be
     /// silently skipped: once the WAL below this checkpoint's LSN is
     /// truncated, the checkpoint is the only durable copy of the CRDT state,

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

@@ -16,7 +16,7 @@ impl CoreLoop {
     /// durable through, plus the number of checkpoint files published.
     ///
     /// Each tenant's Loro state is exported per collection and written to
-    /// `{data_dir}/crdt-ckpt/core-{core_id}/tenant-{tid}-coll-{hex(collection)}.ckpt`
+    /// `{data_dir}/crdt-ckpt/core-{core_id}/db-{dbid}-tenant-{tid}-coll-{hex(collection)}.ckpt`
     /// with atomic temp+rename. The per-core subdir is required because `data_dir` is
     /// shared across all cores and a tenant's CRDT state is fragmented across
     /// cores by collection — without the subdir, cores would race-overwrite
@@ -62,10 +62,10 @@ impl CoreLoop {
         for ((database_id, tenant_id), engine) in &self.crdt_engines {
             let database_id = database_id.as_u64();
             let tid = tenant_id.as_u64();
-            // One checkpoint file per (tenant, collection) — each collection
-            // owns its own LoroDoc. Filenames are
-            // `tenant-{id}-coll-{hex(collection)}.ckpt`, matching the loader's
-            // parse and the cluster-restore writer.
+            // One checkpoint file per (database, tenant, collection) — each
+            // collection owns its own LoroDoc. Filenames are
+            // `db-{dbid}-tenant-{id}-coll-{hex(collection)}.ckpt`, matching the
+            // loader's parse and the cluster-restore writer.
             let snapshots = engine
                 .export_all_snapshots()
                 .map_err(|e| crate::Error::Storage {

nodedb/src/data/executor/handlers/snapshot/create.rs

@@ -173,13 +173,15 @@ impl CoreLoop {
                 if failed || encoded.is_empty() {
                     continue;
                 }
-                snapshot.crdt_constraints.push((
-                    task.request.database_id.as_u64(),
-                    tenant_id,
-                    collection,
-                    version,
-                    encoded,
-                ));
+                snapshot
+                    .crdt_constraints
+                    .push(crate::types::snapshot::CrdtConstraintEntry {
+                        database_id: task.request.database_id.as_u64(),
+                        tenant_id,
+                        collection,
+                        version,
+                        constraints: encoded,
+                    });
             }
         }
 

nodedb/src/data/executor/handlers/snapshot/restore/tenant_snapshot.rs

@@ -276,14 +276,15 @@ impl CoreLoop {
             // on error — warn and continue, matching the `crdt_state` loop, since
             // a failed reconstruction only reverts to the pre-fix (over-rejecting)
             // behavior rather than corrupting state.
-            for (database_raw, tid_raw, collection, version, encoded) in &snap.crdt_constraints {
+            for entry in &snap.crdt_constraints {
                 if let Err(e) = self.restore_crdt_constraints(
-                    *database_raw,
-                    *tid_raw,
-                    collection,
-                    *version,
-                    encoded,
+                    entry.database_id,
+                    entry.tenant_id,
+                    &entry.collection,
+                    entry.version,
+                    &entry.constraints,
                 ) {
+                    let (tid_raw, collection) = (entry.tenant_id, &entry.collection);
                     warn!(tid_raw, %collection, error = %e, "failed to restore crdt constraints");
                 } else {
                     crdt_constraints_written += 1;

nodedb/src/types/snapshot.rs

@@ -156,7 +156,35 @@ pub struct TenantDataSnapshot {
     /// pre-fix (fail-safe, over-rejecting) behavior rather than a hard error.
     #[msgpack(default)]
     #[serde(default)]
-    pub crdt_constraints: Vec<(u64, u64, String, u64, Vec<Vec<u8>>)>,
+    pub crdt_constraints: Vec<CrdtConstraintEntry>,
+}
+
+/// One collection's CRDT constraint set plus its installed version, carried
+/// through a snapshot so `restore_crdt_constraints` can reconstruct the
+/// validator and the version together.
+#[derive(
+    Debug,
+    Clone,
+    PartialEq,
+    Eq,
+    serde::Serialize,
+    serde::Deserialize,
+    zerompk::ToMessagePack,
+    zerompk::FromMessagePack,
+    Default,
+)]
+pub struct CrdtConstraintEntry {
+    /// Database owning the collection.
+    pub database_id: u64,
+    /// Tenant owning the collection.
+    pub tenant_id: u64,
+    /// Collection the constraint set applies to.
+    pub collection: String,
+    /// Installed constraint version. The write gate rejects peer deltas whose
+    /// `constraint_version_required` exceeds this.
+    pub version: u64,
+    /// Encoded constraint definitions.
+    pub constraints: Vec<Vec<u8>>,
 }
 
 /// A single PK → surrogate identity binding carried in a snapshot/backup.
@@ -529,13 +557,13 @@ mod tests {
         let not_null_bytes = zerompk::to_msgpack_vec(&not_null).expect("encode NotNull constraint");
 
         let snap = TenantDataSnapshot {
-            crdt_constraints: vec![(
-                0,
-                7,
-                "users".to_string(),
-                3,
-                vec![unique_bytes, not_null_bytes],
-            )],
+            crdt_constraints: vec![CrdtConstraintEntry {
+                database_id: 0,
+                tenant_id: 7,
+                collection: "users".to_string(),
+                version: 3,
+                constraints: vec![unique_bytes, not_null_bytes],
+            }],
             ..Default::default()
         };
 
@@ -544,11 +572,12 @@ mod tests {
             zerompk::from_msgpack(&bytes).expect("decode snapshot with crdt_constraints");
 
         assert_eq!(decoded.crdt_constraints.len(), 1);
-        let (database_id, tid, collection, version, encoded) = &decoded.crdt_constraints[0];
-        assert_eq!(*database_id, 0);
-        assert_eq!(*tid, 7);
-        assert_eq!(collection, "users");
-        assert_eq!(*version, 3);
+        let entry = &decoded.crdt_constraints[0];
+        let encoded = &entry.constraints;
+        assert_eq!(entry.database_id, 0);
+        assert_eq!(entry.tenant_id, 7);
+        assert_eq!(entry.collection, "users");
+        assert_eq!(entry.version, 3);
         assert_eq!(encoded.len(), 2);
 
         let decoded_unique: nodedb_crdt::Constraint =

nodedb/tests/warm_storage_object_store.rs

@@ -292,14 +292,15 @@ async fn snapshot_bytes_roundtrip_write_and_restore() {
     assert_eq!(on_disk, hnsw_bytes, "HNSW checkpoint bytes must match");
 
     // ── Verify CRDT checkpoint file ──────────────────────────────────────────
-    // CRDT checkpoints are per-collection and per-core:
-    // `crdt-ckpt/core-{id}/tenant-{tid}-coll-{hex(collection)}.ckpt`. The hex
-    // encoding mirrors the engine's filename scheme (collection bytes, lowercase).
+    // CRDT checkpoints are per-collection, per-database and per-core:
+    // `crdt-ckpt/core-{id}/db-{dbid}-tenant-{tid}-coll-{hex(collection)}.ckpt`.
+    // The hex encoding mirrors the engine's filename scheme (collection bytes,
+    // lowercase).
     let crdt_coll_hex: String = "testcoll".bytes().map(|b| format!("{b:02x}")).collect();
     let crdt_ckpt = data_dir
         .join("crdt-ckpt")
         .join("core-0")
-        .join(format!("tenant-1-coll-{crdt_coll_hex}.ckpt"));
+        .join(format!("db-0-tenant-1-coll-{crdt_coll_hex}.ckpt"));
     assert!(
         crdt_ckpt.exists(),
         "CRDT checkpoint file must exist after restore"