fix(control): classify native error frames authored as a bare SQLSTATE

nodedb-types/src/error/sqlstate.rs

@@ -90,6 +90,11 @@ pub const TYPE_GUARD_VIOLATION: &str = "23608";
 /// `28000` — `invalid_authorization_specification` (no valid credentials)
 pub const INVALID_AUTHORIZATION: &str = "28000";
 
+// ── Class 3D — Invalid Catalog Name ──────────────────────────────────────────
+
+/// `3D000` — `invalid_catalog_name` (the selected database does not exist)
+pub const INVALID_CATALOG_NAME: &str = "3D000";
+
 // ── Class 40 — Transaction Rollback ──────────────────────────────────────────
 
 /// `40001` — `serialization_failure` (write conflict; client should retry)

nodedb/src/control/server/native/dispatch/conversion.rs

@@ -7,6 +7,7 @@ use nodedb_types::conversion::json_to_value_ref;
 use nodedb_types::protocol::NativeResponse;
 
 use crate::bridge::envelope::Response;
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 use crate::control::server::response_shape::types::ShapedRows;
 use crate::control::server::shared::ddl::sqlstate::error_code_to_sqlstate;
 use crate::control::server::shared::ddl::{DdlError, DdlResult};
@@ -52,6 +53,29 @@ pub(crate) fn error_to_native(seq: u64, e: &crate::Error) -> NativeResponse {
     NativeResponse::error_with_code(seq, code, message, ndb_code)
 }
 
+/// Convert a Control-Plane error into a native error frame under a SQLSTATE
+/// the call site chooses.
+///
+/// Same classification as [`error_to_native`] — the numeric code comes from
+/// the one `Error` mapping table — but for the guards that render a more
+/// specific SQLSTATE than the error's own variant implies: a plan that cannot
+/// be built is `42601` to a SQL client whatever its internal cause, and an
+/// RLS injection failure is `42501`. Those sites still hold the classified
+/// error, so the code must come from it rather than be inferred back out of
+/// the SQLSTATE they just chose.
+pub(crate) fn error_to_native_with_sqlstate(
+    seq: u64,
+    sqlstate: impl Into<String>,
+    e: &crate::Error,
+) -> NativeResponse {
+    NativeResponse::error_with_code(
+        seq,
+        sqlstate,
+        e.to_string(),
+        crate::error_classify::classify(e).code().0,
+    )
+}
+
 /// Convert a `NodeDbError` produced while shaping a response into a
 /// NativeResponse error frame.
 ///
