fix(tenant): replace count-based id allocation with a durable high-water-mark

farhan-syah · farhan-syah · commit ef8b59974d12 · 2026-06-11T06:07:09.000+08:00
The previous allocator derived a new tenant id from the number of active
tenants. Two back-to-back CREATE TENANTs with no intervening traffic both
computed id 1, causing the second insert to silently overwrite the first.
A dropped tenant also freed its slot for reuse, allowing later tenants to
inherit the dropped id and pick up any residual permissions or quota rows
tied to it.

Introduce a dedicated redb table (tenant_id_hwm) that records the highest
tenant id ever written. The counter is bumped in the same transaction as
the catalog row insertion, so the high-water-mark and the catalog stay
consistent under crashes and Raft-replicated applies alike. Dropped ids
are never reissued because the counter only moves forward.

The in-memory TenantIsolation path (no-catalog / unit-test) gains a
matching id_hwm field that shadows the durable counter, and a new
allocate_tenant_id() method that respects the reserved system (0) and
bootstrap (1) slots, ensuring auto-allocated ids always start at 2.

CREATE TENANT routing is updated to prefer the durable catalog allocator
when one is wired up, and fall back to the in-memory mirror otherwise.
The default_admin_username helper is extracted to a shared function so
the DROP TENANT path can reference it without duplicating the convention.
diff --git a/nodedb/src/control/security/catalog/metadata.rs b/nodedb/src/control/security/catalog/metadata.rs
@@ -2,6 +2,9 @@
 
 //! Metadata counter and tenant operations for the system catalog.
 
+use redb::ReadableTable;
+
+use super::tenant_id_hwm::{HWM_KEY, TENANT_ID_HWM};
 use super::types::{METADATA, StoredTenant, SystemCatalog, TENANTS, catalog_err};
 
 impl SystemCatalog {
@@ -70,6 +73,24 @@ impl SystemCatalog {
                 .insert(key.as_str(), bytes.as_slice())
                 .map_err(|e| catalog_err("insert tenant", e))?;
         }
+        // Advance the durable tenant-id high-water-mark in the same
+        // transaction so the id is never reissued by a later
+        // auto-allocation — covers explicitly-chosen ids and ids
+        // replicated from a leader on follower-side applies alike.
+        {
+            let mut hwm = write_txn
+                .open_table(TENANT_ID_HWM)
+                .map_err(|e| catalog_err("open tenant_id_hwm", e))?;
+            let cur = hwm
+                .get(HWM_KEY)
+                .map_err(|e| catalog_err("get tenant_id_hwm", e))?
+                .map(|v| v.value())
+                .unwrap_or(0);
+            if tenant.tenant_id > cur {
+                hwm.insert(HWM_KEY, tenant.tenant_id)
+                    .map_err(|e| catalog_err("insert tenant_id_hwm", e))?;
+            }
+        }
         write_txn.commit().map_err(|e| catalog_err("commit", e))
     }
 
diff --git a/nodedb/src/control/security/catalog/mod.rs b/nodedb/src/control/security/catalog/mod.rs
@@ -50,6 +50,7 @@ pub mod surrogate_pk;
 pub mod synonym_groups;
 pub mod system_catalog;
 pub mod tables;
+pub mod tenant_id_hwm;
 pub mod tenant_quotas;
 pub mod topics;
 pub mod trigger_types;
