fix(pgwire): answer kv writes with one INSERT tag

nodedb/src/control/server/pgwire/handler/routing/cluster_array.rs

@@ -19,7 +19,7 @@ use crate::control::server::response_shape::redaction::QueryRedaction;
 use crate::control::server::response_shape::schema::OutputSchema;
 use crate::control::server::shared::session::SessionId;
 
-use super::super::super::types::{error_to_sqlstate, sqlstate_error};
+use super::super::super::types::{error_to_sqlstate, numeric_code_to_sqlstate, sqlstate_error};
 use super::super::core::NodeDbPgHandler;
 use super::super::plan::{PlanKind, payload_to_response};
 use super::super::shape_encode;
@@ -128,7 +128,7 @@ impl NodeDbPgHandler {
             projection,
             Some(redaction.ctx(&self.state.redaction)),
         )
-        .map_err(|e| sqlstate_error("XX000", e.message()))?
+        .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))?
         {
             ShapeOutcome::Rows(shaped) => {
                 let (response, notice) =

nodedb/src/control/server/pgwire/handler/routing/dispatch_loop.rs

@@ -23,7 +23,9 @@ use crate::control::server::shared::session::SessionId;
 use crate::types::TenantId;
 use nodedb_physical::physical_task::{PhysicalTask, PostSetOp};
 
-use super::super::super::types::{error_to_sqlstate, response_status_to_sqlstate, sqlstate_error};
+use super::super::super::types::{
+    error_to_sqlstate, numeric_code_to_sqlstate, response_status_to_sqlstate, sqlstate_error,
+};
 use super::super::core::NodeDbPgHandler;
 use super::super::plan::{PlanKind, describe_plan, payload_to_response};
 use super::super::shape_encode;
@@ -402,7 +404,10 @@ impl NodeDbPgHandler {
                     tenant_id,
                     redaction: Some(redaction.ctx(&self.state.redaction)),
                 })
-                .map_err(|e| sqlstate_error("XX000", e.message()))?
+                // Preserve the shaper error's own SQLSTATE: hardcoding XX000
+                // turned a client-visible refusal such as an undefined
+                // sequence into an internal fault.
+                .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))?
                 {
                     ShapeOutcome::Rows(shaped) => {
                         task_rows = Some(shaped.rows.len() as u64);

nodedb/src/control/server/pgwire/handler/routing/execute_dml_hooks.rs

@@ -21,7 +21,7 @@ use crate::control::trigger::dml_hook::DmlWriteInfo;
 use crate::types::TenantId;
 use nodedb_physical::physical_task::PhysicalTask;
 
-use super::super::super::types::{error_to_sqlstate, sqlstate_error};
+use super::super::super::types::{error_to_sqlstate, numeric_code_to_sqlstate, sqlstate_error};
 use super::super::core::NodeDbPgHandler;
 use super::super::plan::PlanKind;
 
@@ -332,7 +332,7 @@ impl NodeDbPgHandler {
                         projection,
                         Some(redaction.ctx(&self.state.redaction)),
                     )
-                    .map_err(|e| sqlstate_error("XX000", e.message()))?
+                    .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))?
                     {
                         ShapeOutcome::Rows(shaped) => {
                             // Clone write-path DML result (PointUpdate/PointDelete):

nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs

@@ -22,7 +22,7 @@ use crate::control::server::shared::metering::{PlanMeteringInfo, meter_dispatch}
 use crate::types::{TenantId, TraceId};
 use nodedb_physical::physical_task::PhysicalTask;
 
-use super::super::super::types::sqlstate_error;
+use super::super::super::types::{numeric_code_to_sqlstate, sqlstate_error};
 use super::super::core::NodeDbPgHandler;
 use super::super::plan::{PlanKind, multirow_payload_to_response};
 use super::super::shape_encode;
@@ -75,7 +75,7 @@ fn push_shaped_response(
         projection,
         Some(redaction.ctx(&state.redaction)),
     )
-    .map_err(|e| sqlstate_error("XX000", e.message()))?
+    .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))?
     {
         ShapeOutcome::Rows(shaped) => {
             let (response, notice) = shape_encode::shaped_query_response(shaped, result_formats);
@@ -228,7 +228,7 @@ impl NodeDbPgHandler {
                         projection,
                         Some(redaction.ctx(&self.state.redaction)),
                     )
-                    .map_err(|e| sqlstate_error("XX000", e.message()))?
+                    .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))?
                     {
                         ShapeOutcome::Rows(shaped) => {
                             task_rows = Some(task_rows.unwrap_or(0) + shaped.rows.len() as u64);

nodedb/src/control/server/pgwire/handler/routing/set_ops.rs

@@ -13,7 +13,7 @@ use crate::control::server::response_shape::compose::{self, ShapeOutcome};
 use crate::control::server::response_shape::redaction::RedactionCtx;
 use crate::control::server::response_shape::schema::OutputSchema;
 
-use super::super::super::types::sqlstate_error;
+use super::super::super::types::{numeric_code_to_sqlstate, sqlstate_error};
 use super::super::plan::{PlanKind, multirow_payload_to_response};
 use super::super::shape_encode;
 
@@ -37,7 +37,7 @@ pub(super) fn apply_set_ops(
     };
     Ok(
         match compose::shape_payload_no_plan(&merged, PlanKind::MultiRow, projection, redaction)
-            .map_err(|e| sqlstate_error("XX000", e.message()))?
+            .map_err(|e| sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message()))?
         {
             ShapeOutcome::Rows(shaped) => {
                 shape_encode::shaped_query_response(shaped, result_formats)

nodedb/src/control/server/pgwire/handler/routing/streaming.rs

@@ -22,7 +22,7 @@ use crate::control::server::response_shape::redaction::QueryRedaction;
 use crate::control::server::shared::session::SessionId;
 
 use super::super::super::types::error_to_sqlstate;
-use super::super::super::types::sqlstate_error;
+use super::super::super::types::{numeric_code_to_sqlstate, sqlstate_error};
 use super::super::core::NodeDbPgHandler;
 use super::super::plan::{PlanKind, multirow_payload_to_response};
 use super::super::stream_response;
@@ -141,8 +141,9 @@ impl NodeDbPgHandler {
                             shaping.projection,
                             Some(redaction.ctx(&state.redaction)),
                         )
-                        .map_err(|e| sqlstate_error("XX000", e.message()))?
-                        {
+                        .map_err(|e| {
+                            sqlstate_error(numeric_code_to_sqlstate(e.code()), e.message())
+                        })? {
                             ShapeOutcome::Rows(shaped) => {
                                 let (response, _notice) = crate::control::server::pgwire::handler::shape_encode::shaped_query_response(
                                 shaped,

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

@@ -197,7 +197,7 @@ pub fn error_to_sqlstate(err: &crate::Error) -> (&'static str, &'static str, Str
 /// numeric code, so a constraint violation (say) maps to the same SQLSTATE
 /// whether it happened locally or on a remote node. Unmapped/unknown codes
 /// fall back to INTERNAL_ERROR — the behaviour before codes were preserved.
-pub(crate) fn numeric_code_to_sqlstate(code: nodedb_types::error::ErrorCode) -> &'static str {
+pub fn numeric_code_to_sqlstate(code: nodedb_types::error::ErrorCode) -> &'static str {
     use nodedb_types::error::ErrorCode as Ec;
     match code {
         // Mirrors the `RejectedConstraint` arm.
@@ -213,6 +213,27 @@ pub(crate) fn numeric_code_to_sqlstate(code: nodedb_types::error::ErrorCode) ->
         Ec::DOCUMENT_NOT_FOUND => sqlstate::NO_DATA,
         // Mirrors the `BadRequest` / `PlanError` arms.
         Ec::BAD_REQUEST | Ec::PLAN_ERROR => sqlstate::SYNTAX_ERROR,
+        // A named catalog object (a sequence, most commonly) that does not
+        // exist. Without this arm the code falls back to INTERNAL_ERROR
+        // (XX000), so a client-visible "undefined object" reaches pgwire as an
+        // internal fault — the same error-class leak the response-shaping path
+        // had.
+        Ec::UNDEFINED_OBJECT => sqlstate::UNDEFINED_OBJECT,
+        // The object exists but its prerequisite step has not run, such as
+        // `currval` before this session called `nextval`. Mirrors the
+        // `ObjectNotInPrerequisiteState` arm.
+        Ec::OBJECT_NOT_READY => sqlstate::OBJECT_NOT_IN_PREREQUISITE_STATE,
+        // A generic lookup miss. Mirrors `crate::Error::DocumentNotFound`'s
+        // no-data class for the callers that use the generic code.
+        Ec::NOT_FOUND => sqlstate::NO_DATA,
+        // The named database does not exist.
+        Ec::DATABASE_NOT_FOUND => sqlstate::INVALID_CATALOG_NAME,
+        // SQL is disabled for this server; not a retryable internal fault.
+        Ec::SQL_NOT_ENABLED => sqlstate::FEATURE_NOT_SUPPORTED,
+        // A value could not be coerced to the target type.
+        Ec::TYPE_MISMATCH => sqlstate::CANNOT_COERCE,
+        // A value outside the target type's range.
+        Ec::OVERFLOW => sqlstate::NUMERIC_VALUE_OUT_OF_RANGE,
         // Mirrors the `UndefinedFunction` arm.
         Ec::UNDEFINED_FUNCTION => sqlstate::UNDEFINED_FUNCTION,
         // Mirrors the `UndefinedColumn` arm.
@@ -266,3 +287,65 @@ pub fn response_status_to_sqlstate(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nodedb_types::error::{ErrorCode as Ec, NodeDbError};
+
+    /// Every routed shaper call site (`cluster_array`, `dispatch_loop`,
+    /// `execute_dml_hooks`, `gateway_dispatch`, `set_ops`, `streaming`) renders a
+    /// shaper error as `sqlstate_error(numeric_code_to_sqlstate(e.code()), ..)`.
+    /// The rendered code is what a client matches on, so a code that already
+    /// classifies the fault must reach it as that class.
+    fn routed_code(err: &NodeDbError) -> String {
+        let rendered = sqlstate_error(numeric_code_to_sqlstate(err.code()), err.message());
+        match rendered {
+            PgWireError::UserError(info) => info.code,
+            other => panic!("sqlstate_error must build a UserError, got {other:?}"),
+        }
+    }
+
+    /// An undefined named object — the sequence case the routed sites exist to
+    /// answer — keeps its own class (42704) instead of the XX000 fallback. A
+    /// client treats XX000 as an internal fault and does not retry; it treats
+    /// 42704 as a missing object it caused.
+    #[test]
+    fn routed_shaper_error_keeps_its_numeric_class() {
+        let err = NodeDbError::undefined_object("seq_routed_probe");
+        assert_eq!(routed_code(&err), sqlstate::UNDEFINED_OBJECT);
+        assert_eq!(routed_code(&err), "42704");
+        assert_ne!(routed_code(&err), sqlstate::INTERNAL_ERROR);
+    }
+
+    /// The read-path codes the routed sites can surface each map to their
+    /// Postgres class rather than collapsing into the XX000 fallback. An
+    /// unmapped code is exactly the leak this pins shut.
+    #[test]
+    fn read_path_numeric_codes_keep_their_class() {
+        let cases = [
+            (Ec::UNDEFINED_OBJECT, sqlstate::UNDEFINED_OBJECT),
+            (
+                Ec::OBJECT_NOT_READY,
+                sqlstate::OBJECT_NOT_IN_PREREQUISITE_STATE,
+            ),
+            (Ec::NOT_FOUND, sqlstate::NO_DATA),
+            (Ec::DATABASE_NOT_FOUND, sqlstate::INVALID_CATALOG_NAME),
+            (Ec::SQL_NOT_ENABLED, sqlstate::FEATURE_NOT_SUPPORTED),
+            (Ec::TYPE_MISMATCH, sqlstate::CANNOT_COERCE),
+            (Ec::OVERFLOW, sqlstate::NUMERIC_VALUE_OUT_OF_RANGE),
+        ];
+        for (numeric, expected) in cases {
+            let mapped = numeric_code_to_sqlstate(numeric);
+            assert_eq!(
+                mapped, expected,
+                "numeric code {numeric} must keep class {expected}"
+            );
+            assert_ne!(
+                mapped,
+                sqlstate::INTERNAL_ERROR,
+                "numeric code {numeric} must not fall back to XX000"
+            );
+        }
+    }
+}

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

@@ -11,7 +11,8 @@ pub mod parse;
 pub mod privilege;
 
 pub use error_map::{
-    error_to_sqlstate, notice_warning, response_status_to_sqlstate, sqlstate_error,
+    error_to_sqlstate, notice_warning, numeric_code_to_sqlstate, response_status_to_sqlstate,
+    sqlstate_error,
 };
 pub use field::{
     bool_field, bytea_field, float4_array_field, float4_field, float8_array_field, float8_field,

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

@@ -147,7 +147,10 @@ pub fn describe_plan(plan: &PhysicalPlan) -> PlanKind {
         PhysicalPlan::Document(DocumentOp::PointPut { .. })
         | PhysicalPlan::Document(DocumentOp::PointInsert { .. })
         | PhysicalPlan::Document(DocumentOp::BatchInsert { .. })
+        | PhysicalPlan::Kv(KvOp::Insert { .. })
         | PhysicalPlan::Kv(KvOp::InsertIfAbsent { .. })
+        | PhysicalPlan::Kv(KvOp::Put { .. })
+        | PhysicalPlan::Kv(KvOp::BatchPut { .. })
         | PhysicalPlan::Columnar(ColumnarOp::Insert { .. }) => DmlResult("INSERT"),
 
         PhysicalPlan::Document(DocumentOp::PointUpdate {

nodedb/src/data/executor/handlers/kv/crud/write_basic.rs

@@ -80,7 +80,10 @@ impl CoreLoop {
             // stored post-image, not an echo of the request.
             return self.kv_stored_returning_response(task, spec, rls_filters, &[(key, value)]);
         }
-        self.response_ok(task)
+        // A put always writes its row, so the statement affected exactly one:
+        // the tag is `INSERT 0 1`, never a bare `OK` (pgwire's generic tag for
+        // a plan that reports nothing).
+        self.response_affected(task, 1)
     }
 
     /// SQL `INSERT` semantics: write only if key doesn't already exist.
@@ -167,7 +170,10 @@ impl CoreLoop {
         if let Some(spec) = returning {
             return self.kv_stored_returning_response(task, spec, rls_filters, &[(key, value)]);
         }
-        self.response_ok(task)
+        // An insert writes exactly one row or fails the statement, so the
+        // affected count is 1 and pgwire renders `INSERT 0 1` — the same tag
+        // the document engine's point insert produces.
+        self.response_affected(task, 1)
     }
 
     /// SQL `INSERT ... ON CONFLICT DO NOTHING` semantics: write if absent,

nodedb/tests/wire/cases/sql_dml_affected_counts.rs

@@ -286,6 +286,52 @@ async fn kv_multi_key_delete_reports_matched_key_count() {
     );
 }
 
+/// A key-value `INSERT` of a new key writes exactly one row, so its command tag
+/// must report `1`. A kv write that answers with no affected count renders as a
+/// bare `OK` tag, which the client's tag parser reads as `0` rows — the count is
+/// the only signal distinguishing "wrote" from "matched nothing".
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn kv_insert_of_new_key_reports_one() {
+    let server = TestServer::start().await;
+    server
+        .exec("CREATE COLLECTION kv_probe (key TEXT PRIMARY KEY, n INT) WITH (engine='kv')")
+        .await
+        .unwrap();
+
+    let count = affected(&server, "INSERT INTO kv_probe (key, n) VALUES ('a', 1)").await;
+    assert_eq!(
+        count, 1,
+        "a KV INSERT that wrote one key must report 1, not a bare tag the client reads as 0"
+    );
+    assert_eq!(
+        live_rows(&server, "SELECT count(*) FROM kv_probe WHERE key = 'a'").await,
+        1,
+        "the key must really be present, so the reported count is the honest one"
+    );
+}
+
+/// A key-value `UPSERT` overwrites or writes exactly one key, so its tag must
+/// report `1` for the same reason: the PUT path always writes the key.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn kv_upsert_of_new_key_reports_one() {
+    let server = TestServer::start().await;
+    server
+        .exec("CREATE COLLECTION kv_probe (key TEXT PRIMARY KEY, n INT) WITH (engine='kv')")
+        .await
+        .unwrap();
+
+    let count = affected(&server, "UPSERT INTO kv_probe (key, n) VALUES ('a', 1)").await;
+    assert_eq!(
+        count, 1,
+        "a KV UPSERT that wrote one key must report 1, not a bare tag the client reads as 0"
+    );
+    assert_eq!(
+        live_rows(&server, "SELECT count(*) FROM kv_probe WHERE key = 'a'").await,
+        1,
+        "the upserted key must really be present"
+    );
+}
+
 /// A CRDT document collection routes its PK-targeted delete through the CRDT
 /// engine, which shares the count contract: a delete that removed the row
 /// reports `1`.