fix catalog ownership reassignment recovery

Author: farhan-syah

nodedb-types/src/wire_version.rs

@@ -16,7 +16,7 @@
 
 /// Cluster-wide wire format version. Stamped on every `NodeInfo` and
 /// returned by `nodedb::version::WIRE_FORMAT_VERSION` (a re-export).
-pub const WIRE_FORMAT_VERSION: u16 = 7;
+pub const WIRE_FORMAT_VERSION: u16 = 8;
 
 /// Minimum wire format version this build can read. Equal to
 /// `WIRE_FORMAT_VERSION`: floor == ceiling, no backward compat window.

nodedb/src/control/catalog_entry/apply/mod.rs

@@ -46,8 +46,19 @@ use crate::control::security::catalog::SystemCatalog;
 /// developer runs instead of deferring to the next restart, so
 /// reviewers don't need to rely on a user report to surface
 /// half-finished sync work. Release builds skip the check.
-pub fn apply_to(entry: &CatalogEntry, catalog: &SystemCatalog) {
-    apply_to_inner(entry, catalog);
+pub fn apply_to(entry: &CatalogEntry, catalog: &SystemCatalog) -> bool {
+    let applied = match entry {
+        CatalogEntry::PutTenantWithAdmin { tenant, admin } => {
+            tenant::put_with_admin(tenant, admin, catalog)
+        }
+        _ => {
+            apply_to_inner(entry, catalog);
+            true
+        }
+    };
+    if !applied {
+        return false;
+    }
     #[cfg(debug_assertions)]
     {
         // Narrow to OrphanRow — the "half-finished sync" class this
@@ -72,6 +83,7 @@ pub fn apply_to(entry: &CatalogEntry, catalog: &SystemCatalog) {
             orphans,
         );
     }
+    true
 }
 
 fn apply_to_inner(entry: &CatalogEntry, catalog: &SystemCatalog) {
@@ -139,6 +151,8 @@ fn apply_to_inner(entry: &CatalogEntry, catalog: &SystemCatalog) {
             name,
         } => continuous_aggregate::delete(*database_id, *tenant_id, name, catalog),
         CatalogEntry::PutTenant(stored) => tenant::put(stored, catalog),
+        // Applied by `apply_to` so its commit outcome can suppress post-apply.
+        CatalogEntry::PutTenantWithAdmin { .. } => {}
         CatalogEntry::DeleteTenant { tenant_id } => tenant::delete(*tenant_id, catalog),
         CatalogEntry::PutRlsPolicy(stored) => rls::put(stored, catalog),
         CatalogEntry::DeleteRlsPolicy {

nodedb/src/control/catalog_entry/apply/tenant.rs

@@ -4,7 +4,9 @@
 
 use tracing::warn;
 
-use crate::control::security::catalog::{StoredCollection, StoredTenant, SystemCatalog};
+use crate::control::security::catalog::{
+    StoredCollection, StoredTenant, StoredUser, SystemCatalog,
+};
 use crate::types::DatabaseId;
 
 pub fn put(stored: &StoredTenant, catalog: &SystemCatalog) {
@@ -18,6 +20,22 @@ pub fn put(stored: &StoredTenant, catalog: &SystemCatalog) {
     }
 }
 
+pub fn put_with_admin(tenant: &StoredTenant, admin: &StoredUser, catalog: &SystemCatalog) -> bool {
+    match catalog.put_tenant_with_admin(tenant, admin) {
+        Ok(()) => true,
+        Err(error) => {
+            warn!(
+                tenant = tenant.tenant_id,
+                name = %tenant.name,
+                admin = %admin.username,
+                %error,
+                "catalog_entry: put_tenant_with_admin failed"
+            );
+            false
+        }
+    }
+}
+
 pub fn delete(tenant_id: u64, catalog: &SystemCatalog) {
     if let Err(e) = catalog.delete_tenant(tenant_id) {
         warn!(

nodedb/src/control/catalog_entry/descriptor_stamp.rs

@@ -194,6 +194,7 @@ pub fn stamp(entry: CatalogEntry, clock: &HlcClock, catalog: &SystemCatalog) ->
         | CatalogEntry::PutApiKey(_)
         | CatalogEntry::RevokeApiKey { .. }
         | CatalogEntry::PutTenant(_)
+        | CatalogEntry::PutTenantWithAdmin { .. }
         | CatalogEntry::DeleteTenant { .. }
         | CatalogEntry::PutRlsPolicy(_)
         | CatalogEntry::DeleteRlsPolicy { .. }

nodedb/src/control/catalog_entry/entry.rs

@@ -195,6 +195,11 @@ pub enum CatalogEntry {
     /// default quota on every node so reads work immediately after
     /// creation.
     PutTenant(Box<StoredTenant>),
+    /// Atomically create a tenant and its authoritative administrator.
+    PutTenantWithAdmin {
+        tenant: Box<StoredTenant>,
+        admin: Box<StoredUser>,
+    },
     /// Hard-delete a tenant identity record. Tenant data is not
     /// purged — that is a separate `PURGE TENANT CONFIRM` Data
     /// Plane meta op.
@@ -364,6 +369,7 @@ impl CatalogEntry {
             Self::PutContinuousAggregate(_) => "put_continuous_aggregate",
             Self::DeleteContinuousAggregate { .. } => "delete_continuous_aggregate",
             Self::PutTenant(_) => "put_tenant",
+            Self::PutTenantWithAdmin { .. } => "put_tenant_with_admin",
             Self::DeleteTenant { .. } => "delete_tenant",
             Self::PutRlsPolicy(_) => "put_rls_policy",
             Self::DeleteRlsPolicy { .. } => "delete_rls_policy",

nodedb/src/control/catalog_entry/post_apply/async_dispatch/dispatcher.rs

@@ -157,6 +157,7 @@ pub fn spawn_post_apply_async_side_effects(
         // PutContinuousAggregate / DeleteContinuousAggregate have their
         // own async branches above; they do not appear here.
         | CatalogEntry::PutTenant(_)
+        | CatalogEntry::PutTenantWithAdmin { .. }
         | CatalogEntry::DeleteTenant { .. }
         | CatalogEntry::PutRlsPolicy(_)
         | CatalogEntry::DeleteRlsPolicy { .. }

nodedb/src/control/catalog_entry/post_apply/gateway_invalidation.rs

@@ -164,7 +164,7 @@ pub(crate) fn invalidate_gateway_cache_for_entry(entry: &CatalogEntry, shared: &
         }
 
         // ── Tenant: identity does not affect plan shape ───────────────────────
-        CatalogEntry::PutTenant(_) => {
+        CatalogEntry::PutTenant(_) | CatalogEntry::PutTenantWithAdmin { .. } => {
             // no-op: tenant identity used for quota enforcement at exec time.
         }
         CatalogEntry::DeleteTenant { .. } => {

nodedb/src/control/catalog_entry/post_apply/sync.rs

@@ -147,6 +147,9 @@ pub fn apply_post_apply_side_effects_sync(entry: &CatalogEntry, shared: &Arc<Sha
         CatalogEntry::PutTenant(stored) => {
             tenant::put((**stored).clone(), Arc::clone(shared));
         }
+        CatalogEntry::PutTenantWithAdmin { tenant, admin } => {
+            tenant::put_with_admin((**tenant).clone(), (**admin).clone(), Arc::clone(shared));
+        }
         CatalogEntry::DeleteTenant { tenant_id } => {
             tenant::delete(*tenant_id, Arc::clone(shared));
         }

nodedb/src/control/catalog_entry/post_apply/tenant.rs

@@ -12,7 +12,7 @@
 
 use std::sync::Arc;
 
-use crate::control::security::catalog::{StoredCollection, StoredTenant};
+use crate::control::security::catalog::{StoredCollection, StoredTenant, StoredUser};
 use crate::control::security::tenant::TenantQuota;
 use crate::control::state::SharedState;
 use crate::types::TenantId;
@@ -35,6 +35,11 @@ pub fn put(stored: StoredTenant, shared: Arc<SharedState>) {
     );
 }
 
+pub fn put_with_admin(tenant: StoredTenant, admin: StoredUser, shared: Arc<SharedState>) {
+    shared.credentials.install_replicated_user(&admin, None);
+    put(tenant, shared);
+}
+
 pub fn delete(tenant_id: u64, shared: Arc<SharedState>) {
     let tid = TenantId::new(tenant_id);
     let mut tenants = match shared.tenants.lock() {

nodedb/src/control/cluster/metadata_applier/audit.rs

@@ -190,6 +190,7 @@ pub(super) fn describe_entry(e: &catalog_entry::CatalogEntry) -> (String, u64, S
         ),
         E::DeleteContinuousAggregate { name, .. } => (name.clone(), 0, String::new()),
         E::PutTenant(t) => (t.name.clone(), 0, String::new()),
+        E::PutTenantWithAdmin { tenant, admin } => (tenant.name.clone(), 0, admin.username.clone()),
         E::DeleteTenant { tenant_id, .. } => (tenant_id.to_string(), 0, String::new()),
         E::PutRlsPolicy(p) => (p.name.clone(), 0, String::new()),
         E::DeleteRlsPolicy { name, .. } => (name.clone(), 0, String::new()),