Repository: diesel-rs/diesel (language: rust)

Issue:
DatabaseError "another command is already in progress\n" upon using the same db connection multiple times
## Setup

### Versions

- **Rust:** 1.63.0
- **Diesel:** v2.0.0
- **Database:** PostgreSQL
- **Operating System** Linux

### Feature Flags

- **diesel:** ["postgres", "chrono"]

## Problem Description
It looks like while making two queries using the same connection (not concurrently), from time to time, first query left the connection (or channel to the db) in "not usable state" and subsequent query failed with the following error,
`DatabaseError(UnableToSendCommand, "another command is already in progress\n")`

> A null pointer is returned when the command is complete and there will be no more results.

From [libpq docs](https://www.postgresql.org/docs/current/libpq-async.html#LIBPQ-PQGETRESULT). As per my understanding, before using the same connection for another query, we need to "clean" connection/channel by fetching result repeatedly until we get NULL. I managed to test this hypothesis and it worked, but I'm not sure whether it's the right solution. For more information about my changes, see "Steps to reproduce".

### What are you trying to accomplish?
The use-case is, 
1. create a connection 
2. try to insert a record 
3. upon failing on 'duplicate violation' error, fetch record.

The steps 2 and 3 can be done multiple times using the same connection. [think pool manger provides an existing connection] 
FYI: the problem is reproducible without any pool manager.

### What is the expected output?
The same use-case flow works first time using newly created connection. Upon using the same connection to do the same flow next time, database error is being thrown on 3rd step (fetch record).
The expected output is to be able to use the same connection for as many queries as possible, sequentially. In this case, fetch record without throwing error.

### What is the actual output?
Throws `DatabaseError` whereas it should fetch the record.

### Steps to reproduce
Here is the public repo, https://github.com/heyrutvik/diesel-error-demo, which reproduces the problem. It uses [my branch of diesel](https://github.com/heyrutvik/diesel/commit/1a8e2d84a672d3d8f5bee4a217cbc896314950e8) that I created to debug the issue but one can change `Cargo.toml` to uncomment a line to use diesel `2.0.0` version. Please read README which has "extended" logs and explanation. I also tried to explain the scenario in its `main.rs` file. There is also a possible solution explained in the README.

## Checklist

- [x] I have already looked over the [issue tracker](https://github.com/diesel-rs/diesel/issues)  and the [disussion forum](https://github.com/diesel-rs/diesel/discussions)  for similar possible closed issues.
- [x] This issue can be reproduced on Rust's stable channel. (Your issue will be
  closed if this is not the case)
- [x] This issue can be reproduced without requiring a third party crate

I believe README and comments in `main.rs` are enough to understand the issue but feel free to ping me your questions. 


Discussion / hints:
Thanks for the great bugreport and for providing a minimal reproducible example. This sounds like a duplicate of https://github.com/diesel-rs/diesel/issues/3282 (which is missing a reproducible example), so it might be useful to add these information there and close this issue as duplicate. 

Its a bit surprising to see that happen in the first way, because we made sure that we poll results till null is returned. That happens in `PgConnection::execute_returning_count` (for anything that uses `ExecuteDsl`), and in the corresponding cursor implementations in `pg/connection/cursor.rs` (for anything that loads data). The streaming cursor uses the drop implementation for this, the ”normal" cursor has a debug_assert that checks for exactly that condition. The example you've provided uses the "normal" cursor implementation and manages to somehow create an situation where the assert is not triggered, but the condition does not hold true. I will need to check why this happens, but that should be fixable.
I had another lock at the relevant code. As often you spot the error if you have an idea where to look. I think [this line](https://github.com/diesel-rs/diesel/blob/master/diesel/src/pg/connection/mod.rs#L225) is the problematic one. We return on error there, without even reaching the relevant assert at all. We should be able to fix this by just "clearing" the connection [here](https://github.com/diesel-rs/diesel/blob/master/diesel/src/pg/connection/result.rs#L50) (this is internally called a few lines above). Mind submitting a PR that:
* adds a test case that triggers this bug to diesel_tests
* adds the polling loop to the linked error match arm?
Thank you @weiznich for promptly looking into it and providing the solution. Here is what I understand from your messages and my tinkering with the code, once the first query hits some error, in this case a unique violation error, we need to "clear" the connection before letting it be used for the next command. Is this reasoning correct? This could be a reason why you suggested adding polling [here](https://github.com/diesel-rs/diesel/blob/master/diesel/src/pg/connection/result.rs#L50) to "clear" the connection.

I would be more than happy helping the project which makes my life a lot easier. :) I'll submit a PR.

Fix patch:
diff --git a/diesel/src/pg/connection/result.rs b/diesel/src/pg/connection/result.rs
index 3036c5d7ec1d..b6169ae6a9a8 100644
--- a/diesel/src/pg/connection/result.rs
+++ b/diesel/src/pg/connection/result.rs
@@ -48,6 +48,12 @@ impl PgResult {
                 ))
             }
             _ => {
+                // "clearing" the connection by polling result till we get a null.
+                // this will indicate that the previous command is complete and
+                // the same connection is ready to process next command.
+                // https://www.postgresql.org/docs/current/libpq-async.html
+                while conn.get_next_result().map_or(true, |r| r.is_some()) {}
+
                 let mut error_kind =
                     match get_result_field(internal_result.as_ptr(), ResultField::SqlState) {
                         Some(error_codes::UNIQUE_VIOLATION) => DatabaseErrorKind::UniqueViolation,


Test patch:
diff --git a/diesel_tests/tests/connection.rs b/diesel_tests/tests/connection.rs
index 9625bd94b362..954818b5fbfe 100644
--- a/diesel_tests/tests/connection.rs
+++ b/diesel_tests/tests/connection.rs
@@ -2,22 +2,23 @@ use crate::schema::*;
 use diesel::connection::BoxableConnection;
 use diesel::*;
 
-table! {
-    auto_time {
-        id -> Integer,
-        n -> Integer,
-        updated_at -> Timestamp,
-    }
-}
-
 #[test]
 #[cfg(any(feature = "postgres", feature = "sqlite"))]
 fn managing_updated_at_for_table() {
-    use self::auto_time::dsl::*;
     use crate::schema_dsl::*;
     use chrono::NaiveDateTime;
     use std::{thread, time::Duration};
 
+    table! {
+        #[sql_name = "auto_time"]
+            auto_time_table {
+            id -> Integer,
+            n -> Integer,
+            updated_at -> Timestamp,
+        }
+    }
+    use auto_time_table::dsl::{auto_time_table as auto_time, n, updated_at};
+
     // transactions have frozen time, so we can't use them
     let connection = &mut connection_without_transaction();
     create_table(
@@ -116,3 +117,77 @@ fn boxable_connection_downcast_mut_usable() {
 
     assert_eq!(Ok(String::from("Sean")), sean);
 }
+
+#[test]
+#[cfg(feature = "postgres")]
+fn use_the_same_connection_multiple_times() {
+    use crate::*;
+    use diesel::result::DatabaseErrorKind;
+    use diesel::result::Error::DatabaseError;
+
+    table! {
+        #[sql_name = "github_issue_3342"]
+        github_issue_3342_table {
+            id -> Serial,
+            uid -> Integer,
+        }
+    }
+    use github_issue_3342_table::dsl::{github_issue_3342_table as github_issue_3342, uid};
+
+    let connection = &mut connection_without_transaction();
+
+    // We can extend `schema_dsl` module to accommodate UNIQUE constraint.
+    sql_query(
+        r#"
+          CREATE TABLE github_issue_3342 (
+            id SERIAL PRIMARY KEY,
+            uid INTEGER NOT NULL UNIQUE
+          )
+        "#,
+    )
+    .execute(connection)
+    .unwrap();
+
+    let mut _drop_conn = connection_without_transaction();
+    let _guard = DropTable {
+        connection: &mut _drop_conn,
+        table_name: "github_issue_3342",
+        can_drop: true,
+    };
+
+    // helper method to simulate database error.
+    fn insert_or_fetch(conn: &mut PgConnection, input: i32) {
+        let result = insert_into(github_issue_3342)
+            .values(uid.eq(input))
+            .get_result::<(i32, i32)>(conn);
+
+        match result {
+            Ok((_, r)) => assert_eq!(r, input),
+            Err(DatabaseError(DatabaseErrorKind::UniqueViolation, _)) => {
+                let result = github_issue_3342
+                    .filter(uid.eq(input))
+                    .first::<(i32, i32)>(conn);
+                match result {
+                    Ok((_, r)) => assert_eq!(r, input),
+                    Err(DatabaseError(DatabaseErrorKind::UnableToSendCommand, message))
+                        if message.message() == "another command is already in progress\n" =>
+                    {
+                        panic!("The fix didn't solve the problem!?")
+                    }
+                    Err(e) => panic!("Caused by: {}", e),
+                }
+            }
+            Err(DatabaseError(DatabaseErrorKind::UnableToSendCommand, message))
+                if message.message() == "another command is already in progress\n" =>
+            {
+                panic!("The fix didn't solve the problem!?")
+            }
+            Err(e) => panic!("Caused by: {}", e),
+        }
+    }
+
+    // simulate multiple queries using the same connection sequentially.
+    for _ in 0..5 {
+        insert_or_fetch(connection, 1);
+    }
+}