fix(gateway): keep the routed error's SQLSTATE class

nodedb-cluster-tests/tests/common_suite/cases/descriptor_lease_planner_integration.rs

@@ -115,16 +115,16 @@ async fn drain_forces_plan_retry_surfacing_typed_error() {
     })
     .await;
 
-    // A SELECT should fail fast: the pgwire retry helper makes
-    // 3 attempts, each hits the drain check, each surfaces
-    // `RetryableSchemaChanged`. After the retry budget the
-    // error maps to a plan error visible to the client.
+    // A SELECT must fail fast: the pgwire retry helper makes 3 attempts,
+    // each hits the drain check and surfaces `RetryableSchemaChanged`; after
+    // the retry budget the client reads that variant's Display text.
     let result = leader.exec("SELECT * FROM retry_me").await;
     assert!(result.is_err(), "SELECT must fail while drain is active");
     let err = result.unwrap_err();
     assert!(
-        err.to_lowercase().contains("retryable schema change")
-            || err.to_lowercase().contains("plan error"),
+        err.to_lowercase()
+            .contains("schema changed during execution")
+            || err.to_lowercase().contains("please retry"),
         "unexpected error variant: {err}"
     );
 

nodedb/src/control/gateway/error_map/http.rs

@@ -23,10 +23,7 @@ impl GatewayErrorMap {
                 format!("cluster in leader election; leader hint: {leader_addr}"),
             ),
             Error::DeadlineExceeded { .. } => (504, err.to_string()),
-            Error::RetryableSchemaChanged { descriptor } => (
-                503,
-                format!("schema changed during execution ({descriptor}); please retry"),
-            ),
+            Error::RetryableSchemaChanged { .. } => (503, err.to_string()),
             Error::CollectionNotFound { collection, .. } => {
                 (404, format!("collection \"{collection}\" does not exist"))
             }

nodedb/src/control/gateway/error_map/native.rs

@@ -26,10 +26,7 @@ impl GatewayErrorMap {
                 (CODE_NOT_LEADER, format!("not leader; hint: {leader_addr}"))
             }
             Error::DeadlineExceeded { .. } => (CODE_DEADLINE, err.to_string()),