diff --git a/nodedb/src/control/security/catalog/tenant_id_hwm.rs b/nodedb/src/control/security/catalog/tenant_id_hwm.rs
@@ -0,0 +1,165 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Durable tenant-id allocator for the `_system.tenant_id_hwm` table.
+//!
+//! Singleton table — one row keyed `"global"` holding the highest
+//! tenant id ever allocated. The high-water-mark is monotonic: it is
+//! advanced on every tenant write and never lowered by `DROP TENANT`,
+//! so a tenant id is never reused for a different tenant across a drop
+//! or a restart. This is the same durable-counter idiom the global
+//! surrogate identity uses; see [`super::surrogate_hwm`].
+//!
+//! Auto-allocation derives the next id from the max of this counter
+//! and the ids actually present in `_system.tenants`, so a database
+//! whose tenants predate the counter self-heals on first allocation
+//! instead of colliding with an existing tenant.
+
+use redb::ReadableTable;
+
+use super::types::{SystemCatalog, TENANTS, catalog_err};
+
+/// Redb table: singleton `"global"` -> highest allocated tenant id (`u64`).
+pub(super) const TENANT_ID_HWM: redb::TableDefinition<&str, u64> =
+    redb::TableDefinition::new("_system.tenant_id_hwm");
+
+/// Singleton row key.
+pub(super) const HWM_KEY: &str = "global";
+
+/// Lowest id handed to an auto-allocated tenant. `0` is the system
+/// tenant and `1` is the bootstrap/default tenant; user tenants start
+/// at `2` so neither reserved slot is ever handed out by allocation.
+pub(crate) const FIRST_USER_TENANT_ID: u64 = 2;
+
+impl SystemCatalog {
+    /// Atomically allocate the next tenant id.
+    ///
+    /// Durable and monotonic: the returned id is strictly greater than
+    /// every id previously allocated or currently stored, and the
+    /// counter is never reused after a `DROP TENANT`. Self-heals on
+    /// first use by taking the max of the stored hwm and the highest id
+    /// in `_system.tenants`, so tenants created before this counter
+    /// existed can never collide with a fresh allocation.
+    pub fn allocate_tenant_id(&self) -> crate::Result<u64> {
+        let write_txn = self
+            .db
+            .begin_write()
+            .map_err(|e| catalog_err("tenant_id_hwm write txn", e))?;
+        let next;
+        {
+            // Highest id currently materialised in the tenants table.
+            let mut max_existing = 0u64;
+            {
+                let tenants = write_txn
+                    .open_table(TENANTS)
+                    .map_err(|e| catalog_err("open tenants", e))?;
+                for entry in tenants
+                    .range::<&str>(..)
+                    .map_err(|e| catalog_err("range tenants", e))?
+                {
+                    let (key, _) = entry.map_err(|e| catalog_err("read tenant", e))?;
+                    if let Ok(id) = key.value().parse::<u64>() {
+                        max_existing = max_existing.max(id);
+                    }
+                }
+            }
+            let mut hwm = write_txn
+                .open_table(TENANT_ID_HWM)
+                .map_err(|e| catalog_err("open tenant_id_hwm", e))?;
+            let stored = hwm
+                .get(HWM_KEY)
+                .map_err(|e| catalog_err("get tenant_id_hwm", e))?
+                .map(|v| v.value())
+                .unwrap_or(0);
+            next = stored.max(max_existing).max(FIRST_USER_TENANT_ID - 1) + 1;
+            hwm.insert(HWM_KEY, next)
+                .map_err(|e| catalog_err("insert tenant_id_hwm", e))?;
+        }
+        write_txn
+            .commit()
+            .map_err(|e| catalog_err("tenant_id_hwm commit", e))?;
+        Ok(next)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::control::security::catalog::StoredTenant;
+
+    fn open() -> (tempfile::TempDir, SystemCatalog) {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("system.redb");
+        let catalog = SystemCatalog::open(&path).unwrap();
+        (dir, catalog)
+    }
+
+    #[test]
+    fn first_allocation_skips_reserved_slots() {
+        let (_dir, catalog) = open();
+        assert_eq!(catalog.allocate_tenant_id().unwrap(), FIRST_USER_TENANT_ID);
+    }
+
+    #[test]
+    fn allocations_are_strictly_increasing_and_distinct() {
+        let (_dir, catalog) = open();
+        let a = catalog.allocate_tenant_id().unwrap();
+        let b = catalog.allocate_tenant_id().unwrap();
+        let c = catalog.allocate_tenant_id().unwrap();
+        assert!(a < b && b < c, "{a} {b} {c}");
+    }
+
+    #[test]
+    fn never_reuses_id_after_delete() {
+        let (_dir, catalog) = open();
+        let id = catalog.allocate_tenant_id().unwrap();
+        catalog
+            .put_tenant(&StoredTenant {
+                tenant_id: id,
+                name: "t".into(),
+                created_at: 0,
+                is_active: true,
+            })
+            .unwrap();
+        catalog.delete_tenant(id).unwrap();
+        // The dropped id must not be handed out again.
+        assert!(catalog.allocate_tenant_id().unwrap() > id);
+    }
+
+    #[test]
+    fn self_heals_against_preexisting_tenants() {
+        let (_dir, catalog) = open();
+        // A tenant row written without ever bumping the counter (the
+        // pre-fix world): insert directly so the hwm stays at 0, then
+        // assert allocation still skips past the existing id.
+        let bytes = zerompk::to_msgpack_vec(&StoredTenant {
+            tenant_id: 500,
+            name: "legacy".into(),
+            created_at: 0,
+            is_active: true,
+        })
+        .unwrap();
+        let txn = catalog.db.begin_write().unwrap();
+        {
+            let mut t = txn.open_table(TENANTS).unwrap();
+            t.insert("500", bytes.as_slice()).unwrap();
+        }
+        txn.commit().unwrap();
+        assert!(catalog.allocate_tenant_id().unwrap() > 500);
+    }
+
+    #[test]
+    fn put_tenant_with_explicit_id_advances_hwm() {
+        let (_dir, catalog) = open();
+        // An explicitly-chosen high id must push the hwm forward so a
+        // later auto-allocation never collides with it.
+        catalog
+            .put_tenant(&StoredTenant {
+                tenant_id: 10,
+                name: "explicit".into(),
+                created_at: 0,
+                is_active: true,
+            })
+            .unwrap();
+        assert_eq!(catalog.allocate_tenant_id().unwrap(), 11);
+    }
+}
diff --git a/nodedb/src/control/security/tenant.rs b/nodedb/src/control/security/tenant.rs
@@ -102,6 +102,11 @@ pub struct TenantIsolation {
     usage: HashMap<TenantId, TenantUsage>,
     /// Default quota applied to tenants without explicit config.
     default_quota: TenantQuota,
+    /// Monotonic high-water-mark of every tenant id observed, used to
+    /// allocate fresh ids without a catalog (the embedded / unit path).
+    /// The catalog's durable `tenant_id_hwm` is authoritative when one
+    /// is wired up; this mirror only serves the no-catalog fallback.
+    id_hwm: u64,
 }
 
 impl TenantIsolation {
@@ -110,14 +115,29 @@ impl TenantIsolation {
             quotas: HashMap::new(),
             usage: HashMap::new(),
             default_quota,
+            id_hwm: 0,
         }
     }
 
     /// Set quota for a specific tenant.
     pub fn set_quota(&mut self, tenant_id: TenantId, quota: TenantQuota) {
+        self.id_hwm = self.id_hwm.max(tenant_id.as_u64());
         self.quotas.insert(tenant_id, quota);
     }
 
+    /// Allocate the next tenant id from the in-memory high-water-mark,
+    /// skipping the reserved system (`0`) and bootstrap (`1`) slots.
+    /// Used only when no catalog is wired up; with a catalog,
+    /// `SystemCatalog::allocate_tenant_id` is the durable authority.
+    pub fn allocate_tenant_id(&mut self) -> u64 {
+        let next = self
+            .id_hwm
+            .max(crate::control::security::catalog::tenant_id_hwm::FIRST_USER_TENANT_ID - 1)
+            + 1;
+        self.id_hwm = next;
+        next
+    }
+
     /// Whether a tenant already has an explicit quota record.
     pub fn has_quota(&self, tenant_id: TenantId) -> bool {
         self.quotas.contains_key(&tenant_id)
diff --git a/nodedb/src/control/server/pgwire/ddl/tenant/create.rs b/nodedb/src/control/server/pgwire/ddl/tenant/create.rs
@@ -54,6 +54,15 @@ fn parse_tenant_options<'a>(rest: &[&'a str]) -> PgWireResult<TenantOptions<'a>>
     })
 }
 
