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

Issue:
A `ReadOnlyTransaction` error within an inner transaction doesn't rollback the savepoint
## Setup

### Versions

- **Rust:** 1.66
- **Diesel:** 2.0.2
- **Database:** PostgreSQL 14.6
- **Operating System:** Ubuntu 22.10

### Feature Flags

- **diesel:** ["postgres"]

## Problem Description

Starting with diesel `2.0.0-rc.1` (commit 53bad857f30f2f32c14ae295487a88da278fb2fc of #3211) when within a nested transaction, if the inner transaction aborts because it attempts to mutate within a read-only transaction, then the connection is left in a bad state even outside of the inner transaction. Any further commands result in `current transaction is aborted, commands ignored until end of transaction block` until the outermost transaction is also rolled back. It appears that the SAVEPOINT for the inner transaction is not actually rolled back, as the equivalent series of SQL commands (including a `ROLLBACK TO SAVEPOINT ...`) does not result in such an error in `psql`.

### What are you trying to accomplish?

I am trying to port crates.io to diesel 2.0. Our tests are run within an outer transaction that rolls back when the test completes. In production the code will continue to work as expected as this code path has just the one transaction, but the additional transaction wrapping the entire test causes this test to fail.

### What is the expected output?

It is expected that if a nested transaction encounters a `ReadOnlyTransaction` error then it should still rollback to the appropriate SAVEPOINT so that future read-only commands (such as SELECT) can be executed.

### What is the actual output?

The observed output is an error stating `current transaction is aborted, commands ignored until end of transaction block`.

### Are you seeing any additional errors?

No additional errors are seen.

### Steps to reproduce

Create a simple test table with `cargo run` and then run `cargo test`. In diesel releases starting with `2.0.0-rc.1` the 2nd test fails with `Error: DatabaseError(Unknown, "current transaction is aborted, commands ignored until end of transaction block")`.

```rust
use diesel::{prelude::*, sql_query};

const DATABASE_URL: &str = "postgres:///";

fn connect() -> PgConnection {
    PgConnection::establish(DATABASE_URL).unwrap()
}

fn main() {
    sql_query("CREATE TABLE rollback_test (id INT PRIMARY KEY, value INT NOT NULL)")
        .execute(&mut connect())
        .unwrap();
}

#[cfg(test)]
mod tests {
    use super::*;
    use diesel::update;

    diesel::table! {
        rollback_test (id) {
            id -> Int4,
            value -> Int4,
        }
    }

    fn update_with_fallback_query(conn: &mut PgConnection) -> QueryResult<()> {
        conn.transaction(|conn| {
            sql_query("SET TRANSACTION READ ONLY").execute(conn)?;
            dbg!(update(rollback_test::table)
                .set(rollback_test::value.eq(0))
                .execute(conn))
        })
        .or_else(|_| sql_query("SELECT 1").execute(conn))
        .map(|_| ())
    }

    #[test]
    fn test_rollback_of_readonly() -> QueryResult<()> {
        update_with_fallback_query(&mut connect())
    }

    /// Test that start failing at rev="53bad857f30f2f32c14ae295487a88da278fb2fc"
    #[test]
    fn test_rollback_of_readonly_within_outer_transaction() -> QueryResult<()> {
        connect().transaction(|conn| update_with_fallback_query(conn))
    }
}
```

For comparison, here are similar SQL commands run via `psql`:

```sql
BEGIN;
SAVEPOINT rollback_test;
SET TRANSACTION READ ONLY;
UPDATE rollback_test SET value=1;
-- ERROR:  cannot execute UPDATE in a read-only transaction
SELECT 1;
-- ERROR:  current transaction is aborted, commands ignored until end of transaction block
ROLLBACK TO SAVEPOINT rollback_test;
SELECT 1;
-- No longer an error, as we've left the inner transaction.
-- So diesel must not be issuing the expected savepoint rollback.
COMMIT;
```

## 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


Discussion / hints:
Thanks for opening this issue and providing a reproduction. I will try to find a solution as soon as possible.
The following post is just a dump of things that I've tried:

* Diesel executes the following commands for your testcase :
```sql
BEGIN;
SAVEPOINT diesel_savepoint_1;
SET TRANSACTION READ ONLY;
update rollback_test SET value=1;
-- ERROR: readonly transaction
SELECT 1; 
-- ERROR: current transaction is aborted, commands ignored until end of transaction block
```
Executing these commands in psql reproduces the same error. The issue is caused by the missing `ROLLBACK TO SAVEPOINT` command. That changed in the referenced commit.

* For the postgres connection we use `libpq` to get the current transaction state. `libpq` reports that the transaction state is in an error state [here](https://github.com/diesel-rs/diesel/blob/master/diesel/src/pg/connection/mod.rs#L243). This causes us to mark the connection as requiring a top level rollback and skipping all `ROLLBACK TO SAVEPOINT …` commands. The main motivation for this is to correctly handle transactions that fail due to serialization errors. 

That all written: I'm not 100% sure yet how to fix that. It might be worth to just try to perform the `ROLLBACK TO SAVEPOINT` nevertheless and ignore potential errors. 

cc @Ten0 as author of the relevant commit
Pretty raw thoughts - couldn't investigate in detail on some points below:

To give more context about this particular change, Diesel 1 had poor tracking of transactions failing, which sometimes led to Diesel thinking the depth was a certain level when in fact it was at a different level (#2123). This initially was fixed by #2935, of which however the interface was not ideal (made matching on errors difficult, backend-specific code was polluting the general code, there were a few bugs left)
#3211 improves all of these points before the 2.0 release, and notably simplifies it and makes it more reliable by using the information sent back by the postgres server about the transaction state instead of trying to guess it from what the last query was and its result. 

So overall the current implementation is probably the safest way to fix an improper behavior we used to have where we would consider as not broken stuff that was broken, but that comes at the cost of apparently over-considering-broken stuff that could be recovered in a more fine-grained manner. 

I can't imagine that serialization failures would be always fixed by savepoint rollback, so it looks like we shouldn't take this libpq status as "you may rollback just one level and you should be good": typically it instead means "you should rollback top-level" (which is the case for serialization failures - or can savepoints recover some of them?), or it's a programmatic misuse (your test case IIUC). 

So it looks like we have essentially two solutions:
- Some specific (consequently risky - not sure to what extent) special casing, trying to identify that precise error and ignoring the in-error status, relying on calling code to try the corresponding rollback and somehow ignoring the status for subsequent rollbacks to savepoints as well. 
- Rewriting these particular tests in such a way that they don't rely on savepoints to recover from database misuse

An alternate track (if it works) could maybe be to always call the save points rollbacks (not flag top level as broken but just low level as broken), but it looks like that would have a cost on production applications that may legitimately hit serialization failures. Unless these can be recovered by savepoints (can they, sometimes?) I would tend to not damage regular use in order to allow better workflow in testing database-misuse (your case IIUC). 
For additional context, our use case looks like this:

```rust
let tokens = api_tokens
    .filter(revoked.eq(false))
    .filter(token.eq(&token_));
conn.transaction(|conn| {
    update(tokens)
        .set(last_used_at.eq(now.nullable()))
        .get_result(conn)
})
.or_else(|_| tokens.first(conn))
```

First we try to obtain the token while updating the `last_used_at` time, but if that fails then we basically rerun the same query but as a `SELECT`. The connection could be read-only because the primary database is offline, or we could be doing maintenance. I did a bit of searching and didn't find another way to determine if the connection is read-only, and we do want test coverage to make sure this is working correctly.

Fortunately, I think we may be able to drop this fallback. We've reduced the endpoints that are allowed to authenticate with an API token and in the 1 remaining case where this fallback helps, it should be fine to just reject the requests when operating in a degraded mode. Though I don't think the general approach is that extreme and it would be nice if Diesel supported it.

> Diesel 1 had poor tracking of transactions failing, which sometimes led to Diesel thinking the depth was a certain level when in fact it was at a different level

I'm slowly working my way though some of the relevant logic, but I'm confused by this. It seems that is how [this logic](https://github.com/diesel-rs/diesel/blob/a02fcfbf57311a220e69dab03dd17e65a61313e9/diesel/src/connection/transaction_manager.rs#L342-L360) behaves, decreasing diesel's internal count without issuing the rollback. I don't know if it would work for the serialization use case, but perhaps we could attempt the rollback first and then test the connection status.

Fix patch:
diff --git a/diesel/src/connection/transaction_manager.rs b/diesel/src/connection/transaction_manager.rs
index 1a6de1f390e9..9901d64a1e39 100644
--- a/diesel/src/connection/transaction_manager.rs
+++ b/diesel/src/connection/transaction_manager.rs
@@ -135,24 +135,6 @@ impl TransactionManagerStatus {
         }
     }
 
-    #[cfg(any(
-        feature = "i-implement-a-third-party-backend-and-opt-into-breaking-changes",
-        feature = "postgres",
-    ))]
-    #[diesel_derives::__diesel_public_if(
-        feature = "i-implement-a-third-party-backend-and-opt-into-breaking-changes"
-    )]
-    /// Whether we may be interested in calling
-    /// `set_top_level_transaction_requires_rollback_if_not_broken`
-    ///
-    /// You should typically not need this outside of a custom backend implementation
-    pub(crate) fn is_not_broken_and_in_transaction(&self) -> bool {
-        match self {
-            TransactionManagerStatus::Valid(valid_status) => valid_status.in_transaction.is_some(),
-            TransactionManagerStatus::InError => false,
-        }
-    }
-
     #[cfg(any(
         feature = "i-implement-a-third-party-backend-and-opt-into-breaking-changes",
         feature = "postgres",
