From ee1946410fdca328f29df4968d44d71b24ce7a96 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Mon, 20 Jul 2026 17:48:14 +0800 Subject: [PATCH 1/2] fix(security): scope collection/index/function ownership by database Owner lookups previously keyed only on tenant_id, so a same-named collection in another database within the same tenant would incorrectly resolve to whichever owner happened to be recorded first. Thread database_id through permission checks, namespace authorization, and ownership DDL so ownership is correctly scoped per database. --- .../src/control/security/permission/check.rs | 255 ++++++++++++++++-- nodedb/src/control/security/rls/namespace.rs | 5 + .../server/shared/authorization/service.rs | 1 + .../ddl/neutral/collection/alter/dispatch.rs | 2 +- .../ddl/neutral/collection/alter/ownership.rs | 41 +-- .../shared/ddl/neutral/collection/index.rs | 2 +- .../shared/ddl/neutral/inspect/grants.rs | 12 +- .../shared/ddl/neutral/router/typed_auth.rs | 3 +- nodedb/tests/http_query_authorization.rs | 1 + 9 files changed, 268 insertions(+), 54 deletions(-) diff --git a/nodedb/src/control/security/permission/check.rs b/nodedb/src/control/security/permission/check.rs index b4e1d1903..bcf5a6767 100644 --- a/nodedb/src/control/security/permission/check.rs +++ b/nodedb/src/control/security/permission/check.rs @@ -9,10 +9,10 @@ use crate::control::security::audit::{AuditEmitContext, AuditEmitter, AuditEvent use crate::control::security::identity::{self, AuthenticatedIdentity, Permission}; use crate::control::security::role::RoleStore; -use crate::types::TenantId; +use crate::types::{DatabaseId, TenantId}; use super::store::PermissionStore; -use super::types::{Grant, collection_target, function_target, tenant_target}; +use super::types::{Grant, collection_target, function_target, owner_key, tenant_target}; impl PermissionStore { /// Does any grant on `target` confer `permission` to this identity — @@ -83,6 +83,7 @@ impl PermissionStore { &self, identity: &AuthenticatedIdentity, permission: Permission, + database_id: DatabaseId, collection: &str, role_store: &RoleStore, emitter: &dyn AuditEmitter, @@ -91,11 +92,18 @@ impl PermissionStore { return true; } - let target = collection_target(identity.tenant_id, collection); - if self.is_owner(&target, &identity.username) { + if self.is_owner( + "collection", + database_id, + identity.tenant_id, + collection, + &identity.username, + ) { return true; } + let target = collection_target(identity.tenant_id, collection); + for role in &identity.roles { if identity::role_grants_permission(role, permission) { return true; @@ -139,6 +147,7 @@ impl PermissionStore { pub fn check_function( &self, identity: &AuthenticatedIdentity, + database_id: DatabaseId, function_name: &str, role_store: &RoleStore, emitter: &dyn AuditEmitter, @@ -147,12 +156,18 @@ impl PermissionStore { return true; } - let target = function_target(identity.tenant_id, function_name); - - if self.is_owner(&target, &identity.username) { + if self.is_owner( + "function", + database_id, + identity.tenant_id, + function_name, + &identity.username, + ) { return true; } + let target = function_target(identity.tenant_id, function_name); + for role in &identity.roles { if identity::role_grants_permission(role, Permission::Execute) { return true; @@ -227,8 +242,28 @@ impl PermissionStore { false } - /// Lookup helper: is `username` recorded as the owner of `target`? - pub(super) fn is_owner(&self, target: &str, username: &str) -> bool { + /// Lookup helper: is `username` recorded as the owner of the object? + /// + /// The owners map is keyed by [`owner_key`] — + /// `{object_type}:{database_id}:{tenant_id}:{object_name}` — which is a + /// different shape from the `{object_type}:{tenant_id}:{object_name}` + /// target strings used for *grants*. The two must not be interchanged: + /// passing a grant target here silently never matches, which reads as + /// "nobody owns anything" rather than as an error. + pub(super) fn is_owner( + &self, + object_type: &str, + database_id: DatabaseId, + tenant_id: TenantId, + object_name: &str, + username: &str, + ) -> bool { + let key = owner_key( + object_type, + database_id.as_u64(), + tenant_id.as_u64(), + object_name, + ); let owners = match self.owners.read() { Ok(o) => o, Err(p) => { @@ -236,7 +271,7 @@ impl PermissionStore { p.into_inner() } }; - owners.get(target).is_some_and(|o| o == username) + owners.get(&key).is_some_and(|o| o == username) } } @@ -272,7 +307,14 @@ mod tests { let store = PermissionStore::new(); let roles = RoleStore::new(); let id = identity("admin", vec![], true); - assert!(store.check(&id, Permission::Write, "secret", &roles, NOOP)); + assert!(store.check( + &id, + Permission::Write, + DatabaseId::DEFAULT, + "secret", + &roles, + NOOP + )); } #[test] @@ -284,9 +326,68 @@ mod tests { .unwrap(); let id = identity("alice", vec![], false); - assert!(store.check(&id, Permission::Read, "users", &roles, NOOP)); - assert!(store.check(&id, Permission::Write, "users", &roles, NOOP)); - assert!(store.check(&id, Permission::Drop, "users", &roles, NOOP)); + assert!(store.check( + &id, + Permission::Read, + DatabaseId::DEFAULT, + "users", + &roles, + NOOP + )); + assert!(store.check( + &id, + Permission::Write, + DatabaseId::DEFAULT, + "users", + &roles, + NOOP + )); + assert!(store.check( + &id, + Permission::Drop, + DatabaseId::DEFAULT, + "users", + &roles, + NOOP + )); + } + + /// Owner rows are keyed by database. A check against the database the + /// row was written to must recognise the owner, and a check against any + /// other database must not — a same-named collection elsewhere belongs + /// to whoever owns it there, not to this user. + #[test] + fn ownership_is_scoped_to_its_database() { + let store = PermissionStore::new(); + let roles = RoleStore::new(); + let db = DatabaseId::new(7); + store + .set_owner_in_database( + "collection", + db.as_u64(), + TenantId::new(1), + "users", + "alice", + None, + ) + .unwrap(); + + let id = identity("alice", vec![], false); + assert!( + store.check(&id, Permission::Read, db, "users", &roles, NOOP), + "owner must hold implicit permissions in their own database" + ); + assert!( + !store.check( + &id, + Permission::Read, + DatabaseId::DEFAULT, + "users", + &roles, + NOOP + ), + "ownership must not leak into a same-named collection in another database" + ); } #[test] @@ -298,7 +399,14 @@ mod tests { .unwrap(); let id = identity("bob", vec![], false); - assert!(!store.check(&id, Permission::Write, "users", &roles, NOOP)); + assert!(!store.check( + &id, + Permission::Write, + DatabaseId::DEFAULT, + "users", + &roles, + NOOP + )); } #[test] @@ -311,8 +419,22 @@ mod tests { .unwrap(); let id = identity("bob", vec![], false); - assert!(store.check(&id, Permission::Read, "orders", &roles, NOOP)); - assert!(!store.check(&id, Permission::Write, "orders", &roles, NOOP)); + assert!(store.check( + &id, + Permission::Read, + DatabaseId::DEFAULT, + "orders", + &roles, + NOOP + )); + assert!(!store.check( + &id, + Permission::Write, + DatabaseId::DEFAULT, + "orders", + &roles, + NOOP + )); } #[test] @@ -325,7 +447,14 @@ mod tests { .unwrap(); let id = identity("viewer", vec![Role::Custom("readonly".into())], false); - assert!(store.check(&id, Permission::Read, "reports", &roles, NOOP)); + assert!(store.check( + &id, + Permission::Read, + DatabaseId::DEFAULT, + "reports", + &roles, + NOOP + )); } #[test] @@ -342,7 +471,14 @@ mod tests { .unwrap(); let id = identity("alice", vec![Role::Custom("analyst".into())], false); - assert!(perm_store.check(&id, Permission::Read, "data", &role_store, NOOP)); + assert!(perm_store.check( + &id, + Permission::Read, + DatabaseId::DEFAULT, + "data", + &role_store, + NOOP + )); } #[test] @@ -360,7 +496,14 @@ mod tests { let roles = RoleStore::new(); let id = identity("bob", vec![], false); - assert!(!store.check(&id, Permission::Read, "users", &roles, NOOP)); + assert!(!store.check( + &id, + Permission::Read, + DatabaseId::DEFAULT, + "users", + &roles, + NOOP + )); } #[test] @@ -368,9 +511,30 @@ mod tests { let store = PermissionStore::new(); let roles = RoleStore::new(); let id = identity("writer", vec![Role::ReadWrite], false); - assert!(store.check(&id, Permission::Read, "anything", &roles, NOOP)); - assert!(store.check(&id, Permission::Write, "anything", &roles, NOOP)); - assert!(!store.check(&id, Permission::Drop, "anything", &roles, NOOP)); + assert!(store.check( + &id, + Permission::Read, + DatabaseId::DEFAULT, + "anything", + &roles, + NOOP + )); + assert!(store.check( + &id, + Permission::Write, + DatabaseId::DEFAULT, + "anything", + &roles, + NOOP + )); + assert!(!store.check( + &id, + Permission::Drop, + DatabaseId::DEFAULT, + "anything", + &roles, + NOOP + )); } #[test] @@ -382,7 +546,14 @@ mod tests { let emitter = CapturingEmitter::new(); let id = identity("eve", vec![], false); - let allowed = store.check(&id, Permission::Write, "secrets", &roles, &emitter); + let allowed = store.check( + &id, + Permission::Write, + DatabaseId::DEFAULT, + "secrets", + &roles, + &emitter, + ); assert!(!allowed); let recorded = emitter.recorded(); @@ -399,7 +570,14 @@ mod tests { let emitter = CapturingEmitter::new(); let id = identity("admin", vec![], true); - let allowed = store.check(&id, Permission::Write, "anything", &roles, &emitter); + let allowed = store.check( + &id, + Permission::Write, + DatabaseId::DEFAULT, + "anything", + &roles, + &emitter, + ); assert!(allowed); assert!(emitter.recorded().is_empty()); } @@ -417,10 +595,31 @@ mod tests { let id = identity("bob", vec![], false); // A tenant-wide grant confers the permission on any collection in // the tenant, with no per-collection grant. - assert!(store.check(&id, Permission::Read, "orders", &roles, NOOP)); - assert!(store.check(&id, Permission::Read, "invoices", &roles, NOOP)); + assert!(store.check( + &id, + Permission::Read, + DatabaseId::DEFAULT, + "orders", + &roles, + NOOP + )); + assert!(store.check( + &id, + Permission::Read, + DatabaseId::DEFAULT, + "invoices", + &roles, + NOOP + )); // It does not widen to permissions that were not granted. - assert!(!store.check(&id, Permission::Write, "orders", &roles, NOOP)); + assert!(!store.check( + &id, + Permission::Write, + DatabaseId::DEFAULT, + "orders", + &roles, + NOOP + )); } #[test] diff --git a/nodedb/src/control/security/rls/namespace.rs b/nodedb/src/control/security/rls/namespace.rs index 64c287e51..35a02177e 100644 --- a/nodedb/src/control/security/rls/namespace.rs +++ b/nodedb/src/control/security/rls/namespace.rs @@ -10,11 +10,13 @@ use crate::control::security::audit::NoopAuditEmitter; use crate::control::security::identity::{AuthenticatedIdentity, Permission}; use crate::control::security::permission::PermissionStore; use crate::control::security::role::RoleStore; +use crate::types::DatabaseId; /// Check tenant + namespace authorization for a collection. /// Order: direct collection grant → namespace prefix grants → wildcard. pub fn check_namespace_authz( identity: &AuthenticatedIdentity, + database_id: DatabaseId, collection: &str, required_permission: Permission, permission_store: &PermissionStore, @@ -27,6 +29,7 @@ pub fn check_namespace_authz( if permission_store.check( identity, required_permission, + database_id, collection, role_store, &NoopAuditEmitter, @@ -41,6 +44,7 @@ pub fn check_namespace_authz( && permission_store.check( identity, required_permission, + database_id, &namespace, role_store, &NoopAuditEmitter, @@ -53,6 +57,7 @@ pub fn check_namespace_authz( permission_store.check( identity, required_permission, + database_id, "*", role_store, &NoopAuditEmitter, diff --git a/nodedb/src/control/server/shared/authorization/service.rs b/nodedb/src/control/server/shared/authorization/service.rs index 2918c5de8..38b7af5b0 100644 --- a/nodedb/src/control/server/shared/authorization/service.rs +++ b/nodedb/src/control/server/shared/authorization/service.rs @@ -158,6 +158,7 @@ fn authorize_collection_requirement( if permissions.check( &scoped_identity, permission, + database_id, &grant_name, roles, &NoopAuditEmitter, diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/alter/dispatch.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/alter/dispatch.rs index ffe0a39bd..d6b66868a 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/alter/dispatch.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/collection/alter/dispatch.rs @@ -69,7 +69,7 @@ pub async fn dispatch_alter_collection( } AlterCollectionOp::OwnerTo { new_owner } => { - super::ownership::alter_collection_owner(state, identity, name, new_owner) + super::ownership::alter_collection_owner(state, identity, database_id, name, new_owner) } AlterCollectionOp::SetRetention { value } => { diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/alter/ownership.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/alter/ownership.rs index ee47aa5c5..95d2fd4e8 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/alter/ownership.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/collection/alter/ownership.rs @@ -29,13 +29,19 @@ use super::support::{err, status}; pub(super) fn alter_collection_owner( state: &SharedState, identity: &AuthenticatedIdentity, + database_id: DatabaseId, collection: &str, new_owner: &str, ) -> Result, DdlError> { - // Check authorization: current owner or admin. - let current_owner = state - .permissions - .get_owner("collection", identity.tenant_id, collection); + // Check authorization: current owner or admin. Owner rows are keyed by + // database, so a collection outside the default database is only found + // when the lookup carries the same `database_id` the row was written with. + let current_owner = state.permissions.get_owner_in_database( + "collection", + database_id.as_u64(), + identity.tenant_id, + collection, + ); let is_current_owner = current_owner .as_ref() @@ -65,27 +71,24 @@ pub(super) fn alter_collection_owner( // the parent record. A separate `PutOwner` would be silently // overwritten the next time anyone re-proposed the collection. let catalog = state.credentials.catalog(); - let mut stored = match catalog.get_collection( - DatabaseId::DEFAULT, - identity.tenant_id.as_u64(), - collection, - ) { - Ok(Some(c)) => c, - Ok(None) => { - return Err(err( - "42P01", - format!("collection '{collection}' does not exist"), - )); - } - Err(e) => return Err(err("XX000", format!("catalog read: {e}"))), - }; + let mut stored = + match catalog.get_collection(database_id, identity.tenant_id.as_u64(), collection) { + Ok(Some(c)) => c, + Ok(None) => { + return Err(err( + "42P01", + format!("collection '{collection}' does not exist"), + )); + } + Err(e) => return Err(err("XX000", format!("catalog read: {e}"))), + }; stored.owner = new_owner.to_string(); let entry = CatalogEntry::PutCollection(Box::new(stored.clone())); let log_index = propose_catalog_entry(state, &entry) .map_err(|e| err("XX000", format!("metadata propose: {e}")))?; if log_index == 0 { catalog - .put_collection(DatabaseId::DEFAULT, &stored) + .put_collection(database_id, &stored) .map_err(|e| err("XX000", format!("catalog write: {e}")))?; state.permissions.install_replicated_owner( &crate::control::security::catalog::StoredOwner { diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/index.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/index.rs index 6f49da951..68fb8daec 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/collection/index.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/collection/index.rs @@ -312,7 +312,7 @@ pub async fn drop_index( // Check ownership or admin. let is_owner = state .permissions - .get_owner("index", tenant_id, &index_name) + .get_owner_in_database("index", database_id.as_u64(), tenant_id, &index_name) .as_deref() == Some(&identity.username); diff --git a/nodedb/src/control/server/shared/ddl/neutral/inspect/grants.rs b/nodedb/src/control/server/shared/ddl/neutral/inspect/grants.rs index 0474669fb..7e6d61178 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/inspect/grants.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/inspect/grants.rs @@ -13,6 +13,7 @@ use serde_json::{Map, Value as JsonValue}; use crate::control::security::identity::AuthenticatedIdentity; use crate::control::server::response_shape::types::{DdlColType, ShapedRows}; use crate::control::state::SharedState; +use crate::types::DatabaseId; use super::super::super::result::{DdlError, DdlResult}; use super::support::ddl_err; @@ -82,6 +83,7 @@ pub fn show_grants( pub fn show_permissions( state: &SharedState, identity: &AuthenticatedIdentity, + database_id: DatabaseId, on_collection: Option<&str>, for_grantee: Option<&str>, ) -> Result, DdlError> { @@ -117,10 +119,12 @@ pub fn show_permissions( // Show owner row (only when collection is specified). if for_grantee.is_none() - && let Some(owner) = - state - .permissions - .get_owner("collection", identity.tenant_id, collection) + && let Some(owner) = state.permissions.get_owner_in_database( + "collection", + database_id.as_u64(), + identity.tenant_id, + collection, + ) { let mut row = Map::new(); row.insert("grantee".to_string(), JsonValue::String(owner)); diff --git a/nodedb/src/control/server/shared/ddl/neutral/router/typed_auth.rs b/nodedb/src/control/server/shared/ddl/neutral/router/typed_auth.rs index 05b44f22d..de719e1b7 100644 --- a/nodedb/src/control/server/shared/ddl/neutral/router/typed_auth.rs +++ b/nodedb/src/control/server/shared/ddl/neutral/router/typed_auth.rs @@ -20,7 +20,7 @@ pub(super) async fn try_typed( state: &SharedState, identity: &AuthenticatedIdentity, _sql: &str, - _database_id: DatabaseId, + database_id: DatabaseId, stmt: &NodedbStatement, ) -> Option, DdlError>> { match stmt { @@ -148,6 +148,7 @@ pub(super) async fn try_typed( }) => Some(inspect::show_permissions( state, identity, + database_id, on_collection.as_deref(), for_grantee.as_deref(), )), diff --git a/nodedb/tests/http_query_authorization.rs b/nodedb/tests/http_query_authorization.rs index 6bacf4e6c..c83647a4e 100644 --- a/nodedb/tests/http_query_authorization.rs +++ b/nodedb/tests/http_query_authorization.rs @@ -293,6 +293,7 @@ async fn query_honors_explicit_collection_grant_for_custom_role() { srv.shared.permissions.check( &identity, Permission::Read, + DatabaseId::DEFAULT, "granted_rows", &srv.shared.roles, &NoopAuditEmitter, From cd6bd22e456a9a6008892491c09941d24b0340f6 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Mon, 20 Jul 2026 17:48:39 +0800 Subject: [PATCH 2/2] 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. --- .../tests/cluster_collection_hard_delete.rs | 10 +++- .../cluster_post_apply_follower_dispatch.rs | 7 ++- ...stall_snapshot_crdt_constraints_cluster.rs | 11 ++-- .../src/control/cluster/snapshot_builder.rs | 8 +-- nodedb/src/data/executor/crdt_checkpoint.rs | 10 ++-- .../handlers/control/checkpoint_crdt.rs | 10 ++-- .../data/executor/handlers/snapshot/create.rs | 16 +++--- .../snapshot/restore/tenant_snapshot.rs | 13 +++-- nodedb/src/types/snapshot.rs | 55 ++++++++++++++----- nodedb/tests/warm_storage_object_store.rs | 9 +-- 10 files changed, 97 insertions(+), 52 deletions(-) diff --git a/nodedb-cluster-tests/tests/cluster_collection_hard_delete.rs b/nodedb-cluster-tests/tests/cluster_collection_hard_delete.rs index 9f59ee0b9..33204a0ab 100644 --- a/nodedb-cluster-tests/tests/cluster_collection_hard_delete.rs +++ b/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) } diff --git a/nodedb-cluster-tests/tests/cluster_post_apply_follower_dispatch.rs b/nodedb-cluster-tests/tests/cluster_post_apply_follower_dispatch.rs index 77855a8ac..ec41a58f7 100644 --- a/nodedb-cluster-tests/tests/cluster_post_apply_follower_dispatch.rs +++ b/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() } diff --git a/nodedb-cluster-tests/tests/install_snapshot_crdt_constraints_cluster.rs b/nodedb-cluster-tests/tests/install_snapshot_crdt_constraints_cluster.rs index cc6d232b5..668a95661 100644 --- a/nodedb-cluster-tests/tests/install_snapshot_crdt_constraints_cluster.rs +++ b/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 diff --git a/nodedb/src/control/cluster/snapshot_builder.rs b/nodedb/src/control/cluster/snapshot_builder.rs index 0726b436f..16782790e 100644 --- a/nodedb/src/control/cluster/snapshot_builder.rs +++ b/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); } } diff --git a/nodedb/src/data/executor/crdt_checkpoint.rs b/nodedb/src/data/executor/crdt_checkpoint.rs index 17564c970..a600673c3 100644 --- a/nodedb/src/data/executor/crdt_checkpoint.rs +++ b/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, diff --git a/nodedb/src/data/executor/handlers/control/checkpoint_crdt.rs b/nodedb/src/data/executor/handlers/control/checkpoint_crdt.rs index f212ef681..ce87bfa29 100644 --- a/nodedb/src/data/executor/handlers/control/checkpoint_crdt.rs +++ b/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 { diff --git a/nodedb/src/data/executor/handlers/snapshot/create.rs b/nodedb/src/data/executor/handlers/snapshot/create.rs index 35ee3cd36..6dfce9783 100644 --- a/nodedb/src/data/executor/handlers/snapshot/create.rs +++ b/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, + }); } } diff --git a/nodedb/src/data/executor/handlers/snapshot/restore/tenant_snapshot.rs b/nodedb/src/data/executor/handlers/snapshot/restore/tenant_snapshot.rs index 18d0c61dd..5ebea6575 100644 --- a/nodedb/src/data/executor/handlers/snapshot/restore/tenant_snapshot.rs +++ b/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; diff --git a/nodedb/src/types/snapshot.rs b/nodedb/src/types/snapshot.rs index aa8ecbfc9..a469e21f9 100644 --- a/nodedb/src/types/snapshot.rs +++ b/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>)>, + pub crdt_constraints: Vec, +} + +/// 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>, } /// 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(¬_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 = diff --git a/nodedb/tests/warm_storage_object_store.rs b/nodedb/tests/warm_storage_object_store.rs index 627b0b42d..1fc6d93d5 100644 --- a/nodedb/tests/warm_storage_object_store.rs +++ b/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"