ddl,http: classify the CRDT MERGE refusal and code the HTTP stream errors

Two surfaces still flattened a classified failure while the routed pgwire
paths already carried its class.

- CRDT MERGE: the admission/apply failures wrap a classified Error —
  ExternalCrdtPostImagePolicy::deny returns RejectedAuthz — so they map
  through error_to_sqlstate, which renders 42501 for a policy denial.
  The "authorization returned no capability" site stays XX000: issue
  #344 records it as an internal invariant by decision.
- HTTP stream: the in-band error lines now carry the numeric NodeDB code
  (shape and malformed-batch failures) and the status the gateway map
  already computed.

Consumer trace for the class change: the two wire assertions that pinned
the placeholder are updated with it —
`nodedb/tests/wire/cases/crdt_write_rls_database_scope.rs:141,202`
expected `XX000` and now expect `42501`, and the comment above the first
one no longer describes the old wrapper. No other test, doc, or path keys
on the CRDT merge SQLSTATE (`grep XX000` over `tests/wire/cases` finds no
other crdt/merge site).

The pgwire stream and DDL-dispatch sites are NOT touched here: the local
branch `fix/shaper-error-class` carried that work and its PR #341 was
closed as stacked on #359, with the instruction to resubmit against main
once the mapper PR lands.

Also refreshes the `response_shape/schema.rs` module comment, which still
claimed nothing consumes the module — the session caches the schema with
the physical tasks and shaping receives it as `projection`.

Refs #344. NOT RUN: cargo check -p nodedb --tests (the wire target).

Author: EnRaiha

nodedb/src/control/server/http/routes/query_stream.rs

