fix(tenant): enforce existence gate for unknown numeric tenant ids

Promote the `tenant_exists` helper from drop.rs into tenant/mod.rs so
DROP, ALTER, and PURGE all share a single, consistent existence check.

Previously a bogus numeric tenant id bypassed the name-not-found early
return: DROP proposed a delete for a phantom id, ALTER silently seeded a
default quota for it, and PURGE dispatched a destructive operation. Now
all three enforce the existence gate after the system-tenant guard (42501
keeps precedence), matching the behavior for unknown names — both paths
return 42704 or a no-op under IF EXISTS.

Expand name_resolution.rs regression coverage with: ALTER on a
quoted/empty name, full PURGE paths (success, IF EXISTS no-op, 42501,
catalog-unavailable), and unknown-numeric-id parity tests for
DROP/ALTER/PURGE.

Author: farhan-syah

nodedb/src/control/server/pgwire/ddl/tenant/alter.rs

@@ -39,12 +39,18 @@ pub fn alter_tenant(
     }
 
     // Accept either a numeric id or a tenant name (mirrors CREATE/SHOW/DROP).
-    let tenant_id = super::resolve_tenant_ref(state, parts[2])?.ok_or_else(|| {
-        sqlstate_error(
+    let tenant_id = super::resolve_tenant_ref(state, parts[2])?
+        .ok_or_else(|| sqlstate_error("42704", &format!("tenant '{}' does not exist", parts[2])))?;
+
+    // Existence gate, uniform across numeric ids and resolved names: altering an
+    // unknown tenant must error rather than silently seed a default quota for a
+    // phantom id.
+    if !super::tenant_exists(state, tenant_id)? {
+        return Err(sqlstate_error(
             "42704",
             &format!("tenant '{}' does not exist", parts[2]),
-        )
-    })?;
+        ));
+    }
 
     if !parts[3].eq_ignore_ascii_case("SET") || !parts[4].eq_ignore_ascii_case("QUOTA") {
         return Err(sqlstate_error(

nodedb/src/control/server/pgwire/ddl/tenant/drop.rs

@@ -15,28 +15,10 @@ use crate::control::metadata_proposer::propose_catalog_entry;
 use crate::control::security::audit::AuditEvent;
 use crate::control::security::identity::AuthenticatedIdentity;
 use crate::control::state::SharedState;
-use crate::types::TenantId;
 
 use super::super::super::types::sqlstate_error;
 use super::super::parse_utils::strip_if_exists;
-
-/// Whether a tenant id currently exists, consulting the redb catalog when
-/// one is wired up and falling back to the in-memory quota table otherwise.
-fn tenant_exists(state: &SharedState, tenant_id: TenantId) -> PgWireResult<bool> {
-    if let Some(catalog) = state.credentials.catalog() {
-        let present = catalog
-            .load_all_tenants()
-            .map_err(|e| sqlstate_error("XX000", &format!("catalog read: {e}")))?
-            .iter()
-            .any(|t| t.tenant_id == tenant_id.as_u64());
-        return Ok(present);
-    }
-    let tenants = match state.tenants.lock() {
-        Ok(t) => t,
-        Err(p) => p.into_inner(),
-    };
-    Ok(tenants.has_quota(tenant_id))
-}
+use super::tenant_exists;
 
 pub fn drop_tenant(
     state: &SharedState,
@@ -60,7 +42,9 @@ pub fn drop_tenant(
     }
 
     // Accept either a numeric id or a tenant name; mirror the existing
-    // CREATE TENANT name-resolution path.
+    // CREATE TENANT name-resolution path. A name that matches no tenant yields
+    // `None` here; an unknown numeric id resolves to a candidate that the
+    // existence gate below rejects, so both forms behave identically.
     let tenant_id = match super::resolve_tenant_ref(state, parts[2])? {
         Some(tid) => tid,
         None => {
@@ -80,11 +64,17 @@ pub fn drop_tenant(
         return Err(sqlstate_error("42501", "cannot drop system tenant (0)"));
     }
 
-    // `IF EXISTS`: dropping a tenant that does not exist is a no-op success.
-    // (Numeric-id-not-in-catalog path; the name-not-found case was handled
-    // above.)
-    if if_exists && !tenant_exists(state, tenant_id)? {
-        return Ok(vec![Response::Execution(Tag::new("DROP TENANT"))]);
+    // Existence gate, uniform across numeric ids and resolved names: an unknown
+    // tenant is a no-op under `IF EXISTS`, otherwise `42704` — never a silent
+    // delete proposal for a tenant that does not exist.
+    if !tenant_exists(state, tenant_id)? {
+        if if_exists {
+            return Ok(vec![Response::Execution(Tag::new("DROP TENANT"))]);
+        }
+        return Err(sqlstate_error(
+            "42704",
+            &format!("tenant '{}' does not exist", parts[2]),
+        ));
     }
 
     let entry = CatalogEntry::DeleteTenant { tenant_id: tid };

nodedb/src/control/server/pgwire/ddl/tenant/mod.rs

@@ -44,8 +44,11 @@ use super::super::types::sqlstate_error;
 ///
 /// The numeric form is the legacy fast path and requires no catalog access;
 /// any `u64`-parseable token returns `Ok(Some(TenantId::new(id)))` whether or
-/// not that id currently exists (callers that need existence still rely on
-/// their own `tenant_exists`-style check).
+/// not that id currently exists. Resolving a token to an id does **not** assert
+/// the tenant exists — callers must gate the operation through [`tenant_exists`]
+/// so that an unknown numeric id and an unknown name behave identically (both
+/// `42704`, or an `IF EXISTS` no-op). Skipping that check reintroduces the
+/// id/name asymmetry where a bogus numeric id silently "succeeds".
 ///
 /// A non-numeric token is treated as a tenant name and resolved via
 /// [`find_tenant_by_name`]. Single-quoted names are unwrapped, mirroring the
@@ -92,3 +95,24 @@ pub(super) fn resolve_tenant_ref(
         .map_err(|e| sqlstate_error("XX000", &format!("catalog read: {e}")))?
         .map(|stored| TenantId::new(stored.tenant_id)))
 }
+
+/// Whether `tenant_id` currently exists, consulting the redb catalog when one
+/// is wired up and falling back to the in-memory quota table otherwise.
+///
+/// Shared by `DROP`, `ALTER`, and `PURGE TENANT` so existence is enforced the
+/// same way for numeric ids and resolved names — see [`resolve_tenant_ref`].
+pub(super) fn tenant_exists(state: &SharedState, tenant_id: TenantId) -> PgWireResult<bool> {
+    if let Some(catalog) = state.credentials.catalog() {
+        let present = catalog
+            .load_all_tenants()
+            .map_err(|e| sqlstate_error("XX000", &format!("catalog read: {e}")))?
+            .iter()
+            .any(|t| t.tenant_id == tenant_id.as_u64());
+        return Ok(present);
+    }
+    let tenants = match state.tenants.lock() {
+        Ok(t) => t,
+        Err(p) => p.into_inner(),
+    };
+    Ok(tenants.has_quota(tenant_id))
+}

nodedb/src/control/server/pgwire/ddl/tenant/purge.rs

@@ -37,18 +37,23 @@ pub async fn purge_tenant(
     }
 
     // Accept either a numeric id or a tenant name (mirrors CREATE/SHOW/DROP).
-    let tenant_id = super::resolve_tenant_ref(state, parts[2])?.ok_or_else(|| {
-        sqlstate_error(
-            "42704",
-            &format!("tenant '{}' does not exist", parts[2]),
-        )
-    })?;
+    let tenant_id = super::resolve_tenant_ref(state, parts[2])?
+        .ok_or_else(|| sqlstate_error("42704", &format!("tenant '{}' does not exist", parts[2])))?;
     let tid = tenant_id.as_u64();
 
     if tid == 0 {
         return Err(sqlstate_error("42501", "cannot purge system tenant (0)"));
     }
 
+    // Existence gate, uniform across numeric ids and resolved names: refuse to
+    // dispatch the destructive meta op for a tenant that does not exist.
+    if !super::tenant_exists(state, tenant_id)? {
+        return Err(sqlstate_error(
+            "42704",
+            &format!("tenant '{}' does not exist", parts[2]),
+        ));
+    }
+
     if !parts[3].eq_ignore_ascii_case("CONFIRM") {
         return Err(sqlstate_error(
             "42601",

nodedb/tests/pgwire_auth_tenants/name_resolution.rs

@@ -8,7 +8,9 @@
 //! shared `resolve_tenant_ref` helper, mirroring the already-shipped
 //! `CREATE TENANT <name>` / `SHOW TENANT <name>` paths.
 
-use crate::common::pgwire_auth_helpers::{ddl_err, ddl_ok, make_state_with_catalog, superuser};
+use crate::common::pgwire_auth_helpers::{
+    ddl_err, ddl_ok, make_state, make_state_with_catalog, superuser,
+};
 
 // ─── DROP TENANT by name ─────────────────────────────────────────────────────
 
@@ -89,12 +91,7 @@ async fn alter_tenant_by_numeric_id_still_works() {
     let su = superuser();
 
     ddl_ok(&state, &su, "CREATE TENANT acme_alter_num ID 7145").await;
-    ddl_ok(
-        &state,
-        &su,
-        "ALTER TENANT 7145 SET QUOTA max_qps = 250",
-    )
-    .await;
+    ddl_ok(&state, &su, "ALTER TENANT 7145 SET QUOTA max_qps = 250").await;
 }
 
 /// `ALTER TENANT <name> SET QUOTA ...` — name resolves to id.
@@ -129,3 +126,205 @@ async fn alter_tenant_unknown_name_errors() {
         "expected 42704/does not exist, got: {err}"
     );
 }
+
+/// `ALTER TENANT '<name>' SET QUOTA ...` — single-quoted name resolves to id,
+/// matching the quoted-name path already covered on DROP.
+#[tokio::test]
+async fn alter_tenant_by_quoted_name() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    ddl_ok(&state, &su, "CREATE TENANT acme_alter_quoted ID 7147").await;
+    ddl_ok(
+        &state,
+        &su,
+        "ALTER TENANT 'acme_alter_quoted' SET QUOTA max_qps = 250",
+    )
+    .await;
+}
+
+/// `ALTER TENANT '' SET QUOTA ...` (empty quoted name) → `42601`, the same
+/// resolver guard exercised on DROP.
+#[tokio::test]
+async fn alter_tenant_empty_name_errors() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    let err = ddl_err(&state, &su, "ALTER TENANT '' SET QUOTA max_qps = 250").await;
+    assert!(
+        err.contains("42601") && err.contains("numeric id or a tenant name"),
+        "expected 42601 empty-name error, got: {err}"
+    );
+}
+
+// ─── PURGE TENANT by name ────────────────────────────────────────────────────
+//
+// `purge_tenant` resolves the tenant reference, rejects the system tenant, and
+// only then dispatches the destructive meta op to the Data Plane. The
+// resolution and guard branches all return before that dispatch, so they are
+// covered here without a Data Plane; the post-dispatch happy path is exercised
+// by the executor-level purge tests.
+
+/// `PURGE TENANT <name> CONFIRM` on an unknown name → `42704`, before any
+/// destructive dispatch.
+#[tokio::test]
+async fn purge_tenant_unknown_name_errors() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    let err = ddl_err(&state, &su, "PURGE TENANT no_such_tenant CONFIRM").await;
+    assert!(
+        err.contains("does not exist") && err.contains("42704"),
+        "expected 42704/does not exist, got: {err}"
+    );
+}
+
+/// `PURGE TENANT '' CONFIRM` (empty quoted name) → `42601` resolver guard.
+#[tokio::test]
+async fn purge_tenant_empty_name_errors() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    let err = ddl_err(&state, &su, "PURGE TENANT '' CONFIRM").await;
+    assert!(
+        err.contains("42601") && err.contains("numeric id or a tenant name"),
+        "expected 42601 empty-name error, got: {err}"
+    );
+}
+
+/// `PURGE TENANT <name>` with a malformed confirmation token resolves the name
+/// first, then fails the `CONFIRM` gate with `42601`. The `42601`/CONFIRM error
+/// (rather than `42704`) proves the name resolved — covering the by-name purge
+/// path up to the Data Plane boundary.
+#[tokio::test]
+async fn purge_tenant_by_name_resolves_then_requires_confirm() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    ddl_ok(&state, &su, "CREATE TENANT acme_purge_name ID 7148").await;
+    let err = ddl_err(&state, &su, "PURGE TENANT acme_purge_name PLEASE").await;
+    assert!(
+        err.contains("42601") && err.contains("CONFIRM"),
+        "expected name to resolve then hit the CONFIRM gate, got: {err}"
+    );
+}
+
+/// `PURGE TENANT <id>` (numeric) for an existing tenant with a malformed
+/// confirmation token — the legacy numeric path resolves and passes the
+/// existence gate, then fails the `CONFIRM` gate. Regression that the resolver
+/// refactor kept the numeric fast path intact for purge.
+#[tokio::test]
+async fn purge_tenant_by_numeric_id_resolves_then_requires_confirm() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    ddl_ok(&state, &su, "CREATE TENANT acme_purge_num ID 7149").await;
+    let err = ddl_err(&state, &su, "PURGE TENANT 7149 PLEASE").await;
+    assert!(
+        err.contains("42601") && err.contains("CONFIRM"),
+        "expected numeric id to resolve then hit the CONFIRM gate, got: {err}"
+    );
+}
+
+// ─── Unknown numeric id parity (id form must match name form) ─────────────────
+//
+// A numeric id that matches no tenant must behave exactly like an unknown
+// name: `42704`, or an `IF EXISTS` no-op for DROP. Before the existence gate
+// these silently proceeded (DROP proposed a delete, ALTER seeded a default
+// quota, PURGE dispatched a destructive op) — the id/name asymmetry.
+
+/// `DROP TENANT <unknown_id>` without `IF EXISTS` → `42704`, matching the
+/// unknown-name behavior.
+#[tokio::test]
+async fn drop_tenant_unknown_numeric_id_without_if_exists_errors() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    let err = ddl_err(&state, &su, "DROP TENANT 999001").await;
+    assert!(
+        err.contains("42704") && err.contains("does not exist"),
+        "expected 42704 for unknown numeric id, got: {err}"
+    );
+}
+
+/// `DROP TENANT IF EXISTS <unknown_id>` is a no-op success, matching the
+/// unknown-name `IF EXISTS` behavior.
+#[tokio::test]
+async fn drop_tenant_if_exists_unknown_numeric_id_is_noop() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    ddl_ok(&state, &su, "DROP TENANT IF EXISTS 999002").await;
+}
+
+/// `ALTER TENANT <unknown_id> SET QUOTA ...` → `42704`, never a silent
+/// default-quota seed for a phantom id.
+#[tokio::test]
+async fn alter_tenant_unknown_numeric_id_errors() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    let err = ddl_err(&state, &su, "ALTER TENANT 999003 SET QUOTA max_qps = 250").await;
+    assert!(
+        err.contains("42704") && err.contains("does not exist"),
+        "expected 42704 for unknown numeric id, got: {err}"
+    );
+}
+
+/// `PURGE TENANT <unknown_id> CONFIRM` → `42704`, never a destructive dispatch
+/// for a tenant that does not exist.
+#[tokio::test]
+async fn purge_tenant_unknown_numeric_id_errors() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    let err = ddl_err(&state, &su, "PURGE TENANT 999004 CONFIRM").await;
+    assert!(
+        err.contains("42704") && err.contains("does not exist"),
+        "expected 42704 for unknown numeric id, got: {err}"
+    );
+}
+
+// ─── System tenant + catalog-unavailable guards ──────────────────────────────
+
+/// `DROP TENANT 0` — the system tenant is protected with `42501` regardless of
+/// the resolver refactor.
+#[tokio::test]
+async fn drop_system_tenant_numeric_errors() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    let err = ddl_err(&state, &su, "DROP TENANT 0").await;
+    assert!(
+        err.contains("42501") && err.contains("system tenant"),
+        "expected 42501 system-tenant guard, got: {err}"
+    );
+}
+
+/// `PURGE TENANT 0 CONFIRM` — the system tenant cannot be purged.
+#[tokio::test]
+async fn purge_system_tenant_numeric_errors() {
+    let state = make_state_with_catalog();
+    let su = superuser();
+
+    let err = ddl_err(&state, &su, "PURGE TENANT 0 CONFIRM").await;
+    assert!(
+        err.contains("42501") && err.contains("system tenant"),
+        "expected 42501 system-tenant guard, got: {err}"
+    );
+}
+
+/// `DROP TENANT <name>` with no catalog wired up → `42601`, directing the
+/// caller to use a numeric id. Exercises the catalog-unavailable branch of
+/// `resolve_tenant_ref` that the catalog-backed fixtures cannot reach.
+#[tokio::test]
+async fn drop_tenant_by_name_without_catalog_errors() {
+    let state = make_state();
+    let su = superuser();
+
+    let err = ddl_err(&state, &su, "DROP TENANT some_tenant_name").await;
+    assert!(
+        err.contains("42601") && err.contains("catalog"),
+        "expected 42601 catalog-unavailable error, got: {err}"
+    );
+}