-            Error::RetryableSchemaChanged { descriptor } => (
-                CODE_SCHEMA_CHANGED,
-                format!("schema changed ({descriptor})"),
-            ),
+            Error::RetryableSchemaChanged { .. } => (CODE_SCHEMA_CHANGED, err.to_string()),
             Error::CollectionNotFound { collection, .. } => (
                 CODE_NOT_FOUND,
                 format!("collection \"{collection}\" not found"),

nodedb/src/control/gateway/error_map/pgwire.rs

@@ -2,56 +2,18 @@
 
 //! pgwire error shape: `(SQLSTATE, message)`.
 
-use nodedb_types::error::sqlstate;
-
 use super::gateway_map::GatewayErrorMap;
 use crate::Error;
 
 impl GatewayErrorMap {
     /// Map a gateway error into `(sqlstate, message)` for pgwire.
     ///
-    /// Returns a `'static` SQLSTATE string and an owned message string.
-    /// The SQLSTATE codes match those in `pgwire::types::error_to_sqlstate`
-    /// so migrated call-sites are wire-compatible with the old forwarding path.
+    /// One table answers for the direct and the routed path, so a client sees
+    /// the same class and message for a given error either way.
     pub fn to_pgwire(err: &Error) -> (&'static str, String) {
-        match err {
-            Error::NotLeader { leader_addr, .. } => (
-                sqlstate::DATABASE_DROPPED,
-                format!("cluster in leader election; leader hint: {leader_addr}"),
-            ),
-            Error::DeadlineExceeded { .. } => (sqlstate::QUERY_CANCELED, err.to_string()),
-            Error::RetryableSchemaChanged { descriptor } => (
-                sqlstate::INTERNAL_ERROR,
-                format!("schema changed during execution ({descriptor}); please retry"),
-            ),
-            Error::CollectionNotFound { collection, .. } => (
-                sqlstate::UNDEFINED_TABLE,
-                format!("collection \"{collection}\" does not exist"),
-            ),
-            Error::RejectedAuthz { .. } => (sqlstate::INSUFFICIENT_PRIVILEGE, err.to_string()),
-            Error::BadRequest { detail } => (sqlstate::SYNTAX_ERROR, detail.clone()),
-            Error::PlanError { detail } => (sqlstate::SYNTAX_ERROR, detail.clone()),
-            Error::Serialization { .. } | Error::Codec { .. } => {
-                (sqlstate::INTERNAL_ERROR, err.to_string())
-            }
-            Error::Internal { .. } => (sqlstate::INTERNAL_ERROR, err.to_string()),
-            Error::NoLeader { .. } => (sqlstate::LOCK_NOT_AVAILABLE, err.to_string()),
-            Error::CrossCollectionNotColocated { .. } => {
-                (sqlstate::FEATURE_NOT_SUPPORTED, err.to_string())
-            }
-            Error::RemoteTyped { code, message } => (
-                crate::control::server::pgwire::types::error_map::numeric_code_to_sqlstate(*code),
-                message.clone(),
-            ),
-            // A shard verdict that rode back as a typed code keeps the exact
-            // SQLSTATE and message the direct dispatch path would have given.
-            Error::DataPlane(code) => {
-                let (_severity, state, message) =
-                    crate::control::server::shared::ddl::sqlstate::error_code_to_sqlstate(code);
-                (state, message)
-            }
-            _ => (sqlstate::INTERNAL_ERROR, err.to_string()),
-        }
+        let (_severity, state, message) =
+            crate::control::server::pgwire::types::error_to_sqlstate(err);
+        (state, message)
     }
 }
 
@@ -61,6 +23,7 @@ mod tests {
         authz, deadline, internal, not_found, not_leader, schema_changed, serialization,
     };
     use super::*;
+    use nodedb_types::error::sqlstate;
 
     #[test]
     fn pgwire_not_leader() {
@@ -74,6 +37,8 @@ mod tests {
         assert_eq!(code, sqlstate::QUERY_CANCELED);
     }
 
+    /// Both paths answer `INTERNAL_ERROR` from the shared table, with the
+    /// variant's own message naming the descriptor.
     #[test]
     fn pgwire_schema_changed() {
         let (code, msg) = GatewayErrorMap::to_pgwire(&schema_changed());
@@ -105,4 +70,107 @@ mod tests {
         let (code, _) = GatewayErrorMap::to_pgwire(&serialization());
         assert_eq!(code, sqlstate::INTERNAL_ERROR);
     }
+
+    /// For every classified variant, the routed class equals the direct class,
+    /// and the direct class is not `XX000`.
+    #[test]
+    fn gateway_and_direct_mapper_agree_on_classified_errors() {
+        use crate::types::{DatabaseId, TenantId};
+
+        let samples = vec![
+            Error::RejectedConstraint {
+                collection: "orders".into(),
+                constraint: "unique".into(),
+                detail: "duplicate key".into(),
+            },
+            Error::TxnOverlayMemoryExceeded { limit: 1 << 20 },
+            Error::UndefinedFunction {
+                name: "no_such_fn".into(),
+            },
+            Error::UndefinedObject {
+                kind: "sequence",
+                name: "s".into(),
+            },
+            Error::ObjectNotInPrerequisiteState {
+                object: "s".into(),
+                detail: "currval before nextval".into(),
+            },
+            Error::UndefinedColumn {
+                column: "nope".into(),
+            },
+            Error::AmbiguousColumn {
+                column: "id".into(),
+            },
+            Error::UnknownStrictField {
+                collection: "c".into(),
+                column: "x".into(),
+            },
+            Error::DivisionByZero,
+            Error::InvalidLimitValue {
+                clause: "LIMIT",
+                value: "-1".into(),
+            },
+            Error::DocumentNotFound {
+                collection: "c".into(),
+                document_id: "d".into(),
+            },
+            Error::ConflictRetry {
+                collection: "c".into(),
+                document_id: "d".into(),
+            },
+            Error::CalvinSerializationConflict,
+            Error::CalvinParticipantError,
+            Error::SourceFrozen {
+                database_id: DatabaseId::new(7),
+            },
+            Error::CloneWriteRequiresMaterialize {
+                collection: "c".into(),
+                engine: "kv".into(),
+                database: "db".into(),
+                reason: "shadowed",
+            },
+            Error::RateExceeded {
+                gate: "write".into(),
+                detail: "over budget".into(),
+                retry_after_ms: 50,
+            },
+            Error::MemoryExhausted {
+                engine: "kv".into(),
+            },
+            Error::FanOutExceeded {
+                shards_touched: 9,
+                limit: 8,
+            },
+            Error::MaterializedSumTargetNotFound {
+                target_collection: "t".into(),
+                join_column: "k".into(),
+                join_value: "1".into(),
+            },
+            Error::CollectionDeactivated {
+                tenant_id: TenantId::new(0),
+                collection: "c".into(),
+                retention_expires_at_ns: 1,
+            },
+            Error::FeatureNotSupported {
+                detail: "sparse".into(),
+            },
+            Error::BackupTenantMismatch {
+                expected: 1,
+                actual: 2,
+            },
+            Error::BackupKeyMismatch,
+        ];
+
+        for err in samples {
+            let (_sev, direct, _msg) =
+                crate::control::server::pgwire::types::error_to_sqlstate(&err);
+            let (gateway, _) = GatewayErrorMap::to_pgwire(&err);
+            assert_ne!(
+                direct,
+                sqlstate::INTERNAL_ERROR,
+                "sample has no class of its own: {err:?}"
+            );
+            assert_eq!(gateway, direct, "routed class differs for {err:?}");
+        }
+    }
 }

nodedb/src/control/gateway/error_map/resp.rs

@@ -26,9 +26,7 @@ impl GatewayErrorMap {
             }
             Error::RejectedConstraint { detail, .. } => format!("CONSTRAINT {detail}"),
             Error::TypeMismatch { detail, .. } => format!("WRONGTYPE {detail}"),
-            Error::RetryableSchemaChanged { descriptor } => {
-                format!("ERR schema changed ({descriptor}); please retry")
-            }
+            Error::RetryableSchemaChanged { .. } => format!("ERR {err}"),
             Error::RemoteTyped { code, message } => {
                 format!("{} {message}", remote_code_to_resp_prefix(*code))
             }

nodedb/src/error/types.rs

@@ -346,8 +346,8 @@ pub enum Error {
     #[error("invalid {clause} value: {value}")]
     InvalidLimitValue { clause: &'static str, value: String },
 
-    /// Descriptor lease conflict; pgwire retries within `PLAN_RETRY_BUDGET`.
-    #[error("retryable schema change on {descriptor}")]
+    /// Retried within `PLAN_RETRY_BUDGET`; every path renders this wording from `Display`.
+    #[error("schema changed during execution ({descriptor}); please retry")]
     RetryableSchemaChanged { descriptor: String },
 
     /// Leader-change overwrite; callers must re-propose to avoid false success.

nodedb/src/error_classify.rs

@@ -177,9 +177,7 @@ pub(crate) fn classify(e: &Error) -> NodeDbError {
         Error::InvalidLimitValue { clause, value } => {
             NodeDbError::invalid_limit_value(*clause, value.clone())
         }
-        Error::RetryableSchemaChanged { descriptor } => {
-            NodeDbError::plan_error(format!("retryable schema change on {descriptor}"))
-        }
+        Error::RetryableSchemaChanged { .. } => NodeDbError::plan_error(e.to_string()),
         Error::RetryableLeaderChange {
             group_id,
             log_index,

nodedb/tests/crash_harness/pgwire.rs

@@ -11,7 +11,8 @@ use std::time::{Duration, Instant};
 use super::CrashHarness;
 
 /// Bounded retry budget for the `Error::RetryableSchemaChanged` condition
-/// (rendered over pgwire as `XX000: retryable schema change on <descriptor>`).
+/// (rendered over pgwire as `XX000: schema changed during execution
+/// (<descriptor>); please retry`).
 ///
 /// The server already retries this condition server-side for ~750ms
 /// (`retry_on_schema_change`, `control/server/shared/retry.rs`, 5 attempts,
@@ -30,15 +31,10 @@ const SCHEMA_CHANGE_RETRY_ATTEMPTS: usize = 5;
 const SCHEMA_CHANGE_RETRY_BACKOFF: Duration = Duration::from_millis(150);
 
 /// Substring of `Error::RetryableSchemaChanged`'s Display text
-/// (`#[error("retryable schema change on {descriptor}")]` in
-/// `nodedb/src/error.rs`). No distinct SQLSTATE is assigned to this
-/// condition — `error_to_sqlstate` in `control/server/pgwire/types/error_map.rs`
-/// has no arm for it, so it falls through to the generic
-/// `sqlstate::INTERNAL_ERROR` (`XX000`) bucket shared by every other
-/// unmapped error. Matching on that code alone would blanket-retry
-/// unrelated internal errors, so the message text — which is the error
-/// type's own stable Display string, not free-form prose — is the only
-/// durable signal available.
+/// (`#[error("schema changed during execution ({descriptor}); please retry")]`
+/// in `nodedb/src/error/types.rs`). The message is the durable signal: a code
+/// alone would blanket-retry unrelated internal errors, because the class this
+/// condition carries depends on the mapper in front of it.
 /// The server was still reporting `RetryableSchemaChanged` when the
 /// client-side retry budget ran out.
 ///
@@ -51,7 +47,7 @@ pub struct RetryableSchemaChange;
 
 fn is_retryable_schema_change(e: &tokio_postgres::Error) -> bool {
     e.as_db_error()
-        .is_some_and(|db| db.message().contains("retryable schema change"))
+        .is_some_and(|db| db.message().contains("schema changed during execution"))
 }
 
 /// Substring of the `calvin-submit` internal error text