@@ -162,18 +144,21 @@ impl TransactionManagerStatus {
     #[diesel_derives::__diesel_public_if(
         feature = "i-implement-a-third-party-backend-and-opt-into-breaking-changes"
     )]
-    /// If in transaction and transaction manager is not broken, registers that the
-    /// connection can not be used anymore until top-level transaction is rolled back
-    pub(crate) fn set_top_level_transaction_requires_rollback(&mut self) {
+    /// If in transaction and transaction manager is not broken, registers that it's possible that
+    /// the connection can not be used anymore until top-level transaction is rolled back.
+    ///
+    /// If that is registered, savepoints rollbacks will still be attempted, but failure to do so
+    /// will not result in an error. (Some may succeed, some may not.)
+    pub(crate) fn set_requires_rollback_maybe_up_to_top_level(&mut self, to: bool) {
         if let TransactionManagerStatus::Valid(ValidTransactionManagerStatus {
             in_transaction:
                 Some(InTransactionStatus {
-                    top_level_transaction_requires_rollback,
+                    requires_rollback_maybe_up_to_top_level,
                     ..
                 }),
         }) = self
         {
-            *top_level_transaction_requires_rollback = true;
+            *requires_rollback_maybe_up_to_top_level = to;
         }
     }
 
@@ -213,7 +198,9 @@ pub struct ValidTransactionManagerStatus {
 #[derive(Debug)]
 struct InTransactionStatus {
     transaction_depth: NonZeroU32,
-    top_level_transaction_requires_rollback: bool,
+    /// If that is registered, savepoints rollbacks will still be attempted, but failure to do so
+    /// will not result in an error. (Some may succeed, some may not.)
+    requires_rollback_maybe_up_to_top_level: bool,
     test_transaction: bool,
 }
 
@@ -252,7 +239,7 @@ impl ValidTransactionManagerStatus {
             (None, TransactionDepthChange::IncreaseDepth) => {
                 self.in_transaction = Some(InTransactionStatus {
                     transaction_depth: NonZeroU32::new(1).expect("1 is non-zero"),
-                    top_level_transaction_requires_rollback: false,
+                    requires_rollback_maybe_up_to_top_level: false,
                     test_transaction: false,
                 });
                 Ok(())
@@ -332,40 +319,38 @@ where
     fn rollback_transaction(conn: &mut Conn) -> QueryResult<()> {
         let transaction_state = Self::get_transaction_state(conn)?;
 
-        let rollback_sql = match transaction_state.in_transaction {
-            Some(ref mut in_transaction) => {
+        let (
+            (rollback_sql, rolling_back_top_level),
+            requires_rollback_maybe_up_to_top_level_before_execute,
+        ) = match transaction_state.in_transaction {
+            Some(ref in_transaction) => (
                 match in_transaction.transaction_depth.get() {
-                    1 => Cow::Borrowed("ROLLBACK"),
-                    depth_gt1 => {
-                        if in_transaction.top_level_transaction_requires_rollback {
-                            // There's no point in *actually* rolling back this one
-                            // because we won't be able to do anything until top-level
-                            // is rolled back.
-
-                            // To make it easier on the user (that they don't have to really look
-                            // at actual transaction depth and can just rely on the number of
-                            // times they have called begin/commit/rollback) we don't mark the
-                            // transaction manager as out of the savepoints as soon as we
-                            // realize there is that issue, but instead we still decrement here:
-                            in_transaction.transaction_depth = NonZeroU32::new(depth_gt1 - 1)
-                                .expect("Depth was checked to be > 1");
-                            return Ok(());
-                        } else {
-                            Cow::Owned(format!(
-                                "ROLLBACK TO SAVEPOINT diesel_savepoint_{}",
-                                depth_gt1 - 1
-                            ))
-                        }
-                    }
-                }
-            }
+                    1 => (Cow::Borrowed("ROLLBACK"), true),
+                    depth_gt1 => (
+                        Cow::Owned(format!(
+                            "ROLLBACK TO SAVEPOINT diesel_savepoint_{}",
+                            depth_gt1 - 1
+                        )),
+                        false,
+                    ),
+                },
+                in_transaction.requires_rollback_maybe_up_to_top_level,
+            ),
             None => return Err(Error::NotInTransaction),
         };
 
         match conn.batch_execute(&rollback_sql) {
             Ok(()) => {
-                Self::get_transaction_state(conn)?
-                    .change_transaction_depth(TransactionDepthChange::DecreaseDepth)?;
+                match Self::get_transaction_state(conn)?
+                    .change_transaction_depth(TransactionDepthChange::DecreaseDepth)
+                {
+                    Ok(()) => {}
+                    Err(Error::NotInTransaction) if rolling_back_top_level => {
+                        // Transaction exit may have already been detected by connection
+                        // implementation. It's fine.
+                    }
+                    Err(e) => return Err(e),
+                }
                 Ok(())
             }
             Err(rollback_error) => {
@@ -375,17 +360,33 @@ where
                         in_transaction:
                             Some(InTransactionStatus {
                                 transaction_depth,
-                                top_level_transaction_requires_rollback,
+                                requires_rollback_maybe_up_to_top_level,
                                 ..
                             }),
-                    }) if transaction_depth.get() > 1
-                        && !*top_level_transaction_requires_rollback =>
-                    {
+                    }) if transaction_depth.get() > 1 => {
                         // A savepoint failed to rollback - we may still attempt to repair
-                        // the connection by rolling back top-level transaction.
+                        // the connection by rolling back higher levels.
+
+                        // To make it easier on the user (that they don't have to really
+                        // look at actual transaction depth and can just rely on the number
+                        // of times they have called begin/commit/rollback) we still
+                        // decrement here:
                         *transaction_depth = NonZeroU32::new(transaction_depth.get() - 1)
                             .expect("Depth was checked to be > 1");
-                        *top_level_transaction_requires_rollback = true;
+                        *requires_rollback_maybe_up_to_top_level = true;
+                        if requires_rollback_maybe_up_to_top_level_before_execute {
+                            // In that case, we tolerate that savepoint releases fail
+                            // -> we should ignore errors
+                            return Ok(());
+                        }
+                    }
+                    TransactionManagerStatus::Valid(ValidTransactionManagerStatus {
+                        in_transaction: None,
+                    }) => {
+                        // we would have returned `NotInTransaction` if that was already the state
+                        // before we made our call
+                        // => Transaction manager status has been fixed by the underlying connection
+                        // so we don't need to set_in_error
                     }
                     _ => tm_status.set_in_error(),
                 }
@@ -402,53 +403,50 @@ where
     fn commit_transaction(conn: &mut Conn) -> QueryResult<()> {
         let transaction_state = Self::get_transaction_state(conn)?;
         let transaction_depth = transaction_state.transaction_depth();
-        let commit_sql = match transaction_depth {
+        let (commit_sql, committing_top_level) = match transaction_depth {
             None => return Err(Error::NotInTransaction),
-            Some(transaction_depth) if transaction_depth.get() == 1 => Cow::Borrowed("COMMIT"),
-            Some(transaction_depth) => Cow::Owned(format!(
-                "RELEASE SAVEPOINT diesel_savepoint_{}",
-                transaction_depth.get() - 1
-            )),
+            Some(transaction_depth) if transaction_depth.get() == 1 => {
+                (Cow::Borrowed("COMMIT"), true)
+            }
+            Some(transaction_depth) => (
+                Cow::Owned(format!(
+                    "RELEASE SAVEPOINT diesel_savepoint_{}",
+                    transaction_depth.get() - 1
+                )),
+                false,
+            ),
         };
         match conn.batch_execute(&commit_sql) {
             Ok(()) => {
-                Self::get_transaction_state(conn)?
-                    .change_transaction_depth(TransactionDepthChange::DecreaseDepth)?;
+                match Self::get_transaction_state(conn)?
+                    .change_transaction_depth(TransactionDepthChange::DecreaseDepth)
+                {
+                    Ok(()) => {}
+                    Err(Error::NotInTransaction) if committing_top_level => {
+                        // Transaction exit may have already been detected by connection.
+                        // It's fine
+                    }
+                    Err(e) => return Err(e),
+                }
                 Ok(())
             }
             Err(commit_error) => {
                 if let TransactionManagerStatus::Valid(ValidTransactionManagerStatus {
                     in_transaction:
                         Some(InTransactionStatus {
-                            ref mut transaction_depth,
-                            top_level_transaction_requires_rollback: true,
+                            requires_rollback_maybe_up_to_top_level: true,
                             ..
                         }),
                 }) = conn.transaction_state().status
                 {
-                    match transaction_depth.get() {
-                        1 => match Self::rollback_transaction(conn) {
-                            Ok(()) => {}
-                            Err(rollback_error) => {
-                                conn.transaction_state().status.set_in_error();
-                                return Err(Error::RollbackErrorOnCommit {
-                                    rollback_error: Box::new(rollback_error),
-                                    commit_error: Box::new(commit_error),
-                                });
-                            }
-                        },
-                        depth_gt1 => {
-                            // There's no point in *actually* rolling back this one
-                            // because we won't be able to do anything until top-level
-                            // is rolled back.
-
-                            // To make it easier on the user (that they don't have to really look
-                            // at actual transaction depth and can just rely on the number of
-                            // times they have called begin/commit/rollback) we don't mark the
-                            // transaction manager as out of the savepoints as soon as we
-                            // realize there is that issue, but instead we still decrement here:
-                            *transaction_depth = NonZeroU32::new(depth_gt1 - 1)
-                                .expect("Depth was checked to be > 1");
+                    match Self::rollback_transaction(conn) {
+                        Ok(()) => {}
+                        Err(rollback_error) => {
+                            conn.transaction_state().status.set_in_error();
+                            return Err(Error::RollbackErrorOnCommit {
+                                rollback_error: Box::new(rollback_error),
+                                commit_error: Box::new(commit_error),
+                            });
                         }
                     }
                 }
@@ -490,7 +488,7 @@ mod test {
                 if self.top_level_requires_rollback_after_next_batch_execute {
                     self.transaction_state
                         .status
-                        .set_top_level_transaction_requires_rollback();
+                        .set_requires_rollback_maybe_up_to_top_level(true);
                 }
                 res
             }
@@ -995,4 +993,181 @@ mod test {
             ).transaction_depth().expect("Transaction depth")
         );
     }
+
+    // regression test for #3470
+    // crates.io depends on this behaviour
+    #[test]
+    #[cfg(feature = "postgres")]
+    fn some_libpq_failures_are_recoverable_by_rolling_back_the_savepoint_only() {
+        use crate::connection::{AnsiTransactionManager, TransactionManager};
+        use crate::prelude::*;
+        use crate::sql_query;
+
+        crate::table! {
+            rollback_test (id) {
+                id -> Int4,
+                value -> Int4,
+            }
+        }
+
+        let conn = &mut crate::test_helpers::pg_connection_no_transaction();
+        assert_eq!(
+            None,
+            <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
+                conn
+            ).transaction_depth().expect("Transaction depth")
+        );
+
+        let res = conn.transaction(|conn| {
+            sql_query(
+                "CREATE TABLE IF NOT EXISTS rollback_test (id INT PRIMARY KEY, value INT NOT NULL)",
+            )
+            .execute(conn)?;
+            conn.transaction(|conn| {
+                sql_query("SET TRANSACTION READ ONLY").execute(conn)?;
+                crate::update(rollback_test::table)
+                    .set(rollback_test::value.eq(0))
+                    .execute(conn)
+            })
+            .map(|_| {
+                panic!("Should use the `or_else` branch");
+            })
+            .or_else(|_| sql_query("SELECT 1").execute(conn))
+            .map(|_| ())
+        });
+        assert!(res.is_ok());
+
+        assert_eq!(
+            None,
+            <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
+                conn
+            ).transaction_depth().expect("Transaction depth")
+        );
+    }
+
+    #[test]
+    #[cfg(feature = "postgres")]
+    fn other_libpq_failures_are_not_recoverable_by_rolling_back_the_savepoint_only() {
+        use crate::connection::{AnsiTransactionManager, TransactionManager};
+        use crate::prelude::*;
+        use crate::sql_query;
+        use std::num::NonZeroU32;
+        use std::sync::{Arc, Barrier};
+
+        crate::table! {
+            rollback_test2 (id) {
+                id -> Int4,
+                value -> Int4,
+            }
+        }
+        let conn = &mut crate::test_helpers::pg_connection_no_transaction();
+
+        sql_query(
+            "CREATE TABLE IF NOT EXISTS rollback_test2 (id INT PRIMARY KEY, value INT NOT NULL)",
+        )
+        .execute(conn)
+        .unwrap();
+
+        let start_barrier = Arc::new(Barrier::new(2));
+        let commit_barrier = Arc::new(Barrier::new(2));
+
+        let other_start_barrier = start_barrier.clone();
+        let other_commit_barrier = commit_barrier.clone();
+
+        let t1 = std::thread::spawn(move || {
+            let conn = &mut crate::test_helpers::pg_connection_no_transaction();
+            assert_eq!(
+                None,
+                <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
+                    conn
+                ).transaction_depth().expect("Transaction depth")
+            );
+            let r = conn.build_transaction().serializable().run::<_, crate::result::Error, _>(|conn| {
+                assert_eq!(
+                    NonZeroU32::new(1),
+                    <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
+                        conn
+                    ).transaction_depth().expect("Transaction depth")
+                );
+                rollback_test2::table.load::<(i32, i32)>(conn)?;
+                crate::insert_into(rollback_test2::table)
+                    .values((rollback_test2::id.eq(1), rollback_test2::value.eq(42)))
+                    .execute(conn)?;
+                let r = conn.transaction(|conn| {
+                    assert_eq!(
+                        NonZeroU32::new(2),
+                        <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
+                            conn
+                        ).transaction_depth().expect("Transaction depth")
+                    );
+                    start_barrier.wait();
+                    commit_barrier.wait();
+                    let r = rollback_test2::table.load::<(i32, i32)>(conn);
+                    assert!(r.is_err());
+                    Err::<(), _>(crate::result::Error::RollbackTransaction)
+                });
+                assert_eq!(
+                    NonZeroU32::new(1),
+                    <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
+                        conn
+                    ).transaction_depth().expect("Transaction depth")
+                );
+                assert!(
+                    matches!(r, Err(crate::result::Error::RollbackTransaction)),
+                    "rollback failed (such errors should be ignored by transaction manager): {}",
+                    r.unwrap_err()
+                );
+                let r = rollback_test2::table.load::<(i32, i32)>(conn);
+                assert!(r.is_err());
+                // fun fact: if hitting "commit" after receiving a serialization failure, PG
+                // returns that the commit has succeeded, but in fact it was actually rolled back.
+                // soo.. one should avoid doing that
+                r
+            });
+            assert!(r.is_err());
+            assert_eq!(
+                None,
+                <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
+                    conn
+                ).transaction_depth().expect("Transaction depth")
+            );
+        });
+
+        let t2 = std::thread::spawn(move || {
+            other_start_barrier.wait();
+            let conn = &mut crate::test_helpers::pg_connection_no_transaction();
+            assert_eq!(
+                None,
+                <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
+                    conn
+                ).transaction_depth().expect("Transaction depth")
+            );
+            let r = conn.build_transaction().serializable().run::<_, crate::result::Error, _>(|conn| {
+                assert_eq!(
+                    NonZeroU32::new(1),
+                    <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
+                        conn
+                    ).transaction_depth().expect("Transaction depth")
+                );
+                let _ = rollback_test2::table.load::<(i32, i32)>(conn)?;
+                crate::insert_into(rollback_test2::table)
+                    .values((rollback_test2::id.eq(23), rollback_test2::value.eq(42)))
+                    .execute(conn)?;
+                Ok(())
+            });
+            other_commit_barrier.wait();
+            assert!(r.is_ok(), "{:?}", r.unwrap_err());
+            assert_eq!(
+                None,
+                <AnsiTransactionManager as TransactionManager<PgConnection>>::transaction_manager_status_mut(
+                    conn
+                ).transaction_depth().expect("Transaction depth")
+            );
+        });
+        crate::sql_query("DELETE FROM rollback_test2")
+            .execute(conn)
+            .unwrap();
+        t1.join().unwrap();
+        t2.join().unwrap();
+    }
 }
