Fix ownership and CRDT scoping by database

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);
             }
         }