sql(postgres): reject only the query whose row the client cannot decode

docs/runtime/sql.mdx

@@ -1178,6 +1178,8 @@ try {
 | `ERR_POSTGRES_MULTIDIMENSIONAL_ARRAY_NOT_SUPPORTED_YET` | Multidimensional arrays not supported |
 | `ERR_POSTGRES_NULLS_IN_ARRAY_NOT_SUPPORTED_YET`         | NULL values in arrays not supported   |
 
+The client raises these errors while it decodes a result, after the server ran the query. Only that query rejects. The connection stays open, and the other queries on it continue. An `INSERT`, `UPDATE` or `DELETE ... RETURNING` can already be stored at that point, so do not retry a write blindly. To read the value, cast the column to `text` or use `.raw()`.
+
 ### Protocol Errors
 
 | Protocol Errors                         | Description                 |

src/sql/postgres/protocol/DataRow.rs

@@ -26,35 +26,29 @@ pub fn decode<C: Copy, R: ReaderContext>(
         }
         let byte_length = reader.int4()?;
         remaining_bytes -= 4;
-        match byte_length {
+        let index = u32::try_from(index).expect("int cast");
+        let more = match byte_length {
             0 => {
                 let mut empty = Data::EMPTY;
-                if !for_each(
-                    context,
-                    u32::try_from(index).expect("int cast"),
-                    Some(&mut empty),
-                )? {
-                    break;
-                }
-            }
-            NULL_INT4 => {
-                if !for_each(context, u32::try_from(index).expect("int cast"), None)? {
-                    break;
-                }
+                for_each(context, index, Some(&mut empty))
             }
+            NULL_INT4 => for_each(context, index, None),
             _ => {
                 if byte_length > remaining_bytes {
                     return Err(AnyPostgresError::InvalidMessage);
                 }
                 remaining_bytes -= byte_length;
                 let mut bytes = reader.bytes(usize::try_from(byte_length).expect("int cast"))?;
-                if !for_each(
-                    context,
-                    u32::try_from(index).expect("int cast"),
-                    Some(&mut bytes),
-                )? {
-                    break;
-                }
+                for_each(context, index, Some(&mut bytes))
+            }
+        };
+        match more {
+            Ok(true) => {}
+            Ok(false) => break,
+            // The rest of the row is skipped first, so the caller can fail this row alone.
+            Err(err) => {
+                reader.skip(usize::try_from(remaining_bytes).expect("int cast"))?;
+                return Err(err);
             }
         }
     }

src/sql_jsc/postgres/PostgresSQLConnection.rs

@@ -25,7 +25,9 @@ use crate::postgres::AuthenticationState;
 use crate::postgres::PostgresSQLQuery;
 use crate::postgres::PostgresSQLStatement;
 use crate::postgres::data_cell as DataCell;
-use crate::postgres::error_jsc::{create_postgres_error, postgres_error_to_js};
+use crate::postgres::error_jsc::{
+    create_postgres_error, postgres_error_to_js, postgres_error_to_js_with_hint,
+};
 use crate::postgres::postgres_request as PostgresRequest;
 use crate::postgres::postgres_request::MessageType;
 use crate::postgres::postgres_sql_query::{self, RequestCounter, Status as QueryStatus};
@@ -1764,6 +1766,19 @@ impl PostgresSQLConnection {
         }
     }
 
