Construct error messages properly when libpq returns NULL result

I had originally thought that this was some weird condition that was the
result of the server getting a query with more than `34464` (`i16::MAX`)
bind parameters, where we got back an error but it had no message. We
panic in that case, as PG is documented that all errors have a message.
It turns out I was wrong for two reasons.

First, the number is 65535 (`u16::MAX`), not `34464`. (This might have
been something changed on the PG side in 9.6 which was released since I
last looked at this). Second, we were not getting back an error with no
message, we were getting a `NULL` return value from libpq. From [their
docs][]:

[their docs]: https://www.postgresql.org/docs/9.6/static/libpq-exec.html#LIBPQ-PQPREPARE

> A null result indicates out-of-memory or inability to send the command
> at all. Use PQerrorMessage to get more information about such errors.

`PQerrorMessage` means the error is on the connection. The reason that
we had no result was that the PG wire protocol uses a `u16` for the
number of bind parameters, so there's no way for the driver to have even
sent the message to the server.

Once we realized the error condition that we were seeing is more general
than just the bind parameter case, I've refactored the code to perform a
not null check as early as possible, and hide the `*mut PGresult`.

The test itself is written in a somewhat strange fashion. We're
constructing a query that has 70_000 bind parameters, as this is the
only case that I know of which will always cause libpq to return null. I
don't care what the error message is, just that it's not an empty
string. I've also written the test to explicitly match instead of unwrap
and use `#[should_panic]` because I want to be sure that it *didn't*
panic.

Ideally we would never see `*mut PGresult` anywhere in the code, and use
`Unique<PGresult>` instead. However, it is currently unstable. I had
originally implemented this to use `Unique` if `feature = "unstable"`
was set, but deleted it as the `#[cfg]` branches made the code harder to
comprehend. The only real benefit `Unique` gives us here is clarity of
intent. For the not-null and non-aliased hints to result in any
optimization by the compiler would require libpq to be statically
linked, and LTO to be happening at the LLVM layer, neither of which are
commonly true.

We know that `Send` and `Sync` are safe to implement on our new struct,
because the pointer is unaliased. We can delete those impls if we switch
to using `Unique` in the future, as it already implements those traits.

Fixes diesel-rs#446.

Author: sgrif

CHANGELOG.md

@@ -17,6 +17,10 @@ for Rust libraries in [RFC #1105](https://github.com/rust-lang/rfcs/blob/master/
 * `debug_sql!` can now properly be used with types from `chrono` or
   `std::time`.
 
+* When using PostgreSQL, attempting to get the error message of a query which
+  could not be transmitted to the server (such as a query with greater than
+  65535 bind parameters) will no longer panic.
+
 ## [0.9.0] - 2016-12-08
 
 ### Added

diesel/src/pg/connection/mod.rs

@@ -41,7 +41,7 @@ impl SimpleConnection for PgConnection {
         let inner_result = unsafe {
             self.raw_connection.exec(query.as_ptr())
         };
-        try!(PgResult::new(inner_result));
+        try!(PgResult::new(inner_result?));
         Ok(())
     }
 }

diesel/src/pg/connection/raw.rs

@@ -57,8 +57,8 @@ impl RawConnection {
         }
     }
 