@@ -169,8 +169,11 @@ pub(super) fn ndjson_body_stream(
                 None => break,
                 Some(Ok(b)) => b,
                 Some(Err(e)) => {
-                    let (_status, msg) = GatewayErrorMap::to_http(&e);
-                    let line = format!("{}\n", serde_json::json!({ "error": msg }));
+                    let (status, msg) = GatewayErrorMap::to_http(&e);
+                    let line = format!(
+                        "{}\n",
+                        serde_json::json!({ "error": msg, "status": status })
+                    );
                     yield Ok(Bytes::from(line));
                     return;
                 }
@@ -182,9 +185,13 @@ pub(super) fn ndjson_body_stream(
                     // A malformed batch payload is surfaced as an in-band error
                     // line (matching the mid-stream dispatch-error path above)
                     // rather than silently dropping the batch.
+                    let classified = crate::error_classify::classify(&e);
                     let line = format!(
                         "{}\n",
-                        serde_json::json!({ "error": format!("malformed response batch: {e}") })
+                        serde_json::json!({
+                            "error": format!("malformed response batch: {}", classified.message()),
+                            "code": classified.code().0,
+                        })
                     );
                     yield Ok(Bytes::from(line));
                     return;
@@ -206,7 +213,14 @@ pub(super) fn ndjson_body_stream(
                 Err(e) => {
                     // In-band error line, matching the malformed-batch path
                     // above: the HTTP body itself never errors.
-                    let line = format!("{}\n", serde_json::json!({ "error": format!("{e}") }));
+                    let classified = crate::error_classify::classify(&e);
+                    let line = format!(
+                        "{}\n",
+                        serde_json::json!({
+                            "error": classified.message(),
+                            "code": classified.code().0,
+                        })
+                    );
                     yield Ok(Bytes::from(line));
                     return;
                 }

nodedb/src/control/server/response_shape/schema.rs

@@ -3,10 +3,10 @@
 //! Planner-authoritative output schema types, plus the type mapping from the
 //! planner's `SqlDataType` to the response shaper's wire-facing `DdlColType`.
 //!
-//! Nothing in this module is consumed by existing call sites yet; it is a
-//! purely additive foundation for later threading the planner's resolved
-//! output schema into response shaping (replacing the SQL-string re-parse
-//! path).
+//! The planner derives the schema from the compiled plan and the catalog, and
+//! the session caches it with the physical tasks (`session/plan_cache.rs`).
+//! Shaping receives it as `MaterializedShapeRequest::projection`, which drives
+//! projection and the Control-Plane computed columns.
 
 /// One output column of a resolved query, as known by the planner.
 ///

nodedb/src/control/server/shared/ddl/neutral/dsl/crdt_merge.rs

@@ -8,6 +8,7 @@ use crate::bridge::envelope::PhysicalPlan;
 use crate::control::crdt_post_image_policy::ExternalCrdtPostImagePolicy;
 use crate::control::security::audit::ArcAuditEmitter;
 use crate::control::security::identity::{AuthenticatedIdentity, Permission};
+use crate::control::server::pgwire::types::error_map::error_to_sqlstate;
 use crate::control::server::shared::authorization::{authorize_collection, authorize_task_set};
 use crate::control::state::SharedState;
 use crate::types::DatabaseId;
@@ -87,7 +88,10 @@ pub async fn crdt_merge(
         },
     )
     .await
-    .map_err(|e| ddl_err("XX000", e.to_string()))?;
+    .map_err(|e| {
+        let (_, sqlstate, message) = error_to_sqlstate(&e);
+        ddl_err(sqlstate, message)
+    })?;
     if source_bytes.is_empty() {
         return Err(ddl_err(
             "02000",
@@ -98,7 +102,10 @@ pub async fn crdt_merge(
     let target_surrogate = state
         .surrogate_assigner
         .assign(database_id, tenant_id, collection, target_id.as_bytes())
-        .map_err(|e| ddl_err("XX000", e.to_string()))?;
+        .map_err(|e| {
+            let (_, sqlstate, message) = error_to_sqlstate(&e);
+            ddl_err(sqlstate, message)
+        })?;
 
     let apply_plan = PhysicalPlan::Crdt(CrdtOp::Apply {
         collection: nodedb_types::QualifiedCollection::new(database_id, collection),
@@ -163,7 +170,10 @@ pub async fn crdt_merge(
         },
     )
     .await
-    .map_err(|e| ddl_err("XX000", e.to_string()))?;
+    .map_err(|e| {
+        let (_, sqlstate, message) = error_to_sqlstate(&e);
+        ddl_err(sqlstate, message)
+    })?;
 
     state.audit_record(
         crate::control::security::audit::AuditEvent::AdminAction,

nodedb/tests/wire/cases/crdt_write_rls_database_scope.rs

@@ -132,13 +132,13 @@ async fn crdt_merge_in_non_default_database_is_rls_enforced() {
         "a write policy on a non-default-database CRDT collection must reject \
          a CRDT MERGE its predicate forbids",
     );
-    // `CRDT MERGE`'s handler wraps every admission failure — RLS denial
-    // included — under this one SQLSTATE (`crdt_merge.rs`'s `dispatch_...
-    // .map_err(|e| ddl_err("XX000", ...))`); the substantive assertion is
-    // that an error surfaces here at all, since pre-fix the merge applied
-    // silently and no error, of any code, reached the client.
+    // `CRDT MERGE`'s handler classifies every admission failure — an RLS
+    // denial reaches the client as `42501`. Pre-fix the merge applied
+    // silently and no error, of any code, reached the client, so the
+    // substantive assertion stays "an error surfaces at all"; the code is
+    // pinned because a denial is the one case a client can act on.
     assert_eq!(
-        sqlstate, "XX000",
+        sqlstate, "42501",
         "expected the CRDT admission failure's SQLSTATE, got: {sqlstate}"
     );
     assert_eq!(
@@ -199,7 +199,7 @@ async fn crdt_merge_in_default_database_is_still_rls_enforced() {
          MERGE its predicate forbids",
     );
     assert_eq!(
-        sqlstate, "XX000",
+        sqlstate, "42501",
         "expected the CRDT admission failure's SQLSTATE, got: {sqlstate}"
     );
     assert_eq!(