@@ -93,7 +117,7 @@ pub(crate) fn error_code_to_native(
     code: Option<&crate::bridge::envelope::ErrorCode>,
 ) -> NativeResponse {
     let Some(code) = code else {
-        return NativeResponse::error(seq, "XX000", "unknown data plane error");
+        return sqlstate_error(seq, "XX000", "unknown data plane error");
     };
     let (_, sqlstate, message) = error_code_to_sqlstate(code);
     let public = nodedb_types::NodeDbError::from(crate::Error::DataPlane(code.clone()));
@@ -108,12 +132,18 @@ pub(crate) fn error_code_to_native(
 /// row-returning / status / empty result determines the response (a status tag
 /// becomes a single-column status row, a row result becomes a columns+rows
 /// frame, an empty result or an empty vec becomes a bare OK).
+///
+/// `DdlError` is authored as a SQLSTATE and a message and never holds a
+/// classified `Error`, so the numeric code the client rebuilds its typed error
+/// from comes from `sqlstate_code`. Without it every DDL refusal — a `DROP
+/// TABLE` naming a collection that does not exist, a denied `GRANT` — reaches
+/// the client as a generic internal failure.
 pub(crate) fn ddl_result_to_native(
     seq: u64,
     result: Result<Vec<DdlResult>, DdlError>,
 ) -> NativeResponse {
     match result {
-        Err(DdlError { sqlstate, message }) => NativeResponse::error(seq, sqlstate, message),
+        Err(DdlError { sqlstate, message }) => sqlstate_error(seq, sqlstate, message),
         // Unknown pgwire response variants are dropped during translation, so
         // the first element is the first meaningful result — mirroring the
         // previous bridge, which returned on the first known variant.
@@ -313,6 +343,57 @@ mod tests {
         );
     }
 
+    /// A DDL refusal is authored as a SQLSTATE with no `Error` behind it, so
+    /// its numeric code comes from the SQLSTATE table. Without it the frame
+    /// ships `ndb_code == 0` and a `DROP TABLE` naming an absent collection
+    /// arrives as a generic internal failure while the identical `SELECT`
+    /// arrives typed.
+    #[test]
+    fn ddl_refusals_carry_their_numeric_code() {
+        let response = ddl_result_to_native(
+            1,
+            Err(DdlError {
+                sqlstate: "42P01".to_owned(),
+                message: "collection 'missing' does not exist".to_owned(),
+            }),
+        );
+
+        let error = response
+            .error
+            .expect("error responses must carry a payload");
+        assert_eq!(error.code, "42P01");
+        assert_eq!(
+            error.ndb_code,
+            nodedb_types::error::ErrorCode::COLLECTION_NOT_FOUND.0
+        );
+        assert_eq!(error.message, "collection 'missing' does not exist");
+    }
+
+    /// A site that renders a more specific SQLSTATE than the error implies
+    /// must still take the classification from the error rather than from the
+    /// SQLSTATE it just chose: `42601` is shared by several conditions, while
+    /// the error in hand names exactly one.
+    #[test]
+    fn a_site_chosen_sqlstate_keeps_the_errors_classification() {
+        let response = error_to_native_with_sqlstate(
+            1,
+            "42601",
+            &crate::Error::PlanError {
+                detail: "no such column".to_owned(),
+            },
+        );
+
+        let error = response
+            .error
+            .expect("error responses must carry a payload");
+        assert_eq!(error.code, "42601");
+        assert_eq!(
+            error.ndb_code,
+            nodedb_types::error::ErrorCode::PLAN_ERROR.0,
+            "the classification must come from the error, not from the SQLSTATE"
+        );
+    }
+
     /// The numeric code is populated for every variant, including the ones
     /// whose SQLSTATE falls through to `XX000` — otherwise the fix would be a
     /// per-variant special case rather than one classification.

nodedb/src/control/server/native/dispatch/direct_ops.rs

@@ -16,7 +16,8 @@ use nodedb_physical::physical_task::{PhysicalTask, PostSetOp};
 
 use super::response::data_plane_response_to_native;
 use super::single_task::dispatch_single_task;
-use super::{DispatchCtx, error_to_native};
+use super::{DispatchCtx, error_to_native, error_to_native_with_sqlstate};
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 /// Dispatch a direct Data Plane operation by opcode.
 pub(crate) async fn handle_direct_op(
@@ -55,7 +56,7 @@ pub(crate) async fn handle_direct_op(
 
     // Per-operation cap enforcement (vector dim, top_k, batch size, etc.).
     if let Err(e) = super::limits::check_op_limits(ctx.state, fields) {
-        return NativeResponse::error(seq, "0A000", e.to_string());
+        return error_to_native_with_sqlstate(seq, "0A000", &e);
     }
 
     // Quota enforcement — reject before planning or dispatch.
@@ -65,7 +66,7 @@ pub(crate) async fn handle_direct_op(
 
     let mut plan = match super::plan_builder::build_plan(ctx, op, fields, &collection) {
         Ok(p) => p,
-        Err(e) => return NativeResponse::error(seq, "42601", e.to_string()),
+        Err(e) => return error_to_native_with_sqlstate(seq, "42601", &e),
     };
 
     // Apply RLS before any special Control-Plane orchestration can observe the plan.
@@ -75,7 +76,7 @@ pub(crate) async fn handle_direct_op(
         &ctx.state.rls,
         ctx.auth_context(),
     ) {
-        return NativeResponse::error(seq, "42501", e.to_string());
+        return error_to_native_with_sqlstate(seq, "42501", &e);
     }
 
     // Refuse what column redaction cannot cover (an aggregate over a redacted
@@ -86,7 +87,7 @@ pub(crate) async fn handle_direct_op(
         ctx.auth_context(),
         &ctx.state.redaction,
     ) {
-        return NativeResponse::error(seq, "0A000", e.to_string());
+        return error_to_native_with_sqlstate(seq, "0A000", &e);
     }
 
     // Extracted before `plan` is moved/cloned into any of the branches below
@@ -107,7 +108,7 @@ pub(crate) async fn handle_direct_op(
     if let Some(info) = &plan_metering_info
         && let Err(e) = admit_quota_for_dispatch(ctx.state, &ctx.scope, info)
     {
-        return NativeResponse::error(seq, "53400", e.to_string());
+        return error_to_native_with_sqlstate(seq, "53400", &e);
     }
 
     // Whether the blanket metering call below (after the block) still needs
@@ -328,7 +329,7 @@ pub(crate) async fn handle_direct_op(
             let task = match authorized_tasks.into_tasks().into_iter().next() {
                 Some(task) => task,
                 None => {
-                    return NativeResponse::error(
+                    return sqlstate_error(
                         seq,
                         "XX000",
                         "authorization returned no task capability",

nodedb/src/control/server/native/dispatch/graph_match.rs

@@ -10,7 +10,7 @@ use crate::control::server::shared::quota_admission::admit_quota_for_dispatch;
 
 use super::raw_dispatch::dispatch_authorized_single_task;
 use super::response::data_plane_response_to_native;
-use super::{DispatchCtx, error_to_native};
+use super::{DispatchCtx, error_to_native, error_to_native_with_sqlstate};
 
 /// Dispatch a native `GraphMatch` op, unwrapping the DP `{rows, frontier}`
 /// envelope into a bare rows array before native conversion.
@@ -36,7 +36,7 @@ pub(crate) async fn handle_graph_match(
     let tenant_id = ctx.tenant_id();
 
     if let Err(error) = super::limits::check_op_limits(ctx.state, fields) {
-        return NativeResponse::error(seq, "0A000", error.to_string());
+        return error_to_native_with_sqlstate(seq, "0A000", &error);
     }
     if let Err(error) = ctx.state.check_tenant_quota(tenant_id) {
         return error_to_native(seq, &error);
@@ -45,15 +45,15 @@ pub(crate) async fn handle_graph_match(
     let mut plan =
         match super::plan_builder::build_plan(ctx, OpCode::GraphMatch, fields, &collection) {
             Ok(plan) => plan,
-            Err(error) => return NativeResponse::error(seq, "42601", error.to_string()),
+            Err(error) => return error_to_native_with_sqlstate(seq, "42601", &error),
         };
     if let Err(error) = crate::control::planner::rls_injection::inject_rls_for_single_plan(
         tenant_id.as_u64(),
         &mut plan,
         &ctx.state.rls,
         ctx.auth_context(),
     ) {
-        return NativeResponse::error(seq, "42501", error.to_string());
+        return error_to_native_with_sqlstate(seq, "42501", &error);
     }
     // Refuse what column redaction cannot cover: a MATCH returns graph
     // topology, which the result-path masking hook has no columns to rewrite.
@@ -63,7 +63,7 @@ pub(crate) async fn handle_graph_match(
         ctx.auth_context(),
         &ctx.state.redaction,
     ) {
-        return NativeResponse::error(seq, "0A000", error.to_string());
+        return error_to_native_with_sqlstate(seq, "0A000", &error);
     }
 
     // Stamp the active transaction id so MATCH reads resolve this connection's
@@ -84,7 +84,7 @@ pub(crate) async fn handle_graph_match(
     if let Some(info) = &plan_metering_info
         && let Err(e) = admit_quota_for_dispatch(ctx.state, &ctx.scope, info)
     {
-        return NativeResponse::error(seq, "53400", e.to_string());
+        return error_to_native_with_sqlstate(seq, "53400", &e);
     }
     let _request = ctx.state.tenant_request_guard(tenant_id);
     let raw = dispatch_authorized_single_task(ctx, tenant_id, vshard_id, plan, txn_id).await;

nodedb/src/control/server/native/dispatch/mod.rs

@@ -28,7 +28,7 @@ pub(crate) use admission_op::admission_operation;
 pub(crate) use auth::{NativeAuthOutcome, handle_auth, handle_ping};
 pub(crate) use conversion::{
     ddl_result_to_native, error_code_to_native, error_response_to_native, error_to_native,
-    shape_error_to_native, to_native_columns_rows,
+    error_to_native_with_sqlstate, shape_error_to_native, to_native_columns_rows,
 };
 pub(crate) use ctx::DispatchCtx;
 pub(crate) use direct_ops::handle_direct_op;

nodedb/src/control/server/native/dispatch/single_task.rs

@@ -16,7 +16,7 @@ use crate::types::{Lsn, RequestId};
 
 use super::raw_dispatch::dispatch_authorized_single_task;
 use super::response::data_plane_response_to_native;
-use super::{DispatchCtx, error_code_to_native, error_to_native};
+use super::{DispatchCtx, error_code_to_native, error_to_native, error_to_native_with_sqlstate};
 
 /// Dispatch one plan via the gateway (when wired) or the local SPSC path,
 /// converting the Data-Plane response into a `NativeResponse`.
@@ -76,7 +76,7 @@ pub(super) async fn dispatch_single_task(
     if let Some(info) = &plan_metering_info
         && let Err(e) = admit_quota_for_dispatch(ctx.state, &ctx.scope, info)
     {
-        return NativeResponse::error(seq, "53400", e.to_string());
+        return error_to_native_with_sqlstate(seq, "53400", &e);
     }
 
     let task = match route_in_tx_write(

nodedb/src/control/server/native/dispatch/sql.rs

@@ -27,6 +27,7 @@ use super::transaction_savepoint::{
     handle_release_savepoint, handle_rollback_to_savepoint, handle_savepoint,
 };
 use super::{DispatchCtx, error_to_native};
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 /// Handle a SQL statement: transaction control, SET/SHOW, DDL, or DataFusion.
 ///
@@ -110,7 +111,7 @@ async fn handle_sql_inner(
     }
 
     if ctx.sessions.transaction_state(ctx.peer_addr) == TransactionState::Failed {
-        return resp(NativeResponse::error(
+        return resp(sqlstate_error(
             seq,
             "25P02",
             "current transaction is aborted, commands ignored until end of transaction block",
@@ -361,7 +362,7 @@ async fn execute_planned(
         match try_open_sql_stream(ctx, seq, &tasks, database_id, Some(&output_schema)).await {
             Ok(Some(mut stream)) => {
                 let Some(scope) = lease_scope.take() else {
-                    return resp(NativeResponse::error(
+                    return resp(sqlstate_error(
                         seq,
                         "XX000",
                         "internal error: query lease scope missing before SQL stream dispatch",
@@ -382,7 +383,7 @@ async fn execute_planned(
     // after this local owner is dropped. Lazy streams above retain the raw
     // scope directly in their stream owner.
     let Some(lease_scope) = lease_scope.take() else {
-        return resp(NativeResponse::error(
+        return resp(sqlstate_error(
             seq,
             "XX000",
             "internal error: query lease scope missing before materialized SQL dispatch",

nodedb/src/control/server/native/dispatch/sql_admin.rs

@@ -9,6 +9,7 @@ use nodedb_types::protocol::NativeResponse;
 use nodedb_types::value::Value;
 
 use super::{DispatchCtx, error_to_native};
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 // ─── SET / SHOW / RESET (SQL form) ─────────────────────────────────
 
@@ -36,7 +37,7 @@ pub(super) fn handle_set_sql(ctx: &DispatchCtx<'_>, seq: u64, sql: &str) -> Nati
                 .to_string(),
         )
     } else {
-        return NativeResponse::error(seq, "42601", "invalid SET syntax");
+        return sqlstate_error(seq, "42601", "invalid SET syntax");
     };
 
     ctx.sessions.set_parameter(ctx.peer_addr, key, value);

nodedb/src/control/server/native/dispatch/sql_loop.rs

@@ -32,8 +32,9 @@ use super::sql_dispatch_task::dispatch_task;
 use super::streaming::SqlOutcome;
 use super::{
     DispatchCtx, error_code_to_native, error_response_to_native, error_to_native,
-    shape_error_to_native, to_native_columns_rows,
+    error_to_native_with_sqlstate, shape_error_to_native, to_native_columns_rows,
 };
+use crate::control::server::native::sqlstate_code::sqlstate_error;
 
 /// Wrap a materialized response as a non-streaming [`SqlOutcome`].
 #[inline]
@@ -69,11 +70,7 @@ pub(super) async fn run_dispatch_loop(
 
     for task in tasks {
         if task.tenant_id != ctx.tenant_id() {
-            return resp(NativeResponse::error(
-                seq,
-                "42501",
-                "tenant isolation violation",
-            ));
+            return resp(sqlstate_error(seq, "42501", "tenant isolation violation"));
         }
 
         // Cloned before `route_in_tx_write` consumes `task`, so a staged
@@ -101,7 +98,7 @@ pub(super) async fn run_dispatch_loop(
         if let Some(info) = &plan_metering_info
             && let Err(e) = admit_quota_for_dispatch(ctx.state, &ctx.scope, info)
         {
-            return resp(NativeResponse::error(seq, "53400", e.to_string()));
+            return resp(error_to_native_with_sqlstate(seq, "53400", &e));
         }
 
         // In transaction: route through the protocol-neutral staging gate.
@@ -154,7 +151,7 @@ pub(super) async fn run_dispatch_loop(
                 Arc::clone(&plan_lease_scope),
             )
         {
-            return resp(NativeResponse::error(
+            return resp(sqlstate_error(
                 seq,
                 "XX000",
                 "internal error: failed to retain descriptor leases for buffered transaction tasks",

nodedb/src/control/server/native/dispatch/streaming.rs

@@ -47,7 +47,7 @@ impl SqlOutcome {
     pub(crate) fn into_response(self) -> NativeResponse {
         match self {
             SqlOutcome::Response(r) => *r,
-            SqlOutcome::Stream(s) => NativeResponse::error(
+            SqlOutcome::Stream(s) => crate::control::server::native::sqlstate_code::sqlstate_error(
                 s.seq,
                 "XX000",
                 "internal error: SQL stream produced on a non-streaming path",

nodedb/src/control/server/native/dispatch/transaction.rs

@@ -81,7 +81,12 @@ pub(crate) fn handle_begin(ctx: &DispatchCtx<'_>, seq: u64) -> NativeResponse {
                 crate::Error::BadRequest { detail } => detail.clone(),
                 other => other.to_string(),
             };
-            NativeResponse::error(seq, "25P02", message)
+            NativeResponse::error_with_code(
+                seq,
+                "25P02",
+                message,
+                crate::error_classify::classify(&e).code().0,
+            )
         }
     }
 }