diff --git a/diesel/src/mysql/connection/mod.rs b/diesel/src/mysql/connection/mod.rs
index b7f4522981ae..09a14459d0ac 100644
--- a/diesel/src/mysql/connection/mod.rs
+++ b/diesel/src/mysql/connection/mod.rs
@@ -171,7 +171,7 @@ fn update_transaction_manager_status<T>(
     if let Err(Error::DatabaseError(DatabaseErrorKind::SerializationFailure, _)) = query_result {
         transaction_manager
             .status
-            .set_top_level_transaction_requires_rollback()
+            .set_requires_rollback_maybe_up_to_top_level(true)
     }
     query_result
 }
diff --git a/diesel/src/pg/connection/mod.rs b/diesel/src/pg/connection/mod.rs
index 8a3d6fd29e36..b1474070d928 100644
--- a/diesel/src/pg/connection/mod.rs
+++ b/diesel/src/pg/connection/mod.rs
@@ -230,37 +230,54 @@ fn update_transaction_manager_status<T>(
     query_result: QueryResult<T>,
     conn: &mut ConnectionAndTransactionManager,
 ) -> QueryResult<T> {
-    if let Err(Error::DatabaseError { .. }) = query_result {
-        /// avoid monomorphizing for every result type - this part will not be inlined
-        fn non_generic_inner(conn: &mut ConnectionAndTransactionManager) {
-            let raw_conn: &mut RawConnection = &mut conn.raw_connection;
-            let tm: &mut AnsiTransactionManager = &mut conn.transaction_state;
-            if tm.status.is_not_broken_and_in_transaction() {
-                // libpq keeps track of the transaction status internally, and that is accessible
-                // via `transaction_status`. We can use that to update the AnsiTransactionManager
-                // status
-                match raw_conn.transaction_status() {
-                    PgTransactionStatus::InError => {
-                        tm.status.set_top_level_transaction_requires_rollback()
-                    }
-                    PgTransactionStatus::Unknown => tm.status.set_in_error(),
-                    PgTransactionStatus::Idle => {
-                        // This may repair the transaction manager
-                        tm.status = TransactionManagerStatus::Valid(Default::default())
-                    }
-                    PgTransactionStatus::InTransaction => {
-                        let transaction_status = &mut tm.status;
-                        if !matches!(transaction_status, TransactionManagerStatus::Valid(valid_tm) if valid_tm.transaction_depth().is_some())
-                        {
-                            transaction_status.set_in_error()
-                        }
+    /// avoid monomorphizing for every result type - this part will not be inlined
+    fn non_generic_inner(conn: &mut ConnectionAndTransactionManager, is_err: bool) {
+        let raw_conn: &mut RawConnection = &mut conn.raw_connection;
+        let tm: &mut AnsiTransactionManager = &mut conn.transaction_state;
+        // libpq keeps track of the transaction status internally, and that is accessible
+        // via `transaction_status`. We can use that to update the AnsiTransactionManager
+        // status
+        match raw_conn.transaction_status() {
+            PgTransactionStatus::InError => {
+                tm.status.set_requires_rollback_maybe_up_to_top_level(true)
+            }
+            PgTransactionStatus::Unknown => tm.status.set_in_error(),
+            PgTransactionStatus::Idle => {
+                // This is useful in particular for commit attempts (even
+                // if `COMMIT` errors it still exits transaction)
+
+                // This may repair the transaction manager
+                tm.status = TransactionManagerStatus::Valid(Default::default())
+            }
+            PgTransactionStatus::InTransaction => {
+                let transaction_status = &mut tm.status;
+                // If we weren't an error, it is possible that we were a transaction start
+                // -> we should tolerate any state
+                if is_err {
+                    // An error may not have us enter a transaction, so if we weren't in one
+                    // we may not be in one now
+                    if !matches!(transaction_status, TransactionManagerStatus::Valid(valid_tm) if valid_tm.transaction_depth().is_some())
+                    {
+                        // -> transaction manager is broken
+                        transaction_status.set_in_error()
                     }
-                    PgTransactionStatus::Active => {}
+                } else {
+                    // If transaction was InError before, but now it's not (because we attempted
+                    // a rollback), we may pretend it's fixed because
+                    // if it isn't Postgres *will* tell us again.
+
+                    // Fun fact: if we have not received an `InTransaction` status however,
+                    // postgres will *not* warn us that transaction is broken when attempting to
+                    // commit, so we may think that commit has succeeded but in fact it hasn't.
+                    tm.status.set_requires_rollback_maybe_up_to_top_level(false)
                 }
             }
+            PgTransactionStatus::Active => {
+                // This is a transient state for libpq - nothing we can deduce here.
+            }
         }
-        non_generic_inner(conn)
     }
+    non_generic_inner(conn, query_result.is_err());
     query_result
 }
 
diff --git a/diesel/src/pg/connection/raw.rs b/diesel/src/pg/connection/raw.rs
index a7483cfa9bac..3696230daa4e 100644
--- a/diesel/src/pg/connection/raw.rs
+++ b/diesel/src/pg/connection/raw.rs
@@ -108,6 +108,8 @@ impl RawConnection {
         RawResult::new(ptr, self)
     }
 
+    /// This is reasonably inexpensive as it just accesses variables internal to the connection
+    /// that are kept up to date by the `ReadyForQuery` messages from the PG server
     pub(super) fn transaction_status(&self) -> PgTransactionStatus {
         unsafe { PQtransactionStatus(self.internal_connection.as_ptr()) }.into()
     }
diff --git a/diesel/src/result.rs b/diesel/src/result.rs
index ce8b1ad00f9b..2a9ff4e0979f 100644
--- a/diesel/src/result.rs
+++ b/diesel/src/result.rs
@@ -73,8 +73,7 @@ pub enum Error {
     RollbackErrorOnCommit {
         /// The error that was encountered when attempting the rollback
         rollback_error: Box<Error>,
-        /// If the rollback attempt resulted from a failed attempt to commit the transaction,
-        /// you will find the related error here.
+        /// The error that was encountered during the failed commit attempt
         commit_error: Box<Error>,
     },
 


Test patch: