fix postgresql: rollback aborted transaction before pipeline cleanup

commit_hash:eb11ea35fecd2a9dabd32bf95299754c97376160

Author: oigolovanova

postgresql/src/storages/postgres/detail/connection_impl.cpp

@@ -763,6 +763,9 @@ bool ConnectionImpl::Cleanup(TimeoutDuration timeout) {
     if (state == ConnectionState::kTranActive) {
         return false;
     }
+    if (state > ConnectionState::kIdle) {
+        Rollback(deadline);
+    }
     // We might need more timeout here
     // We are no more bound with SLA, user has his exception.
     // We need to try and save the connection without canceling current query
@@ -781,9 +784,6 @@ bool ConnectionImpl::Cleanup(TimeoutDuration timeout) {
         // Reenter pipeline mode if necessary
         conn_wrapper_.EnterPipelineMode();
     }
-    if (state > ConnectionState::kIdle) {
-        Rollback(deadline);
-    }
     return GetConnectionState() == ConnectionState::kIdle;
 }
 

postgresql/src/storages/postgres/tests/connection_transaction_mode_pgtest.cpp

@@ -24,6 +24,11 @@ constexpr pg::CommandControl kTransactionPoolerDefaultCmdCtl{
     std::chrono::seconds{10},
     std::chrono::seconds{10},
 };
+constexpr pg::CommandControl kTransactionPoolerNoPrepareCmdCtl{
+    kTransactionPoolerNetworkTimeout,
+    kTransactionPoolerStatementTimeout,
+    pg::CommandControl::PreparedStatementsOptionOverride::kDisabled,
+};
 
 pg::ConnectionSettings MakeTransactionPoolerSettings(const pg::ConnectionSettings& base = kCachePreparedStatements) {
     auto settings = base;
@@ -214,6 +219,39 @@ UTEST_F(PostgreTransactionModeConnection, TransactionPoolerResendsSetConfigWhenC
     EXPECT_TRUE(conn->IsIdle());
 }
 
+UTEST_F(PostgreTransactionModeConnection, DuplicatePreparedStatementDirtyConnectionIsCleanedUp) {
+    const pg::Query query{"SELECT 42"};
+
+    std::string statement_name;
+    {
+        const auto conn = MakeConn();
+        UEXPECT_NO_THROW(
+            statement_name = conn->PrepareStatement(query, {}, utest::kMaxTestWaitTime).meta_statement_name
+        );
+    }
+    ASSERT_FALSE(statement_name.empty());
+
+    const DefaultCommandControlScope begin_scope{kTransactionPoolerCmdCtl};
+
+    const auto conn = MakeConn();
+
+    UEXPECT_NO_THROW(conn->Execute("PREPARE " + statement_name + " AS SELECT 42", {}, kTransactionPoolerNoPrepareCmdCtl)
+    );
+
+    UEXPECT_NO_THROW(conn->Begin({}, {}));
+    UEXPECT_THROW(conn->Execute(query), pg::DuplicatePreparedStatement);
+    ASSERT_EQ(pg::ConnectionState::kTranError, conn->GetState());
+
+    const DefaultCommandControlScope cleanup_scope{kTransactionPoolerDefaultCmdCtl};
+
+    UEXPECT_NO_THROW(conn->Cleanup(utest::kMaxTestWaitTime));
+    EXPECT_EQ(pg::ConnectionState::kIdle, conn->GetState());
+    EXPECT_FALSE(conn->IsBroken());
+
+    UEXPECT_NO_THROW(conn->Execute("SELECT 1"));
+    EXPECT_FALSE(conn->IsBroken());
+}
+
 }  // namespace
 
 USERVER_NAMESPACE_END