+/// The default username auto-provisioned as a tenant's `tenant_admin`
+/// when `CREATE TENANT` is run without an explicit `WITH ADMIN <user>`
+/// clause. Defined once here so the `DROP TENANT` cleanup path can
+/// identify and remove this lifecycle-owned account without the
+/// convention drifting between the two sites.
+pub(super) fn default_admin_username(tenant_name: &str) -> String {
+    format!("{tenant_name}_admin")
+}
+
 /// `CREATE TENANT [IF NOT EXISTS] <name> [ID <id>] [WITH ADMIN <user>]`
 ///
 /// Creates a tenant with default quotas. Only superuser can create tenants.
@@ -95,21 +104,27 @@ pub fn create_tenant(
         return Ok(vec![Response::Execution(Tag::new("CREATE TENANT"))]);
     }
 
-    // Pick the tenant id under a short lock scope; do NOT mutate the
-    // store yet — the post_apply side effect on every node seeds the
-    // default quota when `PutTenant` commits.
-    let tenant_id = {
-        let tenants = match state.tenants.lock() {
-            Ok(t) => t,
-            Err(p) => p.into_inner(),
-        };
-        match opts.explicit_id {
-            Some(id) => TenantId::new(id),
+    // Pick the tenant id. An explicit `ID <n>` is honored verbatim;
+    // otherwise allocate a fresh id from the durable, monotonic
+    // high-water-mark so two `CREATE TENANT`s never collide and a
+    // dropped id is never reused. The catalog counter is authoritative
+    // when wired up; the in-memory mirror covers the no-catalog path.
+    let tenant_id = match opts.explicit_id {
+        Some(id) => TenantId::new(id),
+        None => match state.credentials.catalog().as_ref() {
+            Some(catalog) => TenantId::new(
+                catalog
+                    .allocate_tenant_id()
+                    .map_err(|e| sqlstate_error("XX000", &format!("tenant id alloc: {e}")))?,
+            ),
             None => {
-                let count = tenants.tenant_count() as u64;
-                TenantId::new(count + 1)
+                let mut tenants = match state.tenants.lock() {
+                    Ok(t) => t,
+                    Err(p) => p.into_inner(),
+                };
+                TenantId::new(tenants.allocate_tenant_id())
             }
-        }
+        },
     };
 
     let now = std::time::SystemTime::now()
@@ -148,7 +163,7 @@ pub fn create_tenant(
     let admin_name = opts
         .admin_override
         .map(str::to_string)
-        .unwrap_or_else(|| format!("{name}_admin"));
+        .unwrap_or_else(|| default_admin_username(name));
     let admin_password = {
         use sha2::{Digest, Sha256};
         let mut hasher = Sha256::new();
