fix(tenant): reject duplicate CREATE TENANT names

A duplicate tenant name previously succeeded as a silent no-op, minting
a second tenant id that shares the name. That makes every by-name
lookup (ownership fallback, admin provisioning, DROP TENANT) ambiguous
and strands the older tenant's objects. Now a duplicate name errors
unless IF NOT EXISTS is given, in which case it remains a no-op.

Author: farhan-syah

nodedb/src/control/cluster/recovery_check/integrity.rs

@@ -23,12 +23,29 @@
 //! - Every `StoredRlsPolicy.collection` exists as a
 //!   `StoredCollection` row.
 //!
-//! None of these are auto-repaired. Redb is not the source of
-//! truth — the raft log is — and the safe recovery for any
-//! redb corruption is "re-run the applier from the log",
-//! which is the operator's job. The integrity check reports
-//! every violation and the sanity-check wrapper aborts
-//! startup on any non-empty violation list.
+//! This module only *detects*; it never repairs. Redb is not the
+//! source of truth — the raft log is — and the general recovery
+//! for redb corruption is "re-run the applier from the log",
+//! which is the operator's job.
+//!
+//! Three ownership/grant classes are healed before the abort gate,
+//! because they are reachable from ordinary DDL rather than from
+//! storage corruption, and each would otherwise leave an existing
+//! data directory permanently unbootable with no repair path.
+//! `verify_and_repair` runs `repair_integrity::heal_orphan_rows`
+//! to a fixpoint over this module's output:
+//!
+//! - a primary row whose `StoredOwner` is missing — the owner row
+//!   is rebuilt from the primary's in-band owner;
+//! - `owner(...)` → `user(...)` dangling — the owner row is
+//!   restored from the primary's in-band owner when that user
+//!   still exists, otherwise reassigned to the tenant's resolved
+//!   administrative principal;
+//! - `permission(...)` → `user(...)` dangling — the grant is
+//!   revoked, since a grant to a nonexistent user confers nothing.
+//!
+//! Every other violation above is reported and left alone.
+//! Startup aborts only on violations that survive the pass.
 
 use std::collections::HashSet;
 

nodedb/src/control/server/shared/ddl/neutral/tenant/create.rs

@@ -98,17 +98,22 @@ pub fn create_tenant(
     let name = parts[2];
     let opts = parse_tenant_options(&parts[3..])?;
 
-    // `IF NOT EXISTS`: if a tenant with this name already exists, the
-    // statement is a no-op success — do not allocate a second id.
-    if if_not_exists
-        && state
-            .credentials
-            .catalog()
-            .find_tenant_by_name(name)
-            .map_err(|e| ddl_err("XX000", format!("catalog read: {e}")))?
-            .is_some()
+    // Tenant names are unique. A duplicate is a no-op success under
+    // `IF NOT EXISTS` and an error otherwise — never a second tenant id
+    // sharing the name, which would make the name ambiguous for every
+    // by-name lookup (ownership fallback, admin provisioning, DROP TENANT)
+    // and silently strand the older tenant's objects.
+    if state
+        .credentials
+        .catalog()
+        .find_tenant_by_name(name)
+        .map_err(|e| ddl_err("XX000", format!("catalog read: {e}")))?
+        .is_some()
     {
-        return Ok(status("CREATE TENANT"));
+        if if_not_exists {
+            return Ok(status("CREATE TENANT"));
+        }
+        return Err(ddl_err("42710", format!("tenant '{name}' already exists")));
     }
 
     // Pick the tenant id. An explicit `ID <n>` is honored verbatim;

nodedb/tests/pgwire_tenant_scoping.rs

@@ -574,3 +574,79 @@ async fn show_tenants_includes_tenant_name() {
         "SHOW TENANTS must report the tenant name; row was {acme:?}"
     );
 }
+
+/// Tenant names are unique: a second `CREATE TENANT` with the same name is
+/// rejected rather than minting a second tenant id sharing that name. A
+/// duplicate name makes every by-name lookup ambiguous (ownership fallback,
+/// admin provisioning, `DROP TENANT`) and strands the older tenant's objects.
+#[tokio::test]
+async fn duplicate_tenant_name_is_rejected() {
+    let server = TestServer::start().await;
+    server
+        .exec("CREATE TENANT probe")
+        .await
+        .expect("first CREATE TENANT must succeed");
+
+    let err = server
+        .exec("CREATE TENANT probe")
+        .await
+        .expect_err("duplicate tenant name must be rejected");
+    assert!(
+        err.contains("already exists"),
+        "duplicate CREATE TENANT must report the name conflict, got: {err}"
+    );
+
+    // IF NOT EXISTS remains a no-op success on the same name.
+    server
+        .exec("CREATE TENANT IF NOT EXISTS probe")
+        .await
+        .expect("CREATE TENANT IF NOT EXISTS on an existing name must succeed");
+
+    let rows = server
+        .query_named_rows("SHOW TENANTS")
+        .await
+        .expect("SHOW TENANTS");
+    let named: Vec<_> = rows
+        .iter()
+        .filter(|r| r.get("name").map(String::as_str) == Some("probe"))
+        .collect();
+    assert_eq!(
+        named.len(),
+        1,
+        "exactly one tenant may carry the name 'probe'; rows were {named:?}"
+    );
+}
+
+/// A superuser can administer a pre-existing tenant-homed collection it does
+/// not own by switching the session's effective tenant. Without `SET TENANT`
+/// the superuser is scoped to its own tenant and the collection is correctly
+/// invisible; with it, ownership is satisfied by superuser status.
+#[tokio::test]
+async fn superuser_can_drop_tenant_homed_collection_after_set_tenant() {
+    let server = TestServer::start().await;
+    bootstrap_tenant_user(&server, "acme_user", "acme_notes").await;
+
+    // Unqualified, the superuser session does not see the tenant's collection.
+    let unscoped = server.exec("DROP COLLECTION acme_notes").await;
+    assert!(
+        unscoped.is_err(),
+        "superuser must not resolve a tenant-homed collection without SET TENANT"
+    );
+
+    let (svc, _h) = server
+        .connect_as("nodedb", "nodedb")
+        .await
+        .expect("superuser connect");
+    svc.simple_query("SET TENANT = 'acme'")
+        .await
+        .expect("superuser SET TENANT");
+    svc.simple_query("DROP COLLECTION acme_notes")
+        .await
+        .expect("superuser must drop a tenant-homed collection it does not own");
+
+    let remaining = svc
+        .simple_query("SELECT id FROM acme_notes")
+        .await
+        .expect_err("dropped collection must no longer resolve");
+    let _ = remaining;
+}