fix(pgwire): answer a real class for the arm-less error variants

nodedb/src/control/server/pgwire/types/error_map.rs

@@ -201,7 +201,14 @@ pub fn error_to_sqlstate(err: &crate::Error) -> (&'static str, &'static str, Str
         crate::Error::DataPlane(code) => {
             crate::control::server::shared::ddl::sqlstate::error_code_to_sqlstate(code)
         }
-        _ => ("ERROR", sqlstate::INTERNAL_ERROR, err.to_string()),
+        // A variant with no arm of its own carries its class on the error:
+        // `classify` borrows it (the crate's one Error-to-NodeDbError map),
+        // then this table answers instead of a blanket `XX000`.
+        _ => (
+            "ERROR",
+            numeric_code_to_sqlstate(crate::error_classify::classify(err).code()),
+            err.to_string(),
+        ),
     }
 }
 
@@ -466,4 +473,105 @@ mod tests {
 
         assert!(message.contains("timestamp"));
     }
+
+    /// A variant with no arm of its own still carries a class: the fallback
+    /// borrows it through `classify` and this table answers it.
+    #[test]
+    fn arm_less_variants_keep_their_class_through_the_fallback() {
+        use crate::Error;
+
+        let cases = vec![
+            (
+                Error::AppendOnlyViolation {
+                    collection: "c".into(),
+                    detail: "d".into(),
+                },
+                sqlstate::APPEND_ONLY_VIOLATION,
+            ),
+            (
+                Error::BalanceViolation {
+                    collection: "c".into(),
+                    detail: "d".into(),
+                },
+                sqlstate::BALANCE_VIOLATION,
+            ),
+            (
+                Error::InsufficientBalance {
+                    collection: "c".into(),
+                    key: "k".into(),
+                    detail: "d".into(),
+                },
+                sqlstate::CHECK_VIOLATION,
+            ),
+            (
+                Error::PeriodLocked {
+                    collection: "c".into(),
+                    detail: "d".into(),
+                },
+                sqlstate::PERIOD_LOCKED,
+            ),
+            (
+                Error::RetentionViolation {
+                    collection: "c".into(),
+                    detail: "d".into(),
+                },
+                sqlstate::RETENTION_VIOLATION,
+            ),
+            (
+                Error::LegalHoldActive {
+                    collection: "c".into(),
+                    detail: "d".into(),
+                },
+                sqlstate::LEGAL_HOLD_ACTIVE,
+            ),
+            (
+                Error::StateTransitionViolation {
+                    collection: "c".into(),
+                    detail: "d".into(),
+                },
+                sqlstate::STATE_TRANSITION_VIOLATION,
+            ),
+            (
+                Error::TransitionCheckViolation {
+                    collection: "c".into(),
+                    detail: "d".into(),
+                },
+                sqlstate::TRANSITION_CHECK_VIOLATION,
+            ),
+            (
+                Error::TypeGuardViolation {
+                    collection: "c".into(),
+                    detail: "d".into(),
+                },
+                sqlstate::TYPE_GUARD_VIOLATION,
+            ),
+            (
+                Error::TypeMismatch {
+                    collection: "c".into(),
+                    key: "k".into(),
+                    detail: "d".into(),
+                },
+                sqlstate::DATATYPE_MISMATCH,
+            ),
+            (
+                Error::MirrorReadOnly {
+                    database: "db".into(),
+                },
+                sqlstate::READ_ONLY_SQL_TRANSACTION,
+            ),
+            (
+                // Retryable in the lease paths, so the fallback must not read
+                // as a plan/syntax error now that it answers through `classify`.
+                Error::RetryableSchemaChanged {
+                    descriptor: "users".into(),
+                },
+                sqlstate::SERIALIZATION_FAILURE,
+            ),
+        ];
+
+        for (err, expected) in cases {
+            let (_severity, state, _message) = error_to_sqlstate(&err);
+            assert_eq!(state, expected, "{err:?}");
+        }
+    }
 }

nodedb/src/error_classify.rs

@@ -177,8 +177,11 @@ pub(crate) fn classify(e: &Error) -> NodeDbError {
         Error::InvalidLimitValue { clause, value } => {
             NodeDbError::invalid_limit_value(*clause, value.clone())
         }
+        // Retryable, like a write conflict: the descriptor-lease paths already
+        // retry this variant, so it must not reach a client as a plan/syntax
+        // error, which reads as "fix your SQL".
         Error::RetryableSchemaChanged { descriptor } => {
-            NodeDbError::plan_error(format!("retryable schema change on {descriptor}"))
+            NodeDbError::write_conflict("catalog", descriptor.clone())
         }
         Error::RetryableLeaderChange {
             group_id,