-    pub unsafe fn exec(&self, query: *const libc::c_char) -> *mut PGresult {
-        PQexec(self.internal_connection, query)
+    pub unsafe fn exec(&self, query: *const libc::c_char) -> QueryResult<RawResult> {
+        RawResult::new(PQexec(self.internal_connection, query), self)
     }
 
     pub unsafe fn exec_params(
@@ -70,8 +70,8 @@ impl RawConnection {
         param_lengths: *const libc::c_int,
         param_formats: *const libc::c_int,
         result_format: libc::c_int,
-    ) -> *mut PGresult {
-        PQexecParams(
+    ) -> QueryResult<RawResult> {
+        let ptr = PQexecParams(
             self.internal_connection,
             query,
             param_count,
@@ -80,7 +80,8 @@ impl RawConnection {
             param_lengths,
             param_formats,
             result_format,
-        )
+        );
+        RawResult::new(ptr, self)
     }
 
     pub unsafe fn exec_prepared(
@@ -91,16 +92,17 @@ impl RawConnection {
         param_lengths: *const libc::c_int,
         param_formats: *const libc::c_int,
         result_format: libc::c_int,
-    ) -> *mut PGresult {
-        PQexecPrepared(
+    ) -> QueryResult<RawResult> {
+        let ptr = PQexecPrepared(
             self.internal_connection,
             stmt_name,
             param_count,
             param_values,
             param_lengths,
             param_formats,
             result_format,
-        )
+        );
+        RawResult::new(ptr, self)
     }
 
     pub unsafe fn prepare(
@@ -109,14 +111,15 @@ impl RawConnection {
         query: *const libc::c_char,
         param_count: libc::c_int,
         param_types: *const Oid,
-    ) -> *mut PGresult {
-        PQprepare(
+    ) -> QueryResult<RawResult> {
+        let ptr = PQprepare(
             self.internal_connection,
             stmt_name,
             query,
             param_count,
             param_types,
-        )
+        );
+        RawResult::new(ptr, self)
     }
 }
 
@@ -167,3 +170,37 @@ impl Drop for PgString {
         }
     }
 }
+
+/// Internal wrapper around a `*mut PGresult` which is known to be not-null, and
+/// have no aliases.  This wrapper is to ensure that it's always properly
+/// dropped.
+///
+/// If `Unique` is ever stabilized, we should use it here.
+#[allow(missing_debug_implementations)]
+pub struct RawResult(*mut PGresult);
+
+unsafe impl Send for RawResult {}
+unsafe impl Sync for RawResult {}
+
+impl RawResult {
+    fn new(ptr: *mut PGresult, conn: &RawConnection) -> QueryResult<Self> {
+        if ptr.is_null() {
+            Err(Error::DatabaseError(
+                DatabaseErrorKind::UnableToSendCommand,
+                Box::new(conn.last_error_message()),
+            ))
+        } else {
+            Ok(RawResult(ptr))
+        }
+    }
+
+    pub fn as_ptr(&self) -> *mut PGresult {
+        self.0
+    }
+}
+
+impl Drop for RawResult {
+    fn drop(&mut self) {
+        unsafe { PQclear(self.0) }
+    }
+}

diesel/src/pg/connection/result.rs

@@ -3,38 +3,39 @@ extern crate libc;
 
 use result::{Error, QueryResult, DatabaseErrorInformation, DatabaseErrorKind};
 use super::row::PgRow;
+use super::raw::RawResult;
 
 use self::pq_sys::*;
 use std::ffi::CStr;
 use std::{str, slice};
 
 pub struct PgResult {
-    internal_result: *mut PGresult,
+    internal_result: RawResult,
 }
 
 impl PgResult {
-    pub fn new(internal_result: *mut PGresult) -> QueryResult<Self> {
-        let result_status = unsafe { PQresultStatus(internal_result) };
+    pub fn new(internal_result: RawResult) -> QueryResult<Self> {
+        let result_status = unsafe { PQresultStatus(internal_result.as_ptr()) };
         match result_status {
             PGRES_COMMAND_OK | PGRES_TUPLES_OK => {
                 Ok(PgResult {
                     internal_result: internal_result,
                 })
             },
             _ => {
-                let error_information = Box::new(PgErrorInformation(internal_result));
-                let error_kind = match get_result_field(internal_result, ResultField::SqlState) {
+                let error_kind = match get_result_field(internal_result.as_ptr(), ResultField::SqlState) {
                     Some(error_codes::UNIQUE_VIOLATION) => DatabaseErrorKind::UniqueViolation,
                     _ => DatabaseErrorKind::__Unknown,
                 };
+                let error_information = Box::new(PgErrorInformation(internal_result));
                 Err(Error::DatabaseError(error_kind, error_information))
             }
         }
     }
 
     pub fn rows_affected(&self) -> usize {
         unsafe {
-            let count_char_ptr = PQcmdTuples(self.internal_result);
+            let count_char_ptr = PQcmdTuples(self.internal_result.as_ptr());
             let count_bytes = CStr::from_ptr(count_char_ptr).to_bytes();
             let count_str = str::from_utf8_unchecked(count_bytes);
             match count_str {
@@ -45,7 +46,7 @@ impl PgResult {
     }
 
     pub fn num_rows(&self) -> usize {
-        unsafe { PQntuples(self.internal_result) as usize }
+        unsafe { PQntuples(self.internal_result.as_ptr()) as usize }
     }
 
     pub fn get_row(&self, idx: usize) -> PgRow {
@@ -59,8 +60,9 @@ impl PgResult {
             let row_idx = row_idx as libc::c_int;
             let col_idx = col_idx as libc::c_int;
             unsafe {
-                let value_ptr = PQgetvalue(self.internal_result, row_idx, col_idx) as *const u8;
-                let num_bytes = PQgetlength(self.internal_result, row_idx, col_idx);
+                let value_ptr = PQgetvalue(self.internal_result.as_ptr(), row_idx, col_idx)
+                    as *const u8;
+                let num_bytes = PQgetlength(self.internal_result.as_ptr(), row_idx, col_idx);
                 Some(slice::from_raw_parts(value_ptr, num_bytes as usize))
             }
         }
@@ -69,56 +71,42 @@ impl PgResult {
     pub fn is_null(&self, row_idx: usize, col_idx: usize) -> bool {
         unsafe {
             0 != PQgetisnull(
-                self.internal_result,
+                self.internal_result.as_ptr(),
                 row_idx as libc::c_int,
                 col_idx as libc::c_int,
             )
         }
     }
 }
 
-impl Drop for PgResult {
-    fn drop(&mut self) {
-        unsafe { PQclear(self.internal_result) };
-    }
-}
-
-struct PgErrorInformation(*mut PGresult);
-
-unsafe impl Send for PgErrorInformation {}
-
-impl Drop for PgErrorInformation {
-    fn drop(&mut self) {
-        unsafe { PQclear(self.0) };
-    }
-}
+struct PgErrorInformation(RawResult);
 
 impl DatabaseErrorInformation for PgErrorInformation {
     fn message(&self) -> &str {
-        match get_result_field(self.0, ResultField::MessagePrimary) {
+        match get_result_field(self.0.as_ptr(), ResultField::MessagePrimary) {
             Some(e) => e,
             None => unreachable!("Per PGs documentation, all errors should have a message"),
         }
     }
 
     fn details(&self) -> Option<&str> {
-        get_result_field(self.0, ResultField::MessageDetail)
+        get_result_field(self.0.as_ptr(), ResultField::MessageDetail)
     }
 
     fn hint(&self) -> Option<&str> {
-        get_result_field(self.0, ResultField::MessageHint)
+        get_result_field(self.0.as_ptr(), ResultField::MessageHint)
     }
 
     fn table_name(&self) -> Option<&str> {
-        get_result_field(self.0, ResultField::TableName)
+        get_result_field(self.0.as_ptr(), ResultField::TableName)
     }
 
     fn column_name(&self) -> Option<&str> {
-        get_result_field(self.0, ResultField::ColumnName)
+        get_result_field(self.0.as_ptr(), ResultField::ColumnName)
     }
 
     fn constraint_name(&self) -> Option<&str> {
-        get_result_field(self.0, ResultField::ConstraintName)
+        get_result_field(self.0.as_ptr(), ResultField::ConstraintName)
     }
 }
 

diesel/src/pg/connection/stmt/mod.rs

@@ -63,7 +63,7 @@ impl Query {
             }
         };
 
-        PgResult::new(internal_res)
+        PgResult::new(internal_res?)
     }
 
     pub fn sql(sql: &str, param_types: Option<Vec<u32>>) -> QueryResult<Self> {
@@ -90,7 +90,7 @@ impl Query {
                 param_types_to_ptr(Some(&param_types)),
             )
         };
-        try!(PgResult::new(internal_result));
+        try!(PgResult::new(internal_result?));
 
         Ok(Query::Prepared {
             name: name,

diesel/src/result.rs

@@ -26,6 +26,7 @@ pub enum Error {
 /// bump.
 pub enum DatabaseErrorKind {
     UniqueViolation,
+    UnableToSendCommand,
     #[doc(hidden)]
     __Unknown, // Match against _ instead, more variants may be added in the future
 }

diesel_tests/tests/internal_details.rs

@@ -11,3 +11,23 @@ fn bind_params_are_passed_for_null_when_not_inserting() {
         .filter(AsExpression::<Nullable<Integer>>::as_expression(None::<i32>).is_null());
     assert_eq!(Ok(1), query.first(&connection));
 }
+
+#[test]
+#[cfg(feature = "postgres")]
+fn query_which_cannot_be_transmitted_gives_proper_error_message() {
+    use schema::comments::dsl::*;
+    use diesel::result::Error::DatabaseError;
+    use diesel::result::DatabaseErrorKind::UnableToSendCommand;
+
+    // Create a query with 90000 binds, 2 binds per row
+    let data: &[NewComment<'static>] = &[NewComment(1, "hi"); 45_000];
+    let query_with_to_many_binds = insert(data).into(comments);
+
+    match query_with_to_many_binds.execute(&connection()) {
+        Ok(_) => panic!("We successfully executed a query with 90k binds. \
+            We need to find a new query to test which can't be represented by \
+            the wire protocol."),
+        Err(DatabaseError(UnableToSendCommand, info)) => assert_ne!("", info.message()),
+        Err(_) => panic!("We got back the wrong kind of error. This test is invalid."),
+    }
+}

diesel_tests/tests/schema.rs

@@ -126,7 +126,7 @@ impl NewPost {
     }
 }
 
-#[derive(Insertable)]
+#[derive(Debug, Clone, Copy, Insertable)]
 #[table_name="comments"]
 pub struct NewComment<'a>(
     #[column_name(post_id)]