+    /// What a request rejects with for a row the client cannot decode. `Err`: the VM is stopping.
+    fn undecodable_row_error(&self, err: AnyPostgresError) -> Result<JSValue, AnyPostgresError> {
+        if self.global().has_pending_termination_exception() {
+            return Err(err);
+        }
+        Ok(postgres_error_to_js_with_hint(
+            self.global(),
+            Some(b"Failed to read data"),
+            Some(b"The query may have run on the server. The client could not decode a value in its result. Cast that column to text, or use .raw()."),
+            err,
+        ))
+    }
+
     pub(crate) fn can_prepare_query(&self) -> bool {
         let flags = self.flags.get();
         flags.contains(ConnectionFlags::IS_READY_FOR_QUERY)
@@ -2317,9 +2332,8 @@ impl PostgresSQLConnection {
         match message_type {
             MessageType::DataRow => {
                 let request = self.current().ok_or(AnyPostgresError::ExpectedRequest)?;
-                if request.status.get() == QueryStatus::Fail {
-                    // ErrorResponse already rejected this request and dropped
-                    // its GC protection; consume and discard until ReadyForQuery.
+                if request.is_rejected() {
+                    // Already rejected, GC protection dropped: discard until ReadyForQuery.
                     return reader.skip_message();
                 }
 
@@ -2383,10 +2397,14 @@ impl PostgresSQLConnection {
                 // `DataRow::decode`'s callback is `FnMut`, so capture `&mut putter`
                 // directly instead of laundering it through a raw `*mut` context —
                 // the by-value `C: Copy` slot is unused (`()`).
+                // Tells a cell the client cannot decode from a framing error of `decode` itself.
+                let mut cell_failed = false;
                 let decode_result = if request_flags.result_mode == SQLQueryResultMode::Raw {
                     protocol::DataRow::decode((), &mut reader, |(), i, b| putter.put_raw(i, b))
                 } else {
-                    protocol::DataRow::decode((), &mut reader, |(), i, b| putter.put(i, b))
+                    protocol::DataRow::decode((), &mut reader, |(), i, b| {
+                        putter.put(i, b).inspect_err(|_| cell_failed = true)
+                    })
                 };
                 // Cell cleanup (deinit each cell, then free the buffer)
                 // runs on ALL exits (decode error, to_js error, success). `putter.count` is final
@@ -2407,7 +2425,14 @@ impl PostgresSQLConnection {
                     }
                     // `if free_cells free(cells)`: heap_cells Vec drops at scope end.
                 };
-                decode_result?;
+                if let Err(err) = decode_result {
+                    if !cell_failed {
+                        return Err(err);
+                    }
+                    let js_err = self.undecodable_row_error(err)?;
+                    request.on_undecodable_row(js_err, self.global());
+                    return Ok(());
+                }
 
                 let Some(this_value) = request.this_value.get().try_get() else {
                     debug_assert!(false, "query value was freed earlier than expected");
@@ -2416,7 +2441,7 @@ impl PostgresSQLConnection {
                 let pending_value = postgres_sql_query::js::pending_value_get_cached(this_value)
                     .unwrap_or_default();
                 pending_value.ensure_still_alive();
-                let result = putter.to_js(
+                let result = match putter.to_js(
                     self.global(),
                     pending_value,
                     structure,
@@ -2425,7 +2450,14 @@ impl PostgresSQLConnection {
                     // `ParentRef::Deref` recovers `&CachedStructure`; statement
                     // outlives this call (held via `request.statement` ref).
                     cached_structure.as_deref(),
-                )?;
+                ) {
+                    Ok(result) => result,
+                    Err(err) => {
+                        let js_err = self.undecodable_row_error(err)?;
+                        request.on_undecodable_row(js_err, self.global());
+                        return Ok(());
+                    }
+                };
 
                 if pending_value.is_empty() {
                     postgres_sql_query::js::pending_value_set_cached(
@@ -2472,7 +2504,11 @@ impl PostgresSQLConnection {
                 self.socket.get().set_timeout(300);
 
                 if let Some(request) = self.current() {
-                    if request.status.get() == QueryStatus::PartialResponse {
+                    if request.flags.get().discard_response {
+                        // The end of the response it was kept in flight for.
+                        self.finish_request(&request);
+                        request.status.set(QueryStatus::Fail);
+                    } else if request.status.get() == QueryStatus::PartialResponse {
                         self.finish_request(&request);
                         // if is a partial response, just signal that the query is now complete
                         request.on_result(
@@ -2490,7 +2526,7 @@ impl PostgresSQLConnection {
             }
             MessageType::CommandComplete => {
                 let request = self.current().ok_or(AnyPostgresError::ExpectedRequest)?;
-                if request.status.get() == QueryStatus::Fail {
+                if request.is_rejected() {
                     return reader.skip_message();
                 }
 
@@ -2969,7 +3005,7 @@ impl PostgresSQLConnection {
             MessageType::CloseComplete => {
                 reader.eat_message(&protocol::CLOSE_COMPLETE)?;
                 let request = self.current().ok_or(AnyPostgresError::ExpectedRequest)?;
-                if request.status.get() == QueryStatus::Fail {
+                if request.is_rejected() {
                     return Ok(());
                 }
                 request.on_result(
@@ -2996,7 +3032,7 @@ impl PostgresSQLConnection {
             MessageType::EmptyQueryResponse => {
                 reader.eat_message(&protocol::EMPTY_QUERY_RESPONSE)?;
                 let request = self.current().ok_or(AnyPostgresError::ExpectedRequest)?;
-                if request.status.get() == QueryStatus::Fail {
+                if request.is_rejected() {
                     return Ok(());
                 }
                 request.on_result(b"", self.global(), self.js_value.get().get(), false);

src/sql_jsc/postgres/PostgresSQLQuery.rs

@@ -77,6 +77,8 @@ pub struct Flags {
     pub(crate) binary: bool,
     pub(crate) bigint: bool,
     pub(crate) simple: bool,
+    /// Rejected for an undecodable row: in flight, its response skipped, until `ReadyForQuery`.
+    pub(crate) discard_response: bool,
     /// Which connection counter this request's dispatch incremented; reset to
     /// `None` when `finish_request` consumes that contribution, so the
     /// decrement is idempotent across its call sites.
@@ -100,6 +102,7 @@ impl Default for Flags {
             binary: false,
             bigint: false,
             simple: false,
+            discard_response: false,
             counter: RequestCounter::None,
             result_mode: PostgresSQLQueryResultMode::Objects,
         }
@@ -215,9 +218,23 @@ impl PostgresSQLQuery {
     }
 
     pub(crate) fn on_js_error(&self, err: JSValue, global_object: &JSGlobalObject) {
+        self.status.set(Status::Fail);
+        self.reject(err, global_object);
+    }
+
+    /// Rejects now, but `status` stays in flight: the server is still answering this query.
+    pub(crate) fn on_undecodable_row(&self, err: JSValue, global_object: &JSGlobalObject) {
+        self.update_flags(|f| f.discard_response = true);
+        self.reject(err, global_object);
+    }
+
+    pub(crate) fn is_rejected(&self) -> bool {
+        self.status.get() == Status::Fail || self.flags.get().discard_response
+    }
+
+    fn reject(&self, err: JSValue, global_object: &JSGlobalObject) {
         // R-2: see `on_write_fail` — `&self` + Cell/JsCell, RefPtr brackets re-entry.
         let _guard = self.ref_guard();
-        self.status.set(Status::Fail);
         let Some(this_value) = self.this_value.get().try_get() else {
             return;
         };

src/sql_jsc/postgres/error_jsc.rs

@@ -52,6 +52,15 @@ pub(crate) fn postgres_error_to_js(
     global: &JSGlobalObject,
     message: Option<&[u8]>,
     err: AnyPostgresError,
+) -> JSValue {
+    postgres_error_to_js_with_hint(global, message, None, err)
+}
+
+pub(crate) fn postgres_error_to_js_with_hint(
+    global: &JSGlobalObject,
+    message: Option<&[u8]>,
+    hint: Option<&[u8]>,
+    err: AnyPostgresError,
 ) -> JSValue {
     use AnyPostgresError::*;
     let code: &'static [u8] = match err {
@@ -134,6 +143,7 @@ pub(crate) fn postgres_error_to_js(
         msg,
         &PostgresErrorOptions {
             code,
+            hint,
             ..Default::default()
         },
     ) {