From 4ce7bb1d99a9f906dd330012e88515085d0eb57c Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Thu, 10 Sep 2026 16:09:39 +0100
Subject: [PATCH 01/11] FIX: prevent pooled connections retaining transactions

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 CHANGELOG.md                                  |  3 +
 mssql_python/pybind/connection/connection.cpp | 50 +++++++++++++++-
 mssql_python/pybind/connection/connection.h   |  3 +-
 .../pybind/connection/connection_pool.cpp     | 35 +++++++++++
 .../pybind/connection/connection_pool.h       |  6 ++
 tests/test_009_pooling.py                     | 58 +++++++++++++++++++
 6 files changed, 151 insertions(+), 4 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index f4ebfb816..ebcf909ae 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -69,6 +69,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
   before; users should call `cursor.setinputsizes()` to work around this.
 
 ### Fixed
+- **GH-754:** Pooled connections are now rolled back and restored to autocommit
+  mode before being parked. This prevents an empty transaction from remaining
+  visible on an idle SQL Server session after `Connection.close()`.
 - **GH-740:** A Python `Decimal` whose value falls in the SQL Server MONEY /
   SMALLMONEY range is now bound as `SQL_NUMERIC` with its own precision and scale
   on both `execute()` paths (native detection, and the legacy path reached when
diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp
index 8ccee11a7..822ef3abf 100644
--- a/mssql_python/pybind/connection/connection.cpp
+++ b/mssql_python/pybind/connection/connection.cpp
@@ -52,8 +52,14 @@ Connection::Connection(const std::u16string& conn_str, bool use_pool)
     allocateDbcHandle();
 }
 
-Connection::~Connection() {
-    disconnect();  // fallback if user forgets to disconnect
+Connection::~Connection() noexcept {
+    try {
+        disconnect();  // fallback if user forgets to disconnect
+    } catch (...) {
+        // Destructors must not propagate ODBC disconnect failures. Releasing
+        // the handle still lets SqlHandle perform SQLFreeHandle cleanup.
+        _dbcHandle.reset();
+    }
 }
 
 // Allocates connection handle
@@ -564,6 +570,21 @@ bool Connection::reset() {
     return true;
 }
 
+void Connection::prepareForPool() {
+    if (!_dbcHandle) {
+        ThrowStdException("Connection handle not allocated");
+    }
+
+    if (!getAutocommit()) {
+        // End any caller transaction before check-in, then park the physical
+        // connection in autocommit mode. The SQL Server ODBC driver can leave
+        // an empty transaction visible after SQLEndTran while manual-commit
+        // mode remains enabled; switching modes ends that transaction.
+        rollback();
+        setAutocommit(true);
+    }
+}
+
 void Connection::updateLastUsed() {
     _lastUsed = std::chrono::steady_clock::now();
 }
@@ -659,7 +680,17 @@ ConnectionHandle::ConnectionHandle(const std::u16string& connStr, bool usePool,
 
 ConnectionHandle::~ConnectionHandle() {
     if (_conn) {
-        close();
+        try {
+            close();
+        } catch (...) {
+            if (_conn) {
+                try {
+                    _conn->disconnect();
+                } catch (...) {
+                }
+            }
+            _conn = nullptr;
+        }
     }
 }
 
@@ -669,6 +700,19 @@ void ConnectionHandle::close() {
         ThrowStdException("Connection object is not initialized");
     }
     if (_usePool) {
+        try {
+            _conn->prepareForPool();
+        } catch (...) {
+            // Never retain a connection whose transaction state could not be
+            // sanitized. Discarding also releases this connection's reserved
+            // pool capacity. Preserve the original check-in error.
+            try {
+                ConnectionPoolManager::getInstance().discardConnection(_poolKey, _conn);
+            } catch (...) {
+            }
+            _conn = nullptr;
+            throw;
+        }
         ConnectionPoolManager::getInstance().returnConnection(_poolKey, _conn);
     } else {
         _conn->disconnect();
diff --git a/mssql_python/pybind/connection/connection.h b/mssql_python/pybind/connection/connection.h
index d5300aff4..fbd1a88b5 100644
--- a/mssql_python/pybind/connection/connection.h
+++ b/mssql_python/pybind/connection/connection.h
@@ -32,7 +32,7 @@ class Connection {
   public:
     Connection(const std::u16string& connStr, bool fromPool);
 
-    ~Connection();
+    ~Connection() noexcept;
 
     // Establish the connection using the stored connection string.
     void connect(const py::dict& attrs_before = py::dict());
@@ -53,6 +53,7 @@ class Connection {
     bool getAutocommit() const;
     bool isAlive() const;
     bool reset();
+    void prepareForPool();
     void updateLastUsed();
     std::chrono::steady_clock::time_point lastUsed() const;
 
diff --git a/mssql_python/pybind/connection/connection_pool.cpp b/mssql_python/pybind/connection/connection_pool.cpp
index 831a01db2..004af708f 100644
--- a/mssql_python/pybind/connection/connection_pool.cpp
+++ b/mssql_python/pybind/connection/connection_pool.cpp
@@ -329,6 +329,20 @@ void ConnectionPool::release(std::shared_ptr<Connection> conn) {
     }
 }
 
+void ConnectionPool::discard(std::shared_ptr<Connection> conn) {
+    {
+        std::lock_guard<std::mutex> lock(_mutex);
+        if (_current_size > 0)
+            --_current_size;
+    }
+    try {
+        conn->disconnect();
+    } catch (...) {
+        // The caller is already handling a sanitation failure. Connection's
+        // noexcept destructor still releases the ODBC handle.
+    }
+}
+
 bool ConnectionPool::canEvict() {
     std::lock_guard<std::mutex> lock(_mutex);
     // Never evict while any connection is checked out or in-flight. Reserved
@@ -506,6 +520,27 @@ void ConnectionPoolManager::returnConnection(const std::u16string& pool_key,
     }
 }
 
+void ConnectionPoolManager::discardConnection(const std::u16string& pool_key,
+                                               const std::shared_ptr<Connection> conn) {
+    std::shared_ptr<ConnectionPool> pool;
+    {
+        std::lock_guard<std::mutex> lock(_manager_mutex);
+        auto it = _pools.find(pool_key);
+        if (it != _pools.end()) {
+            pool = it->second;
+        }
+    }
+    if (pool) {
+        pool->discard(conn);
+    } else if (conn) {
+        try {
+            conn->disconnect();
+        } catch (...) {
+            // Connection's noexcept destructor still releases the ODBC handle.
+        }
+    }
+}
+
 void ConnectionPoolManager::configure(int max_size, int idle_timeout_secs) {
     std::lock_guard<std::mutex> lock(_manager_mutex);
     _default_max_size = max_size;
diff --git a/mssql_python/pybind/connection/connection_pool.h b/mssql_python/pybind/connection/connection_pool.h
index edc87c865..40caf801a 100644
--- a/mssql_python/pybind/connection/connection_pool.h
+++ b/mssql_python/pybind/connection/connection_pool.h
@@ -34,6 +34,9 @@ class ConnectionPool {
     // Returns a connection to the pool for reuse
     void release(std::shared_ptr<Connection> conn);
 
+    // Permanently removes a checked-out connection and releases its capacity.
+    void discard(std::shared_ptr<Connection> conn);
+
     // Closes all connections in the pool, releasing resources
     void close();
 
@@ -82,6 +85,9 @@ class ConnectionPoolManager {
     // (the same key passed to acquireConnection).
     void returnConnection(const std::u16string& pool_key, std::shared_ptr<Connection> conn);
 
+    // Discards a connection that cannot safely be returned to its original pool.
+    void discardConnection(const std::u16string& pool_key, std::shared_ptr<Connection> conn);
+
     // Closes all pools and their connections
     void closePools();
 
diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py
index e33f71030..79bdb46d1 100644
--- a/tests/test_009_pooling.py
+++ b/tests/test_009_pooling.py
@@ -127,6 +127,64 @@ def test_connection_pooling_reuse_spid(conn_str):
     assert spid1 == spid2, "Connections not reused - different SPIDs"
 
 
+def test_pooled_close_leaves_no_open_transaction(conn_str):
+    """A physical connection must not retain a transaction while parked."""
+    _run_in_subprocess(
+        """
+        import os
+
+        import mssql_python
+
+        conn_str = os.environ["DB_CONNECTION_STRING"]
+        mssql_python.pooling(enabled=True, max_size=2, idle_timeout=30)
+        subject = mssql_python.connect(conn_str)
+        observer = mssql_python.connect(conn_str, autocommit=True)
+        try:
+            cursor = subject.cursor()
+            cursor.execute("SELECT @@SPID")
+            subject_spid = cursor.fetchone()[0]
+            observer_cursor = observer.cursor()
+            observer_cursor.execute(
+                "SELECT open_transaction_count "
+                "FROM sys.dm_exec_sessions WHERE session_id = ?",
+                [subject_spid],
+            )
+            if observer_cursor.fetchone() is None:
+                import sys
+
+                print(
+                    "Test login cannot inspect another SQL Server session",
+                    file=sys.stderr,
+                )
+                sys.exit(77)
+
+            cursor.execute("SELECT 1")
+            cursor.fetchone()
+            subject.commit()
+            cursor.close()
+            subject.close()
+
+            observer_cursor.execute(
+                "SELECT open_transaction_count "
+                "FROM sys.dm_exec_sessions WHERE session_id = ?",
+                [subject_spid],
+            )
+            row = observer_cursor.fetchone()
+            assert row is not None, "The parked SQL Server session was not visible"
+            assert row[0] == 0, (
+                "Pooled connection retained an open transaction after close: "
+                f"SPID {subject_spid}, open_transaction_count={row[0]}"
+            )
+            observer_cursor.close()
+        finally:
+            subject.close()
+            observer.close()
+            mssql_python.pooling(enabled=False)
+        """,
+        conn_str,
+    )
+
+
 def test_connection_pooling_isolation_level_reset(conn_str):
     """Test that pooling correctly resets session state for isolation level.
 

From bf9c828e7deeec1e7df343a19ac294c240421fde Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Thu, 10 Sep 2026 16:34:11 +0100
Subject: [PATCH 02/11] TEST: expand pooled transaction cleanup coverage

Cover commit, rollback, autocommit reuse, and failed sanitation capacity recovery.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 mssql_python/connection.py |  22 +++---
 tests/test_009_pooling.py  | 156 ++++++++++++++++++++++++++++---------
 2 files changed, 129 insertions(+), 49 deletions(-)

diff --git a/mssql_python/connection.py b/mssql_python/connection.py
index 10aec7103..f63b923fc 100644
--- a/mssql_python/connection.py
+++ b/mssql_python/connection.py
@@ -2158,21 +2158,21 @@ def close(self) -> None:
         # Close the connection even if cursor cleanup had issues
         try:
             if self._conn:
-                if not self.autocommit:
+                if not self.autocommit and not self._pooling:
                     # If autocommit is disabled, rollback any uncommitted changes
-                    # This is important to ensure no partial transactions remain
-                    # For autocommit True, this is not necessary as each statement is
-                    # committed immediately
+                    # before disconnecting a non-pooled connection. Pooled
+                    # connections are rolled back by the native check-in path,
+                    # which can discard the connection and release its capacity
+                    # atomically if sanitation fails.
                     logger.debug("Rolling back uncommitted changes before closing connection.")
-                    try:
-                        self._conn.rollback()
-                    except RuntimeError as e:
-                        # Handle C++ layer RuntimeError with proper DB-API exception mapping
-                        _raise_connection_error(e)
                 # TODO: Check potential race conditions in case of multithreaded scenarios
                 # Close the connection
-                self._conn.close()
-                self._conn = None
+                try:
+                    self._conn.close()
+                except RuntimeError as e:
+                    _raise_connection_error(e)
+                finally:
+                    self._conn = None
         except Exception as e:
             logger.error(f"Error closing database connection: {e}")
             # Re-raise the connection close error as it's more critical
diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py
index 79bdb46d1..b55db146d 100644
--- a/tests/test_009_pooling.py
+++ b/tests/test_009_pooling.py
@@ -127,8 +127,8 @@ def test_connection_pooling_reuse_spid(conn_str):
     assert spid1 == spid2, "Connections not reused - different SPIDs"
 
 
-def test_pooled_close_leaves_no_open_transaction(conn_str):
-    """A physical connection must not retain a transaction while parked."""
+def test_pooled_close_paths_leave_no_open_transaction(conn_str):
+    """Every close path must leave the physical connection transaction-clean."""
     _run_in_subprocess(
         """
         import os
@@ -137,47 +137,73 @@ def test_pooled_close_leaves_no_open_transaction(conn_str):
 
         conn_str = os.environ["DB_CONNECTION_STRING"]
         mssql_python.pooling(enabled=True, max_size=2, idle_timeout=30)
-        subject = mssql_python.connect(conn_str)
         observer = mssql_python.connect(conn_str, autocommit=True)
         try:
-            cursor = subject.cursor()
-            cursor.execute("SELECT @@SPID")
-            subject_spid = cursor.fetchone()[0]
             observer_cursor = observer.cursor()
-            observer_cursor.execute(
-                "SELECT open_transaction_count "
-                "FROM sys.dm_exec_sessions WHERE session_id = ?",
-                [subject_spid],
-            )
-            if observer_cursor.fetchone() is None:
-                import sys
 
-                print(
-                    "Test login cannot inspect another SQL Server session",
-                    file=sys.stderr,
+            scenarios = (
+                ("direct commit", False, "SELECT 1", None, "commit"),
+                ("prepared commit", False, "SELECT CAST(? AS INT)", [1], "commit"),
+                ("explicit rollback", False, "SELECT 1", None, "rollback"),
+                ("implicit close rollback", False, "SELECT 1", None, None),
+                ("autocommit close", True, "SELECT 1", None, None),
+            )
+            expected_spid = None
+            for name, autocommit, sql, params, action in scenarios:
+                subject = mssql_python.connect(conn_str, autocommit=autocommit)
+                try:
+                    assert subject.autocommit is autocommit
+                    cursor = subject.cursor()
+                    cursor.execute("SELECT @@SPID")
+                    subject_spid = cursor.fetchone()[0]
+                    if expected_spid is None:
+                        expected_spid = subject_spid
+                    else:
+                        assert subject_spid == expected_spid, (
+                            f"{name}: expected pooled SPID {expected_spid}, got {subject_spid}"
+                        )
+
+                    observer_cursor.execute(
+                        "SELECT open_transaction_count "
+                        "FROM sys.dm_exec_sessions WHERE session_id = ?",
+                        [subject_spid],
+                    )
+                    if observer_cursor.fetchone() is None:
+                        import sys
+
+                        print(
+                            "Test login cannot inspect another SQL Server session",
+                            file=sys.stderr,
+                        )
+                        sys.exit(77)
+
+                    if params is None:
+                        cursor.execute(sql)
+                    else:
+                        cursor.execute(sql, params)
+                    cursor.fetchone()
+                    if action == "commit":
+                        subject.commit()
+                    elif action == "rollback":
+                        subject.rollback()
+                    cursor.close()
+                finally:
+                    subject.close()
+
+                observer_cursor.execute(
+                    "SELECT open_transaction_count "
+                    "FROM sys.dm_exec_sessions WHERE session_id = ?",
+                    [subject_spid],
+                )
+                row = observer_cursor.fetchone()
+                assert row is not None, f"{name}: parked SQL Server session was not visible"
+                assert row[0] == 0, (
+                    f"{name}: pooled SPID {subject_spid} retained "
+                    f"open_transaction_count={row[0]}"
                 )
-                sys.exit(77)
-
-            cursor.execute("SELECT 1")
-            cursor.fetchone()
-            subject.commit()
-            cursor.close()
-            subject.close()
 
-            observer_cursor.execute(
-                "SELECT open_transaction_count "
-                "FROM sys.dm_exec_sessions WHERE session_id = ?",
-                [subject_spid],
-            )
-            row = observer_cursor.fetchone()
-            assert row is not None, "The parked SQL Server session was not visible"
-            assert row[0] == 0, (
-                "Pooled connection retained an open transaction after close: "
-                f"SPID {subject_spid}, open_transaction_count={row[0]}"
-            )
             observer_cursor.close()
         finally:
-            subject.close()
             observer.close()
             mssql_python.pooling(enabled=False)
         """,
@@ -766,9 +792,9 @@ def session_identity(conn):
             spid, login_time = cur.fetchone()
             return (spid, login_time)
 
-        # Step 1: two distinct, autocommit connections. Autocommit avoids
-        # the implicit rollback in Connection.close(), which would
-        # otherwise fail on the killed session and leak its pool slot.
+        # Step 1: two distinct, autocommit connections. Autocommit keeps this
+        # test focused on detecting dead connections during checkout; failed
+        # manual-commit sanitation is covered separately below.
         victim = connect(conn_str)
         admin = connect(conn_str)
         victim.autocommit = True
@@ -836,6 +862,60 @@ def session_identity(conn):
     )
 
 
+def test_failed_pool_sanitation_releases_capacity(conn_str):
+    """A connection discarded after failed sanitation must not consume a pool slot."""
+    _run_in_subprocess(
+        """
+        import os
+        import sys
+
+        from mssql_python import connect, pooling
+
+        conn_str = os.environ["DB_CONNECTION_STRING"]
+        pooling(max_size=2, idle_timeout=30)
+        victim = connect(conn_str)
+        admin = connect(conn_str, autocommit=True)
+
+        victim_cursor = victim.cursor()
+        victim_cursor.execute("SELECT @@SPID")
+        victim_spid = victim_cursor.fetchone()[0]
+        victim_cursor.close()
+
+        try:
+            admin.cursor().execute(f"KILL {victim_spid}")
+        except Exception as exc:
+            message = str(exc)
+            if "permission" in message.lower() or "kill" in message.lower():
+                print(
+                    f"Skipping: KILL not permitted for this login: {message}",
+                    file=sys.stderr,
+                )
+                victim.close()
+                admin.close()
+                sys.exit(77)
+            raise
+
+        try:
+            victim.close()
+        except Exception:
+            pass
+
+        admin.close()
+
+        first = connect(conn_str)
+        second = connect(conn_str)
+        try:
+            assert first.cursor().execute("SELECT 1").fetchone()[0] == 1
+            assert second.cursor().execute("SELECT 1").fetchone()[0] == 1
+        finally:
+            first.close()
+            second.close()
+            pooling(enabled=False)
+        """,
+        conn_str,
+    )
+
+
 def test_pool_recovery_after_failed_connection(conn_str):
     """Test that the pool recovers after a failed connection attempt."""
     pooling(max_size=1, idle_timeout=30)

From ff02a2a41478c2edca849d12b1e2336f2c197405 Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Thu, 10 Sep 2026 16:51:41 +0100
Subject: [PATCH 03/11] FIX: address pooled cleanup review findings

Preserve non-pooled cleanup, bind check-in to the originating pool generation, and harden regression coverage.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 mssql_python/connection.py                    |  8 ++++
 mssql_python/pybind/connection/connection.cpp |  7 +--
 mssql_python/pybind/connection/connection.h   |  5 ++
 .../pybind/connection/connection_pool.cpp     | 38 ++++++++-------
 .../pybind/connection/connection_pool.h       | 12 +++--
 tests/test_006_exceptions.py                  | 21 +++++++++
 tests/test_009_pooling.py                     | 47 +++++++++++++------
 7 files changed, 99 insertions(+), 39 deletions(-)

diff --git a/mssql_python/connection.py b/mssql_python/connection.py
index f63b923fc..00346b01d 100644
--- a/mssql_python/connection.py
+++ b/mssql_python/connection.py
@@ -2158,6 +2158,7 @@ def close(self) -> None:
         # Close the connection even if cursor cleanup had issues
         try:
             if self._conn:
+                rollback_error = None
                 if not self.autocommit and not self._pooling:
                     # If autocommit is disabled, rollback any uncommitted changes
                     # before disconnecting a non-pooled connection. Pooled
@@ -2165,6 +2166,10 @@ def close(self) -> None:
                     # which can discard the connection and release its capacity
                     # atomically if sanitation fails.
                     logger.debug("Rolling back uncommitted changes before closing connection.")
+                    try:
+                        self._conn.rollback()
+                    except RuntimeError as e:
+                        rollback_error = e
                 # TODO: Check potential race conditions in case of multithreaded scenarios
                 # Close the connection
                 try:
@@ -2173,6 +2178,9 @@ def close(self) -> None:
                     _raise_connection_error(e)
                 finally:
                     self._conn = None
+                if rollback_error is not None:
+                    # Preserve prior DB-API error mapping after deterministic cleanup.
+                    _raise_connection_error(rollback_error)
         except Exception as e:
             logger.error(f"Error closing database connection: {e}")
             # Re-raise the connection close error as it's more critical
diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp
index 822ef3abf..01397382c 100644
--- a/mssql_python/pybind/connection/connection.cpp
+++ b/mssql_python/pybind/connection/connection.cpp
@@ -652,7 +652,8 @@ ConnectionHandle::ConnectionHandle(const std::u16string& connStr, bool usePool,
     PERF_TIMER("ConnectionHandle::ConnectionHandle");
     if (_usePool) {
         _conn = ConnectionPoolManager::getInstance().acquireConnection(_connStr, attrsBefore,
-                                                                       _poolKey, tokenFactory);
+                                                                       _poolKey, tokenFactory,
+                                                                       &_originPool);
         // acquireConnection returns nullptr when pooling was disabled out from
         // under us (a disable_pooling() won the race). Fall back to a non-pooled
         // connection and flip _usePool so close() disconnects it directly rather
@@ -707,13 +708,13 @@ void ConnectionHandle::close() {
             // sanitized. Discarding also releases this connection's reserved
             // pool capacity. Preserve the original check-in error.
             try {
-                ConnectionPoolManager::getInstance().discardConnection(_poolKey, _conn);
+                ConnectionPoolManager::getInstance().discardConnection(_originPool, _conn);
             } catch (...) {
             }
             _conn = nullptr;
             throw;
         }
-        ConnectionPoolManager::getInstance().returnConnection(_poolKey, _conn);
+        ConnectionPoolManager::getInstance().returnConnection(_poolKey, _originPool, _conn);
     } else {
         _conn->disconnect();
     }
diff --git a/mssql_python/pybind/connection/connection.h b/mssql_python/pybind/connection/connection.h
index fbd1a88b5..5fa6f62e2 100644
--- a/mssql_python/pybind/connection/connection.h
+++ b/mssql_python/pybind/connection/connection.h
@@ -132,6 +132,8 @@ class Connection {
     mutable std::mutex _childHandlesMutex;
 };
 
+class ConnectionPool;
+
 class ConnectionHandle {
   public:
     ConnectionHandle(const std::u16string& connStr, bool usePool,
@@ -160,4 +162,7 @@ class ConnectionHandle {
     // Entra access-token auth so distinct identities never share a pool.
     // Empty is never stored; the ctor falls back to _connStr.
     std::u16string _poolKey;
+    // Identifies the exact pool generation that issued _conn. A weak reference
+    // prevents a checked-out connection from keeping a disabled pool alive.
+    std::weak_ptr<ConnectionPool> _originPool;
 };
diff --git a/mssql_python/pybind/connection/connection_pool.cpp b/mssql_python/pybind/connection/connection_pool.cpp
index 004af708f..df4b79101 100644
--- a/mssql_python/pybind/connection/connection_pool.cpp
+++ b/mssql_python/pybind/connection/connection_pool.cpp
@@ -402,7 +402,9 @@ ConnectionPoolManager& ConnectionPoolManager::getInstance() {
 std::shared_ptr<Connection> ConnectionPoolManager::acquireConnection(const std::u16string& connStr,
                                                                      const py::dict& attrs_before,
                                                                      const std::u16string& pool_key,
-                                                                     const py::object& token_factory) {
+                                                                     const py::object& token_factory,
+                                                                     std::weak_ptr<ConnectionPool>*
+                                                                         originating_pool) {
     PERF_TIMER("ConnectionPoolManager::acquireConnection");
     // Key the pool by pool_key when provided (identity-aware),
     // else fall back to the connection string (legacy behavior).
@@ -464,6 +466,9 @@ std::shared_ptr<Connection> ConnectionPoolManager::acquireConnection(const std::
             created = true;
         }
         pool = pool_ref;
+        if (originating_pool) {
+            *originating_pool = pool;
+        }
     }
     // Log after releasing _manager_mutex (#671): LOG() acquires the GIL, and
     // holding a native mutex across a GIL acquisition deadlocks a thread that
@@ -488,18 +493,18 @@ std::shared_ptr<Connection> ConnectionPoolManager::acquireConnection(const std::
     return pool->acquire(connStr, attrs_before, token_factory);
 }
 
-void ConnectionPoolManager::returnConnection(const std::u16string& pool_key,
-                                             const std::shared_ptr<Connection> conn) {
-    std::shared_ptr<ConnectionPool> pool;
+void ConnectionPoolManager::returnConnection(
+    const std::u16string& pool_key, const std::weak_ptr<ConnectionPool>& originating_pool,
+    const std::shared_ptr<Connection> conn) {
+    std::shared_ptr<ConnectionPool> pool = originating_pool.lock();
+    bool registered = false;
     {
         std::lock_guard<std::mutex> lock(_manager_mutex);
         auto it = _pools.find(pool_key);
-        if (it != _pools.end()) {
-            pool = it->second;
-        }
+        registered = pool && it != _pools.end() && it->second == pool;
     }
     // Call release() outside _manager_mutex to avoid deadlock.
-    if (pool) {
+    if (registered) {
         pool->release(conn);
     } else {
         // No pool is registered under this key (e.g. the pool was lazily
@@ -520,19 +525,16 @@ void ConnectionPoolManager::returnConnection(const std::u16string& pool_key,
     }
 }
 
-void ConnectionPoolManager::discardConnection(const std::u16string& pool_key,
-                                               const std::shared_ptr<Connection> conn) {
-    std::shared_ptr<ConnectionPool> pool;
-    {
-        std::lock_guard<std::mutex> lock(_manager_mutex);
-        auto it = _pools.find(pool_key);
-        if (it != _pools.end()) {
-            pool = it->second;
-        }
+void ConnectionPoolManager::discardConnection(
+    const std::weak_ptr<ConnectionPool>& originating_pool,
+    const std::shared_ptr<Connection> conn) {
+    if (!conn) {
+        return;
     }
+    std::shared_ptr<ConnectionPool> pool = originating_pool.lock();
     if (pool) {
         pool->discard(conn);
-    } else if (conn) {
+    } else {
         try {
             conn->disconnect();
         } catch (...) {
diff --git a/mssql_python/pybind/connection/connection_pool.h b/mssql_python/pybind/connection/connection_pool.h
index 40caf801a..6f67eab56 100644
--- a/mssql_python/pybind/connection/connection_pool.h
+++ b/mssql_python/pybind/connection/connection_pool.h
@@ -74,7 +74,8 @@ class ConnectionPoolManager {
     std::shared_ptr<Connection> acquireConnection(
         const std::u16string& conn_str, const py::dict& attrs_before = py::dict(),
         const std::u16string& pool_key = std::u16string(),
-        const py::object& token_factory = py::object());
+        const py::object& token_factory = py::object(),
+        std::weak_ptr<ConnectionPool>* originating_pool = nullptr);
 
     // Arms (true) or disarms (false) new-pool creation. Disarming, done under
     // _manager_mutex, guarantees that any acquireConnection() serialized after
@@ -83,10 +84,13 @@ class ConnectionPoolManager {
 
     // Returns a connection to its original pool, identified by pool_key
     // (the same key passed to acquireConnection).
-    void returnConnection(const std::u16string& pool_key, std::shared_ptr<Connection> conn);
+    void returnConnection(const std::u16string& pool_key,
+                          const std::weak_ptr<ConnectionPool>& originating_pool,
+                          std::shared_ptr<Connection> conn);
 
-    // Discards a connection that cannot safely be returned to its original pool.
-    void discardConnection(const std::u16string& pool_key, std::shared_ptr<Connection> conn);
+    // Discards a connection from the exact pool generation that issued it.
+    void discardConnection(const std::weak_ptr<ConnectionPool>& originating_pool,
+                           std::shared_ptr<Connection> conn);
 
     // Closes all pools and their connections
     void closePools();
diff --git a/tests/test_006_exceptions.py b/tests/test_006_exceptions.py
index b0d117761..1703e96cf 100644
--- a/tests/test_006_exceptions.py
+++ b/tests/test_006_exceptions.py
@@ -276,6 +276,27 @@ def test_connect_runtime_error_mapped_to_correct_dbapi_exception():
     assert not isinstance(exc_info.value, RuntimeError)
 
 
+def test_close_cleans_up_after_non_pooled_rollback_failure():
+    """A rollback error must not defer native close to object destruction."""
+    from unittest.mock import MagicMock, patch
+
+    mock_conn = MagicMock()
+    mock_conn.get_autocommit.return_value = False
+    mock_conn.rollback.side_effect = RuntimeError("SQLSTATE:08S01:Communication link failure")
+
+    with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=mock_conn):
+        conn = connect("Server=testserver;Database=mydb;Trusted_Connection=yes;")
+    conn._pooling = False
+
+    with pytest.raises(OperationalError, match="Communication link failure"):
+        conn.close()
+
+    mock_conn.rollback.assert_called_once_with()
+    mock_conn.close.assert_called_once_with()
+    assert conn._conn is None
+    assert conn.closed
+
+
 def test_truncate_error_message_successful_cases():
     """Test truncate_error_message with valid Microsoft messages for comparison."""
 
diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py
index b55db146d..149540df6 100644
--- a/tests/test_009_pooling.py
+++ b/tests/test_009_pooling.py
@@ -132,6 +132,7 @@ def test_pooled_close_paths_leave_no_open_transaction(conn_str):
     _run_in_subprocess(
         """
         import os
+        import sys
 
         import mssql_python
 
@@ -141,6 +142,23 @@ def test_pooled_close_paths_leave_no_open_transaction(conn_str):
         try:
             observer_cursor = observer.cursor()
 
+            def open_transaction_count(session_id):
+                try:
+                    observer_cursor.execute(
+                        "SELECT open_transaction_count "
+                        "FROM sys.dm_exec_sessions WHERE session_id = ?",
+                        [session_id],
+                    )
+                except Exception as exc:
+                    if "permission" in str(exc).lower():
+                        print(
+                            "Test login cannot inspect another SQL Server session",
+                            file=sys.stderr,
+                        )
+                        sys.exit(77)
+                    raise
+                return observer_cursor.fetchone()
+
             scenarios = (
                 ("direct commit", False, "SELECT 1", None, "commit"),
                 ("prepared commit", False, "SELECT CAST(? AS INT)", [1], "commit"),
@@ -163,14 +181,7 @@ def test_pooled_close_paths_leave_no_open_transaction(conn_str):
                             f"{name}: expected pooled SPID {expected_spid}, got {subject_spid}"
                         )
 
-                    observer_cursor.execute(
-                        "SELECT open_transaction_count "
-                        "FROM sys.dm_exec_sessions WHERE session_id = ?",
-                        [subject_spid],
-                    )
-                    if observer_cursor.fetchone() is None:
-                        import sys
-
+                    if open_transaction_count(subject_spid) is None:
                         print(
                             "Test login cannot inspect another SQL Server session",
                             file=sys.stderr,
@@ -190,12 +201,7 @@ def test_pooled_close_paths_leave_no_open_transaction(conn_str):
                 finally:
                     subject.close()
 
-                observer_cursor.execute(
-                    "SELECT open_transaction_count "
-                    "FROM sys.dm_exec_sessions WHERE session_id = ?",
-                    [subject_spid],
-                )
-                row = observer_cursor.fetchone()
+                row = open_transaction_count(subject_spid)
                 assert row is not None, f"{name}: parked SQL Server session was not visible"
                 assert row[0] == 0, (
                     f"{name}: pooled SPID {subject_spid} retained "
@@ -868,6 +874,7 @@ def test_failed_pool_sanitation_releases_capacity(conn_str):
         """
         import os
         import sys
+        import time
 
         from mssql_python import connect, pooling
 
@@ -895,10 +902,22 @@ def test_failed_pool_sanitation_releases_capacity(conn_str):
                 sys.exit(77)
             raise
 
+        deadline = time.monotonic() + 10
+        while True:
+            try:
+                victim.cursor().execute("SELECT 1").fetchone()
+            except Exception:
+                break
+            if time.monotonic() >= deadline:
+                raise AssertionError("KILL did not terminate the victim connection")
+            time.sleep(0.05)
+
         try:
             victim.close()
         except Exception:
             pass
+        else:
+            raise AssertionError("Expected pooled sanitation to fail after KILL")
 
         admin.close()
 

From 233e3bb5f5da8982eb229eca4b45f72f75e98317 Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Thu, 10 Sep 2026 19:57:49 +0100
Subject: [PATCH 04/11] FIX: preserve pooled pre-close rollback

Always invoke native close after rollback so pooled sanitation remains atomic while rapid pool toggles keep prior transaction behavior.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 mssql_python/connection.py | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

diff --git a/mssql_python/connection.py b/mssql_python/connection.py
index 00346b01d..49de88f88 100644
--- a/mssql_python/connection.py
+++ b/mssql_python/connection.py
@@ -2159,12 +2159,10 @@ def close(self) -> None:
         try:
             if self._conn:
                 rollback_error = None
-                if not self.autocommit and not self._pooling:
-                    # If autocommit is disabled, rollback any uncommitted changes
-                    # before disconnecting a non-pooled connection. Pooled
-                    # connections are rolled back by the native check-in path,
-                    # which can discard the connection and release its capacity
-                    # atomically if sanitation fails.
+                if not self.autocommit:
+                    # End caller work before native close. Pooled connections are
+                    # additionally restored to autocommit by native check-in,
+                    # which atomically discards them if sanitation fails.
                     logger.debug("Rolling back uncommitted changes before closing connection.")
                     try:
                         self._conn.rollback()

From e7e00a35e1b966f930db48a60bc602268f3d8158 Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Fri, 11 Sep 2026 06:37:05 +0100
Subject: [PATCH 05/11] FIX: harden pooled transaction sanitation

Always complete native cleanup after autocommit read failures and sanitize explicit transactions opened while autocommit is enabled.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 mssql_python/connection.py                    | 10 +++++++++-
 mssql_python/pybind/connection/connection.cpp | 19 +++++++++++--------
 tests/test_006_exceptions.py                  | 19 +++++++++++++++++++
 tests/test_009_pooling.py                     | 15 +++++++++++++--
 4 files changed, 52 insertions(+), 11 deletions(-)

diff --git a/mssql_python/connection.py b/mssql_python/connection.py
index 49de88f88..057b5748b 100644
--- a/mssql_python/connection.py
+++ b/mssql_python/connection.py
@@ -2158,8 +2158,14 @@ def close(self) -> None:
         # Close the connection even if cursor cleanup had issues
         try:
             if self._conn:
+                autocommit_error = None
                 rollback_error = None
-                if not self.autocommit:
+                manual_commit = False
+                try:
+                    manual_commit = not self._conn.get_autocommit()
+                except RuntimeError as e:
+                    autocommit_error = e
+                if manual_commit:
                     # End caller work before native close. Pooled connections are
                     # additionally restored to autocommit by native check-in,
                     # which atomically discards them if sanitation fails.
@@ -2179,6 +2185,8 @@ def close(self) -> None:
                 if rollback_error is not None:
                     # Preserve prior DB-API error mapping after deterministic cleanup.
                     _raise_connection_error(rollback_error)
+                if autocommit_error is not None:
+                    _raise_connection_error(autocommit_error)
         except Exception as e:
             logger.error(f"Error closing database connection: {e}")
             # Re-raise the connection close error as it's more critical
diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp
index 01397382c..0f9fa77c7 100644
--- a/mssql_python/pybind/connection/connection.cpp
+++ b/mssql_python/pybind/connection/connection.cpp
@@ -575,14 +575,17 @@ void Connection::prepareForPool() {
         ThrowStdException("Connection handle not allocated");
     }
 
-    if (!getAutocommit()) {
-        // End any caller transaction before check-in, then park the physical
-        // connection in autocommit mode. The SQL Server ODBC driver can leave
-        // an empty transaction visible after SQLEndTran while manual-commit
-        // mode remains enabled; switching modes ends that transaction.
-        rollback();
-        setAutocommit(true);
-    }
+    // Explicit BEGIN TRANSACTION is valid while ODBC autocommit is on, but
+    // SQLEndTran does not end that transaction until the connection enters
+    // manual-commit mode.
+    if (getAutocommit()) {
+        setAutocommit(false);
+    }
+    rollback();
+    // The SQL Server ODBC driver can leave an empty transaction visible after
+    // SQLEndTran while manual-commit mode remains enabled, so always park the
+    // physical connection in autocommit mode.
+    setAutocommit(true);
 }
 
 void Connection::updateLastUsed() {
diff --git a/tests/test_006_exceptions.py b/tests/test_006_exceptions.py
index 1703e96cf..1be24bd8a 100644
--- a/tests/test_006_exceptions.py
+++ b/tests/test_006_exceptions.py
@@ -297,6 +297,25 @@ def test_close_cleans_up_after_non_pooled_rollback_failure():
     assert conn.closed
 
 
+def test_close_cleans_up_after_autocommit_read_failure():
+    """An autocommit read error must not bypass native close and handle release."""
+    from unittest.mock import MagicMock, patch
+
+    mock_conn = MagicMock()
+    mock_conn.get_autocommit.side_effect = RuntimeError("SQLSTATE:08S01:Communication link failure")
+
+    with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=mock_conn):
+        conn = connect("Server=testserver;Database=mydb;Trusted_Connection=yes;")
+
+    with pytest.raises(OperationalError, match="Communication link failure"):
+        conn.close()
+
+    mock_conn.rollback.assert_not_called()
+    mock_conn.close.assert_called_once_with()
+    assert conn._conn is None
+    assert conn.closed
+
+
 def test_truncate_error_message_successful_cases():
     """Test truncate_error_message with valid Microsoft messages for comparison."""
 
diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py
index 149540df6..f2ca73546 100644
--- a/tests/test_009_pooling.py
+++ b/tests/test_009_pooling.py
@@ -165,6 +165,13 @@ def open_transaction_count(session_id):
                 ("explicit rollback", False, "SELECT 1", None, "rollback"),
                 ("implicit close rollback", False, "SELECT 1", None, None),
                 ("autocommit close", True, "SELECT 1", None, None),
+                (
+                    "explicit transaction in autocommit",
+                    True,
+                    "BEGIN TRANSACTION; SELECT 1",
+                    None,
+                    None,
+                ),
             )
             expected_spid = None
             for name, autocommit, sql, params, action in scenarios:
@@ -841,8 +848,12 @@ def session_identity(conn):
         # login_time, so the identity check below catches the only
         # failure mode that matters.
 
-        # Step 3: return both to the pool.
-        victim.close()
+        # Step 3: close both. Sanitation of the killed connection should fail,
+        # discard it, and may surface that connection error to the caller.
+        try:
+            victim.close()
+        except Exception:
+            pass
         admin.close()
 
         # Step 4: re-acquire from the pool. Each must be working; the

From b40587cb08035611bd73c184a7c88ad37bb9c9da Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Fri, 11 Sep 2026 07:14:30 +0100
Subject: [PATCH 06/11] FIX: finalize pooled cleanup safely

Avoid duplicate rollbacks, bypass pool sanitation during interpreter finalization, and disable reconnect in sanitation-failure coverage.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 mssql_python/connection.py                    |  2 +-
 mssql_python/pybind/connection/connection.cpp | 31 ++++++++++++++++---
 mssql_python/pybind/connection/connection.h   |  4 +--
 mssql_python/pybind/ddbc_bindings.cpp         |  3 +-
 tests/test_006_exceptions.py                  | 20 ++++++++++--
 tests/test_009_pooling.py                     | 10 ++++++
 6 files changed, 59 insertions(+), 11 deletions(-)

diff --git a/mssql_python/connection.py b/mssql_python/connection.py
index 057b5748b..0fff87bcb 100644
--- a/mssql_python/connection.py
+++ b/mssql_python/connection.py
@@ -2177,7 +2177,7 @@ def close(self) -> None:
                 # TODO: Check potential race conditions in case of multithreaded scenarios
                 # Close the connection
                 try:
-                    self._conn.close()
+                    self._conn.close(manual_commit and rollback_error is None)
                 except RuntimeError as e:
                     _raise_connection_error(e)
                 finally:
diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp
index 0f9fa77c7..e4a3c7754 100644
--- a/mssql_python/pybind/connection/connection.cpp
+++ b/mssql_python/pybind/connection/connection.cpp
@@ -18,6 +18,17 @@
 #include "logger_bridge.hpp"
 #include "performance_counter.hpp"
 
+static bool isPythonFinalizing() {
+    if (Py_IsInitialized() == 0) {
+        return true;
+    }
+#if PY_VERSION_HEX >= 0x030D0000
+    return Py_IsFinalizing() != 0;
+#else
+    return _Py_IsFinalizing() != 0;
+#endif
+}
+
 static SqlHandlePtr getEnvHandle() {
     static SqlHandlePtr envHandle = []() -> SqlHandlePtr {
         LOG("Allocating ODBC environment handle");
@@ -119,7 +130,7 @@ void Connection::disconnect() {
     // Py_IsInitialized() is checked first: after Py_Finalize() the interpreter is
     // gone and PyGILState_Check() is unreliable, so treat "not initialized" as
     // "no GIL" and skip all Python calls. (#671 follow-up)
-    bool hasGil = Py_IsInitialized() != 0 && PyGILState_Check() != 0;
+    bool hasGil = !isPythonFinalizing() && PyGILState_Check() != 0;
     if (_dbcHandle) {
         if (hasGil) {
             LOG("Disconnecting from database");
@@ -570,7 +581,7 @@ bool Connection::reset() {
     return true;
 }
 
-void Connection::prepareForPool() {
+void Connection::prepareForPool(bool transactionAlreadyRolledBack) {
     if (!_dbcHandle) {
         ThrowStdException("Connection handle not allocated");
     }
@@ -581,7 +592,9 @@ void Connection::prepareForPool() {
     if (getAutocommit()) {
         setAutocommit(false);
     }
-    rollback();
+    if (!transactionAlreadyRolledBack) {
+        rollback();
+    }
     // The SQL Server ODBC driver can leave an empty transaction visible after
     // SQLEndTran while manual-commit mode remains enabled, so always park the
     // physical connection in autocommit mode.
@@ -684,6 +697,14 @@ ConnectionHandle::ConnectionHandle(const std::u16string& connStr, bool usePool,
 
 ConnectionHandle::~ConnectionHandle() {
     if (_conn) {
+        if (isPythonFinalizing()) {
+            try {
+                _conn->disconnect();
+            } catch (...) {
+            }
+            _conn = nullptr;
+            return;
+        }
         try {
             close();
         } catch (...) {
@@ -698,14 +719,14 @@ ConnectionHandle::~ConnectionHandle() {
     }
 }
 
-void ConnectionHandle::close() {
+void ConnectionHandle::close(bool transactionAlreadyRolledBack) {
     PERF_TIMER("ConnectionHandle::close");
     if (!_conn) {
         ThrowStdException("Connection object is not initialized");
     }
     if (_usePool) {
         try {
-            _conn->prepareForPool();
+            _conn->prepareForPool(transactionAlreadyRolledBack);
         } catch (...) {
             // Never retain a connection whose transaction state could not be
             // sanitized. Discarding also releases this connection's reserved
diff --git a/mssql_python/pybind/connection/connection.h b/mssql_python/pybind/connection/connection.h
index 5fa6f62e2..caf71f1d7 100644
--- a/mssql_python/pybind/connection/connection.h
+++ b/mssql_python/pybind/connection/connection.h
@@ -53,7 +53,7 @@ class Connection {
     bool getAutocommit() const;
     bool isAlive() const;
     bool reset();
-    void prepareForPool();
+    void prepareForPool(bool transactionAlreadyRolledBack = false);
     void updateLastUsed();
     std::chrono::steady_clock::time_point lastUsed() const;
 
@@ -142,7 +142,7 @@ class ConnectionHandle {
                      const py::object& tokenFactory = py::object());
     ~ConnectionHandle();
 
-    void close();
+    void close(bool transactionAlreadyRolledBack = false);
     void commit();
     void rollback();
     void setAutocommit(bool enabled);
diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp
index 26948ee88..458334592 100644
--- a/mssql_python/pybind/ddbc_bindings.cpp
+++ b/mssql_python/pybind/ddbc_bindings.cpp
@@ -6094,7 +6094,8 @@ PYBIND11_MODULE(ddbc_bindings, m) {
                       const py::object&>(),
              py::arg("conn_str"), py::arg("use_pool"), py::arg("attrs_before") = py::dict(),
              py::arg("pool_key") = std::u16string(), py::arg("token_factory") = py::none())
-        .def("close", &ConnectionHandle::close, "Close the connection")
+        .def("close", &ConnectionHandle::close,
+             py::arg("transaction_already_rolled_back") = false, "Close the connection")
         .def("commit", &ConnectionHandle::commit, "Commit the current transaction")
         .def("rollback", &ConnectionHandle::rollback, "Rollback the current transaction")
         .def("set_autocommit", &ConnectionHandle::setAutocommit)
diff --git a/tests/test_006_exceptions.py b/tests/test_006_exceptions.py
index 1be24bd8a..74f4178ae 100644
--- a/tests/test_006_exceptions.py
+++ b/tests/test_006_exceptions.py
@@ -292,7 +292,7 @@ def test_close_cleans_up_after_non_pooled_rollback_failure():
         conn.close()
 
     mock_conn.rollback.assert_called_once_with()
-    mock_conn.close.assert_called_once_with()
+    mock_conn.close.assert_called_once_with(False)
     assert conn._conn is None
     assert conn.closed
 
@@ -311,11 +311,27 @@ def test_close_cleans_up_after_autocommit_read_failure():
         conn.close()
 
     mock_conn.rollback.assert_not_called()
-    mock_conn.close.assert_called_once_with()
+    mock_conn.close.assert_called_once_with(False)
     assert conn._conn is None
     assert conn.closed
 
 
+def test_close_reports_successful_rollback_to_native_cleanup():
+    """Native pool cleanup must not repeat a successful Python rollback."""
+    from unittest.mock import MagicMock, patch
+
+    mock_conn = MagicMock()
+    mock_conn.get_autocommit.return_value = False
+
+    with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=mock_conn):
+        conn = connect("Server=testserver;Database=mydb;Trusted_Connection=yes;")
+
+    conn.close()
+
+    mock_conn.rollback.assert_called_once_with()
+    mock_conn.close.assert_called_once_with(True)
+
+
 def test_truncate_error_message_successful_cases():
     """Test truncate_error_message with valid Microsoft messages for comparison."""
 
diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py
index f2ca73546..6ec0513a9 100644
--- a/tests/test_009_pooling.py
+++ b/tests/test_009_pooling.py
@@ -888,8 +888,18 @@ def test_failed_pool_sanitation_releases_capacity(conn_str):
         import time
 
         from mssql_python import connect, pooling
+        from mssql_python.connection_string_builder import _ConnectionStringBuilder
+        from mssql_python.connection_string_parser import _ConnectionStringParser
 
         conn_str = os.environ["DB_CONNECTION_STRING"]
+        parsed = _ConnectionStringParser(validate_keywords=True)._parse(conn_str)
+        normalized = {}
+        for key, value in parsed.items():
+            canonical = _ConnectionStringParser.normalize_key(key)
+            if canonical not in normalized:
+                normalized[canonical] = value
+        normalized["ConnectRetryCount"] = "0"
+        conn_str = _ConnectionStringBuilder(normalized).build()
         pooling(max_size=2, idle_timeout=30)
         victim = connect(conn_str)
         admin = connect(conn_str, autocommit=True)

From da81ba73fd0f72acabd46161fe3b9c4ab451c252 Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Fri, 11 Sep 2026 07:47:27 +0100
Subject: [PATCH 07/11] TEST: complete pooled cleanup coverage

Cover close error precedence, explicit transaction data integrity, pool-generation isolation, and destructor capacity recovery while avoiding ODBC calls during finalization.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 mssql_python/pybind/connection/connection.cpp |  21 ++-
 mssql_python/pybind/connection/connection.h   |   3 +
 tests/test_006_exceptions.py                  |  46 +++++-
 tests/test_009_pooling.py                     | 155 ++++++++++++++++++
 4 files changed, 218 insertions(+), 7 deletions(-)

diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp
index e4a3c7754..e2ed5c13b 100644
--- a/mssql_python/pybind/connection/connection.cpp
+++ b/mssql_python/pybind/connection/connection.cpp
@@ -211,6 +211,17 @@ void Connection::disconnect() {
     }
 }
 
+void Connection::abandonDuringFinalization() noexcept {
+    {
+        std::lock_guard<std::mutex> lock(_childHandlesMutex);
+        _childStatementHandles.clear();
+        _allocationsSinceCompaction = 0;
+    }
+    // SqlHandle::free() already suppresses SQLFreeHandle during finalization.
+    // Clearing the shared pointer leaves process teardown to the operating system.
+    _dbcHandle.reset();
+}
+
 // TODO(microsoft): Add an exception class in C++ for error handling,
 // DB spec compliant
 void Connection::checkError(SQLRETURN ret) const {
@@ -698,15 +709,15 @@ ConnectionHandle::ConnectionHandle(const std::u16string& connStr, bool usePool,
 ConnectionHandle::~ConnectionHandle() {
     if (_conn) {
         if (isPythonFinalizing()) {
-            try {
-                _conn->disconnect();
-            } catch (...) {
-            }
+            _conn->abandonDuringFinalization();
             _conn = nullptr;
             return;
         }
         try {
-            close();
+            // A destructor cannot report sanitation errors to a caller. Discard
+            // instead of running close(), which performs logging and transaction
+            // operations that are unsafe during late object teardown.
+            ConnectionPoolManager::getInstance().discardConnection(_originPool, _conn);
         } catch (...) {
             if (_conn) {
                 try {
diff --git a/mssql_python/pybind/connection/connection.h b/mssql_python/pybind/connection/connection.h
index caf71f1d7..7d35a670a 100644
--- a/mssql_python/pybind/connection/connection.h
+++ b/mssql_python/pybind/connection/connection.h
@@ -40,6 +40,9 @@ class Connection {
     // Disconnect and free the connection handle.
     void disconnect();
 
+    // Relinquish native handles without ODBC calls during interpreter finalization.
+    void abandonDuringFinalization() noexcept;
+
     // Commit the current transaction.
     void commit();
 
diff --git a/tests/test_006_exceptions.py b/tests/test_006_exceptions.py
index 74f4178ae..371b9662f 100644
--- a/tests/test_006_exceptions.py
+++ b/tests/test_006_exceptions.py
@@ -276,7 +276,7 @@ def test_connect_runtime_error_mapped_to_correct_dbapi_exception():
     assert not isinstance(exc_info.value, RuntimeError)
 
 
-def test_close_cleans_up_after_non_pooled_rollback_failure():
+def test_close_cleans_up_after_rollback_failure():
     """A rollback error must not defer native close to object destruction."""
     from unittest.mock import MagicMock, patch
 
@@ -286,7 +286,6 @@ def test_close_cleans_up_after_non_pooled_rollback_failure():
 
     with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=mock_conn):
         conn = connect("Server=testserver;Database=mydb;Trusted_Connection=yes;")
-    conn._pooling = False
 
     with pytest.raises(OperationalError, match="Communication link failure"):
         conn.close()
@@ -332,6 +331,49 @@ def test_close_reports_successful_rollback_to_native_cleanup():
     mock_conn.close.assert_called_once_with(True)
 
 
+def test_autocommit_close_delegates_transaction_cleanup_to_native():
+    """Autocommit may still contain an explicit SQL transaction."""
+    from unittest.mock import MagicMock, patch
+
+    mock_conn = MagicMock()
+    mock_conn.get_autocommit.return_value = True
+
+    with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=mock_conn):
+        conn = connect(
+            "Server=testserver;Database=mydb;Trusted_Connection=yes;",
+            autocommit=True,
+        )
+
+    conn.close()
+
+    mock_conn.rollback.assert_not_called()
+    mock_conn.close.assert_called_once_with(False)
+
+
+@pytest.mark.parametrize("preclose_failure", ["autocommit", "rollback"])
+def test_native_close_error_takes_precedence_over_preclose_failure(preclose_failure):
+    """The native close error wins, but the wrapper still releases its handle."""
+    from unittest.mock import MagicMock, patch
+
+    mock_conn = MagicMock()
+    if preclose_failure == "autocommit":
+        mock_conn.get_autocommit.side_effect = RuntimeError("SQLSTATE:08S01:Autocommit read failed")
+    else:
+        mock_conn.get_autocommit.return_value = False
+        mock_conn.rollback.side_effect = RuntimeError("SQLSTATE:08S01:Rollback failed")
+    mock_conn.close.side_effect = RuntimeError("SQLSTATE:08003:Native close failed")
+
+    with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=mock_conn):
+        conn = connect("Server=testserver;Database=mydb;Trusted_Connection=yes;")
+
+    with pytest.raises(OperationalError, match="Native close failed"):
+        conn.close()
+
+    mock_conn.close.assert_called_once_with(False)
+    assert conn._conn is None
+    assert conn.closed
+
+
 def test_truncate_error_message_successful_cases():
     """Test truncate_error_message with valid Microsoft messages for comparison."""
 
diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py
index 6ec0513a9..57fa3542e 100644
--- a/tests/test_009_pooling.py
+++ b/tests/test_009_pooling.py
@@ -224,6 +224,82 @@ def open_transaction_count(session_id):
     )
 
 
+def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str):
+    """Autocommit normalization must not commit an explicit SQL transaction."""
+    _run_in_subprocess(
+        """
+        import os
+
+        import mssql_python
+
+        conn_str = os.environ["DB_CONNECTION_STRING"]
+        table = "pytest_pool_explicit_autocommit_transaction"
+        mssql_python.pooling(enabled=True, max_size=2, idle_timeout=30)
+        observer = mssql_python.connect(conn_str, autocommit=True)
+        try:
+            observer_cursor = observer.cursor()
+            observer_cursor.execute(f"DROP TABLE IF EXISTS {table}")
+            observer_cursor.execute(f"CREATE TABLE {table} (id INT PRIMARY KEY)")
+
+            subject = mssql_python.connect(conn_str, autocommit=True)
+            subject_cursor = subject.cursor()
+            subject_cursor.execute("SELECT @@SPID")
+            subject_spid = subject_cursor.fetchone()[0]
+            subject_cursor.execute(f"BEGIN TRANSACTION; INSERT INTO {table} VALUES (1)")
+            subject_cursor.close()
+            subject.close()
+
+            try:
+                observer_cursor.execute(
+                    "SELECT open_transaction_count "
+                    "FROM sys.dm_exec_sessions WHERE session_id = ?",
+                    [subject_spid],
+                )
+            except Exception as exc:
+                if "permission" in str(exc).lower():
+                    import sys
+
+                    print(
+                        "Test login cannot inspect another SQL Server session",
+                        file=sys.stderr,
+                    )
+                    sys.exit(77)
+                raise
+            row = observer_cursor.fetchone()
+            if row is None:
+                import sys
+
+                print(
+                    "Test login cannot inspect another SQL Server session",
+                    file=sys.stderr,
+                )
+                sys.exit(77)
+            assert row[0] == 0
+
+            observer_cursor.execute(f"SELECT COUNT(*) FROM {table}")
+            assert observer_cursor.fetchone()[0] == 0
+
+            reused = mssql_python.connect(conn_str, autocommit=True)
+            try:
+                reused_cursor = reused.cursor()
+                reused_cursor.execute("SELECT @@SPID, @@TRANCOUNT")
+                reused_spid, transaction_count = reused_cursor.fetchone()
+                assert reused_spid == subject_spid
+                assert transaction_count == 0
+                reused_cursor.close()
+            finally:
+                reused.close()
+
+            observer_cursor.execute(f"DROP TABLE {table}")
+            observer_cursor.close()
+        finally:
+            observer.close()
+            mssql_python.pooling(enabled=False)
+        """,
+        conn_str,
+    )
+
+
 def test_connection_pooling_isolation_level_reset(conn_str):
     """Test that pooling correctly resets session state for isolation level.
 
@@ -956,6 +1032,85 @@ def test_failed_pool_sanitation_releases_capacity(conn_str):
     )
 
 
+def test_old_pool_generation_cannot_enter_replacement_pool(conn_str):
+    """A stale checked-out connection must not alter its replacement pool."""
+    _run_in_subprocess(
+        """
+        import os
+
+        from mssql_python import connect, pooling
+
+        conn_str = os.environ["DB_CONNECTION_STRING"]
+        pooling(max_size=2, idle_timeout=30)
+        old = connect(conn_str, autocommit=True)
+        old_cursor = old.cursor()
+        old_cursor.execute("SELECT @@SPID")
+        old_spid = old_cursor.fetchone()[0]
+        old_cursor.close()
+
+        pooling(enabled=False)
+        pooling(enabled=True, max_size=2, idle_timeout=30)
+        first = connect(conn_str, autocommit=True)
+        second = connect(conn_str, autocommit=True)
+        try:
+            first_spid = first.cursor().execute("SELECT @@SPID").fetchone()[0]
+            second_spid = second.cursor().execute("SELECT @@SPID").fetchone()[0]
+            assert first_spid != second_spid
+            assert old_spid not in (first_spid, second_spid)
+
+            old.close()
+
+            try:
+                third = connect(conn_str, autocommit=True)
+            except Exception as exc:
+                assert "pool" in str(exc).lower()
+            else:
+                third.close()
+                raise AssertionError(
+                    "Stale connection entered or decremented the replacement pool"
+                )
+        finally:
+            old.close()
+            first.close()
+            second.close()
+            pooling(enabled=False)
+        """,
+        conn_str,
+    )
+
+
+def test_unclosed_native_handle_destructor_releases_pool_capacity(conn_str):
+    """Native destructor fallback must discard its checked-out pool slot."""
+    _run_in_subprocess(
+        """
+        import gc
+        import os
+
+        import mssql_python
+        from mssql_python import connect, pooling
+
+        conn_str = os.environ["DB_CONNECTION_STRING"]
+        pooling(max_size=1, idle_timeout=30)
+        wrapper = connect(conn_str, autocommit=True)
+        native = wrapper._conn
+        wrapper._conn = None
+        wrapper._closed = True
+        mssql_python._active_connections.discard(wrapper)
+        del wrapper
+        del native
+        gc.collect()
+
+        replacement = connect(conn_str, autocommit=True)
+        try:
+            assert replacement.cursor().execute("SELECT 1").fetchone()[0] == 1
+        finally:
+            replacement.close()
+            pooling(enabled=False)
+        """,
+        conn_str,
+    )
+
+
 def test_pool_recovery_after_failed_connection(conn_str):
     """Test that the pool recovers after a failed connection attempt."""
     pooling(max_size=1, idle_timeout=30)

From 7adc5630b2891e548aeb852e91730b2b051e7a8a Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Thu, 17 Sep 2026 08:53:21 +0100
Subject: [PATCH 08/11] FIX: roll back abandoned native connections safely

Dispose native connections without Python callbacks, retain statement wrappers across disconnect, and cover pending transactions and concurrent GC. Tighten sanitation permission skips and pre-close DMV visibility checks.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 CHANGELOG.md                                  |   4 +-
 mssql_python/pybind/connection/connection.cpp | 113 ++++++-----
 mssql_python/pybind/connection/connection.h   |   3 +
 .../pybind/connection/connection_pool.cpp     |  13 +-
 tests/test_009_pooling.py                     | 175 +++++++++++++++++-
 5 files changed, 245 insertions(+), 63 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index ebcf909ae..9ee5ed117 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -71,7 +71,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Fixed
 - **GH-754:** Pooled connections are now rolled back and restored to autocommit
   mode before being parked. This prevents an empty transaction from remaining
-  visible on an idle SQL Server session after `Connection.close()`.
+  visible on an idle SQL Server session after `Connection.close()`. Abandoned
+  native connections also roll back pending work before disconnecting during
+  normal object destruction.
 - **GH-740:** A Python `Decimal` whose value falls in the SQL Server MONEY /
   SMALLMONEY range is now bound as `SQL_NUMERIC` with its own precision and scale
   on both `execute()` paths (native detection, and the legacy path reached when
diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp
index e2ed5c13b..847c5fc8e 100644
--- a/mssql_python/pybind/connection/connection.cpp
+++ b/mssql_python/pybind/connection/connection.cpp
@@ -5,6 +5,7 @@
 #include "connection/connection_pool.h"
 #include "utf_utils.h"
 #include <algorithm>
+#include <cstdio>
 #include <memory>
 #include <pybind11/pybind11.h>
 #include <regex>
@@ -64,13 +65,7 @@ Connection::Connection(const std::u16string& conn_str, bool use_pool)
 }
 
 Connection::~Connection() noexcept {
-    try {
-        disconnect();  // fallback if user forgets to disconnect
-    } catch (...) {
-        // Destructors must not propagate ODBC disconnect failures. Releasing
-        // the handle still lets SqlHandle perform SQLFreeHandle cleanup.
-        _dbcHandle.reset();
-    }
+    disconnectNoThrow();
 }
 
 // Allocates connection handle
@@ -136,13 +131,10 @@ void Connection::disconnect() {
             LOG("Disconnecting from database");
         }
 
-        // CRITICAL FIX: Mark all child statement handles as implicitly freed
-        // When we free the DBC handle below, the ODBC driver will automatically free
-        // all child STMT handles. We need to tell the SqlHandle objects about this
-        // so they don't try to free the handles again during their destruction.
-        
-        // THREAD-SAFETY: Lock mutex to safely access _childStatementHandles
-        // This protects against concurrent allocStatementHandle() calls or GC finalizers
+        // Keep wrappers alive while SQLDisconnect frees their native handles.
+        // Otherwise another thread's GC could free a statement again before we
+        // mark it as implicitly freed. On failure, keep the handles usable.
+        std::vector<SqlHandlePtr> childHandles;
         size_t originalSize = 0, afterCompactSize = 0, badHandleCount = 0;
         {
             std::lock_guard<std::mutex> lock(_childHandlesMutex);
@@ -164,12 +156,36 @@ void Connection::disconnect() {
                         ++badHandleCount;
                         continue;  // Skip marking to prevent leak
                     }
-                    handle->markImplicitlyFreed();
+                    childHandles.push_back(std::move(handle));
                 }
             }
+        }
+
+        SQLRETURN ret;
+        if (hasGil) {
+            py::gil_scoped_release release;
+            ret = SQLDisconnect_ptr(_dbcHandle->get());
+        } else {
+            ret = SQLDisconnect_ptr(_dbcHandle->get());
+        }
+        if (!SQL_SUCCEEDED(ret)) {
+            if (hasGil) {
+                checkError(ret);
+            } else {
+                std::fprintf(stderr, "mssql-python: native disconnect failed (SQLRETURN %d)\n",
+                             static_cast<int>(ret));
+            }
+            // Keep ownership and child-handle tracking intact for a cleanup retry.
+            return;
+        }
+        {
+            std::lock_guard<std::mutex> lock(_childHandlesMutex);
+            for (const auto& handle : childHandles) {
+                handle->markImplicitlyFreed();
+            }
             _childStatementHandles.clear();
             _allocationsSinceCompaction = 0;
-        }  // Release lock before potentially slow SQLDisconnect call
+        }
 
         // Log after releasing _childHandlesMutex (#671): LOG()/LOG_ERROR() acquire
         // the GIL and must not run while a native mutex is held. Also gated on
@@ -184,26 +200,6 @@ void Connection::disconnect() {
             }
         }
 
-        SQLRETURN ret;
-        if (hasGil) {
-            // Release the GIL during the blocking ODBC disconnect call.
-            // This allows other Python threads to run while the network
-            // round-trip completes.
-            py::gil_scoped_release release;
-            ret = SQLDisconnect_ptr(_dbcHandle->get());
-        } else {
-            // Destructor / shutdown path — GIL is not held, call directly.
-            ret = SQLDisconnect_ptr(_dbcHandle->get());
-        }
-        // In destructor/shutdown paths, suppress errors to avoid
-        // std::terminate() if this throws during stack unwinding.
-        if (hasGil) {
-            checkError(ret);
-        } else if (!SQL_SUCCEEDED(ret)) {
-            // Intentionally no LOG() here: LOG() acquires the GIL internally
-            // via py::gil_scoped_acquire, which is unsafe during interpreter
-            // shutdown or stack unwinding (can deadlock or call std::terminate).
-        }
         // triggers SQLFreeHandle via destructor, if last owner
         _dbcHandle.reset();
     } else if (hasGil) {
@@ -211,6 +207,37 @@ void Connection::disconnect() {
     }
 }
 
+void Connection::disconnectNoThrow() noexcept {
+    try {
+        if (isPythonFinalizing()) {
+            abandonDuringFinalization();
+            return;
+        }
+        if (!_dbcHandle) {
+            return;
+        }
+        auto cleanup = [this]() {
+            // SQLEndTran only rolls back explicit SQL transactions after entering
+            // manual-commit mode. Never turn autocommit on here: that could commit.
+            SQLSetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT,
+                                 reinterpret_cast<SQLPOINTER>(SQL_AUTOCOMMIT_OFF), 0);
+            SQLEndTran_ptr(SQL_HANDLE_DBC, _dbcHandle->get(), SQL_ROLLBACK);
+            // Attempt disconnect even if rollback failed (e.g. a dead connection).
+            disconnect();
+        };
+        // disconnect() already supports GIL-less cleanup. Drop the GIL once so
+        // neither its diagnostics nor handle destruction can enter Python.
+        if (PyGILState_Check()) {
+            py::gil_scoped_release release;
+            cleanup();
+        } else {
+            cleanup();
+        }
+    } catch (...) {
+        std::fputs("mssql-python: unexpected failure during native connection cleanup\n", stderr);
+    }
+}
+
 void Connection::abandonDuringFinalization() noexcept {
     {
         std::lock_guard<std::mutex> lock(_childHandlesMutex);
@@ -714,18 +741,12 @@ ConnectionHandle::~ConnectionHandle() {
             return;
         }
         try {
-            // A destructor cannot report sanitation errors to a caller. Discard
-            // instead of running close(), which performs logging and transaction
-            // operations that are unsafe during late object teardown.
+            // Discard ends abandoned work without returning this connection to
+            // the pool or entering Python from a native destructor.
             ConnectionPoolManager::getInstance().discardConnection(_originPool, _conn);
         } catch (...) {
-            if (_conn) {
-                try {
-                    _conn->disconnect();
-                } catch (...) {
-                }
-            }
-            _conn = nullptr;
+            std::fputs("mssql-python: failed to release native connection pool capacity\n", stderr);
+            _conn->disconnectNoThrow();
         }
     }
 }
diff --git a/mssql_python/pybind/connection/connection.h b/mssql_python/pybind/connection/connection.h
index 7d35a670a..f14f83777 100644
--- a/mssql_python/pybind/connection/connection.h
+++ b/mssql_python/pybind/connection/connection.h
@@ -40,6 +40,9 @@ class Connection {
     // Disconnect and free the connection handle.
     void disconnect();
 
+    // Roll back and disconnect without Python callbacks or escaping exceptions.
+    void disconnectNoThrow() noexcept;
+
     // Relinquish native handles without ODBC calls during interpreter finalization.
     void abandonDuringFinalization() noexcept;
 
diff --git a/mssql_python/pybind/connection/connection_pool.cpp b/mssql_python/pybind/connection/connection_pool.cpp
index df4b79101..3ea3ccc5f 100644
--- a/mssql_python/pybind/connection/connection_pool.cpp
+++ b/mssql_python/pybind/connection/connection_pool.cpp
@@ -335,12 +335,7 @@ void ConnectionPool::discard(std::shared_ptr<Connection> conn) {
         if (_current_size > 0)
             --_current_size;
     }
-    try {
-        conn->disconnect();
-    } catch (...) {
-        // The caller is already handling a sanitation failure. Connection's
-        // noexcept destructor still releases the ODBC handle.
-    }
+    conn->disconnectNoThrow();
 }
 
 bool ConnectionPool::canEvict() {
@@ -535,11 +530,7 @@ void ConnectionPoolManager::discardConnection(
     if (pool) {
         pool->discard(conn);
     } else {
-        try {
-            conn->disconnect();
-        } catch (...) {
-            // Connection's noexcept destructor still releases the ODBC handle.
-        }
+        conn->disconnectNoThrow();
     }
 }
 
diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py
index 57fa3542e..3db792abb 100644
--- a/tests/test_009_pooling.py
+++ b/tests/test_009_pooling.py
@@ -245,10 +245,6 @@ def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str
             subject_cursor = subject.cursor()
             subject_cursor.execute("SELECT @@SPID")
             subject_spid = subject_cursor.fetchone()[0]
-            subject_cursor.execute(f"BEGIN TRANSACTION; INSERT INTO {table} VALUES (1)")
-            subject_cursor.close()
-            subject.close()
-
             try:
                 observer_cursor.execute(
                     "SELECT open_transaction_count "
@@ -274,6 +270,18 @@ def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str
                     file=sys.stderr,
                 )
                 sys.exit(77)
+
+            subject_cursor.execute(f"BEGIN TRANSACTION; INSERT INTO {table} VALUES (1)")
+            subject_cursor.close()
+            subject.close()
+
+            observer_cursor.execute(
+                "SELECT open_transaction_count "
+                "FROM sys.dm_exec_sessions WHERE session_id = ?",
+                [subject_spid],
+            )
+            row = observer_cursor.fetchone()
+            assert row is not None, "Previously visible pooled session disappeared on close"
             assert row[0] == 0
 
             observer_cursor.execute(f"SELECT COUNT(*) FROM {table}")
@@ -989,7 +997,7 @@ def test_failed_pool_sanitation_releases_capacity(conn_str):
             admin.cursor().execute(f"KILL {victim_spid}")
         except Exception as exc:
             message = str(exc)
-            if "permission" in message.lower() or "kill" in message.lower():
+            if "does not have permission to use the kill statement" in message.lower():
                 print(
                     f"Skipping: KILL not permitted for this login: {message}",
                     file=sys.stderr,
@@ -1111,6 +1119,163 @@ def test_unclosed_native_handle_destructor_releases_pool_capacity(conn_str):
     )
 
 
+@pytest.mark.parametrize("use_pool", [False, True])
+@pytest.mark.parametrize("autocommit", [False, True])
+def test_native_destructor_rolls_back_pending_dml(conn_str, use_pool, autocommit):
+    """Native destruction must release transactions, locks, and the server session."""
+    _run_in_subprocess(
+        f"use_pool = {use_pool!r}\nautocommit = {autocommit!r}\n" + textwrap.dedent("""
+            import gc
+            import os
+            import sys
+            import time
+            import uuid
+
+            from mssql_python import connect, ddbc_bindings as ddbc, pooling
+
+            conn_str = os.environ["DB_CONNECTION_STRING"]
+            pool_key = "pytest_native_cleanup_" + uuid.uuid4().hex
+            table = pool_key
+            pooling(max_size=1, idle_timeout=30)
+            observer = connect(conn_str, autocommit=True)
+            native = ddbc.Connection(conn_str, use_pool, {}, pool_key, None)
+            statement = native.alloc_statement_handle()
+            try:
+                assert ddbc.DDBCSQLExecDirect(statement, "SELECT @@SPID") in (0, 1)
+                row = []
+                assert ddbc.DDBCSQLFetchOne(statement, row) in (0, 1)
+                session_id = row[0]
+                statement.free()
+
+                cursor = observer.cursor()
+                try:
+                    cursor.execute(
+                        "SELECT session_id FROM sys.dm_exec_sessions WHERE session_id = ?",
+                        [session_id],
+                    )
+                except Exception as exc:
+                    if "permission" in str(exc).lower():
+                        print("Observer cannot inspect the native session", file=sys.stderr)
+                        sys.exit(77)
+                    raise
+                if cursor.fetchone() is None:
+                    print("Observer cannot inspect the native session", file=sys.stderr)
+                    sys.exit(77)
+
+                cursor.execute("SET LOCK_TIMEOUT 1000")
+                cursor.execute(f"CREATE TABLE {table} (id INT)")
+                try:
+                    native.set_autocommit(autocommit)
+                    statement = native.alloc_statement_handle()
+                    sql = f"INSERT INTO {table} VALUES (1)"
+                    if autocommit:
+                        sql = "BEGIN TRANSACTION; " + sql
+                    assert ddbc.DDBCSQLExecDirect(statement, sql) in (0, 1)
+                    statement.free()
+                    statement = None
+                    native = None
+                    gc.collect()
+
+                    cursor.execute(f"SELECT COUNT(*) FROM {table} WITH (READCOMMITTEDLOCK)")
+                    assert cursor.fetchone()[0] == 0, "Destructor committed abandoned work"
+
+                    deadline = time.monotonic() + 5
+                    while True:
+                        cursor.execute(
+                            "SELECT session_id FROM sys.dm_exec_sessions WHERE session_id = ?",
+                            [session_id],
+                        )
+                        if cursor.fetchone() is None:
+                            break
+                        assert time.monotonic() < deadline, "Native session survived destruction"
+                        time.sleep(0.05)
+
+                    replacement = ddbc.Connection(conn_str, use_pool, {}, pool_key, None)
+                    replacement_statement = replacement.alloc_statement_handle()
+                    try:
+                        assert ddbc.DDBCSQLExecDirect(replacement_statement, "SELECT 1") in (0, 1)
+                        row = []
+                        assert ddbc.DDBCSQLFetchOne(replacement_statement, row) in (0, 1)
+                        assert row == [1]
+                    finally:
+                        replacement_statement.free()
+                        replacement.close()
+                finally:
+                    cursor.execute(f"DROP TABLE {table}")
+                cursor.close()
+            finally:
+                if statement is not None:
+                    statement.free()
+                if native is not None:
+                    native.rollback()
+                    native.close()
+                observer.close()
+                pooling(enabled=False)
+            """),
+        conn_str,
+    )
+
+
+@pytest.mark.parametrize("explicit_close", [False, True])
+def test_native_disconnect_with_concurrent_child_gc(conn_str, explicit_close):
+    """Child wrappers collected during disconnect must not double-free statements."""
+    _run_in_subprocess(
+        f"explicit_close = {explicit_close!r}\n" + textwrap.dedent("""
+            import gc
+            import os
+            import threading
+
+            from mssql_python import ddbc_bindings as ddbc
+
+            class StatementCycle:
+                def __init__(self, statement):
+                    self.statement = statement
+                    self.cycle = self
+
+            barrier = threading.Barrier(2, timeout=10)
+            errors = []
+            iterations = 50
+
+            def collect_children():
+                try:
+                    for _ in range(iterations):
+                        barrier.wait()
+                        gc.collect()
+                        barrier.wait()
+                except Exception as exc:
+                    errors.append(exc)
+                    barrier.abort()
+
+            gc.disable()
+            collector = threading.Thread(target=collect_children, daemon=True)
+            collector.start()
+            try:
+                for _ in range(iterations):
+                    native = ddbc.Connection(os.environ["DB_CONNECTION_STRING"], False)
+                    native.set_autocommit(True)
+                    statement = native.alloc_statement_handle()
+                    assert ddbc.DDBCSQLExecDirect(statement, "SELECT 1") in (0, 1)
+                    cycle = StatementCycle(statement)
+                    del statement, cycle
+                    barrier.wait()
+                    if explicit_close:
+                        native.close()
+                    native = None
+                    barrier.wait()
+            finally:
+                collector.join(timeout=10)
+                if collector.is_alive():
+                    barrier.abort()
+                    collector.join(timeout=10)
+                gc.enable()
+            assert not collector.is_alive(), "GC worker did not exit"
+            assert not errors, errors
+            gc.collect()
+            """),
+        conn_str,
+    )
+
+
 def test_pool_recovery_after_failed_connection(conn_str):
     """Test that the pool recovers after a failed connection attempt."""
     pooling(max_size=1, idle_timeout=30)

From 88ce92df2df31ede9d7fb5fca42522d23862d87c Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Thu, 17 Sep 2026 09:42:16 +0100
Subject: [PATCH 09/11] FIX: synchronize statement cleanup with native
 disconnect

Share native cleanup state across connection and statement lifetimes. Serialize explicit frees, finalizers, and disconnect without holding the GIL, preserve handles on disconnect failure, and cover cursor finalization and native free entry points.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 CHANGELOG.md                                  |   3 +-
 mssql_python/pybind/connection/connection.cpp |  99 ++++----
 mssql_python/pybind/connection/connection.h   |   5 +-
 mssql_python/pybind/ddbc_bindings.cpp         | 186 ++++++--------
 mssql_python/pybind/ddbc_bindings.h           |  12 +-
 tests/test_009_pooling.py                     | 240 ++++++++++++++++++
 6 files changed, 384 insertions(+), 161 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index bc660c77c..511edefa7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -98,7 +98,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
   mode before being parked. This prevents an empty transaction from remaining
   visible on an idle SQL Server session after `Connection.close()`. Abandoned
   native connections also roll back pending work before disconnecting during
-  normal object destruction.
+  normal object destruction. Statement-handle cleanup is synchronized with
+  disconnect, including cleanup invoked by cursor finalizers.
 - **GH-769:** Corrected 11 `GetInfoConstants` IDs for scalar functions, outer
   joins, driver handles, cursor attributes, catalog support, and parameter
   descriptions. Added the ODBC name `SQL_TIMEDATE_FUNCTIONS` as an alias of
diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp
index 847c5fc8e..433bec56d 100644
--- a/mssql_python/pybind/connection/connection.cpp
+++ b/mssql_python/pybind/connection/connection.cpp
@@ -113,7 +113,7 @@ void Connection::connect(const py::dict& attrs_before) {
     updateLastUsed();
 }
 
-void Connection::disconnect() {
+void Connection::disconnect(bool rollbackBeforeDisconnect) {
     PERF_TIMER("Connection::disconnect");
     // Determine GIL state once, up front. disconnect() runs both from
     // pybind11-bound methods (GIL held) and from GIL-less destructor / shutdown
@@ -131,62 +131,69 @@ void Connection::disconnect() {
             LOG("Disconnecting from database");
         }
 
-        // Keep wrappers alive while SQLDisconnect frees their native handles.
-        // Otherwise another thread's GC could free a statement again before we
-        // mark it as implicitly freed. On failure, keep the handles usable.
         std::vector<SqlHandlePtr> childHandles;
         size_t originalSize = 0, afterCompactSize = 0, badHandleCount = 0;
-        {
-            std::lock_guard<std::mutex> lock(_childHandlesMutex);
-            
-            // First compact: remove expired weak_ptrs (they're already destroyed)
-            originalSize = _childStatementHandles.size();
-            _childStatementHandles.erase(
-                std::remove_if(_childStatementHandles.begin(), _childStatementHandles.end(),
-                               [](const std::weak_ptr<SqlHandle>& wp) { return wp.expired(); }),
-                _childStatementHandles.end());
-            afterCompactSize = _childStatementHandles.size();
-
-            for (auto& weakHandle : _childStatementHandles) {
-                if (auto handle = weakHandle.lock()) {
-                    // SAFETY ASSERTION: Only STMT handles should be in this vector
-                    // This is guaranteed by allocStatementHandle() which only creates STMT handles
-                    // If this assertion fails, it indicates a serious bug in handle tracking
-                    if (handle->type() != SQL_HANDLE_STMT) {
-                        ++badHandleCount;
-                        continue;  // Skip marking to prevent leak
+        auto disconnectNative = [&]() {
+            // Serialize explicit child free() calls as well as destruction.
+            // This lock must be released before reacquiring the GIL or logging.
+            std::lock_guard<std::mutex> cleanupLock(_cleanupState->mutex);
+            {
+                std::lock_guard<std::mutex> lock(_childHandlesMutex);
+                originalSize = _childStatementHandles.size();
+                _childStatementHandles.erase(
+                    std::remove_if(_childStatementHandles.begin(), _childStatementHandles.end(),
+                                   [](const std::weak_ptr<SqlHandle>& wp) { return wp.expired(); }),
+                    _childStatementHandles.end());
+                afterCompactSize = _childStatementHandles.size();
+                childHandles.reserve(afterCompactSize);
+                for (auto& weakHandle : _childStatementHandles) {
+                    if (auto handle = weakHandle.lock()) {
+                        if (handle->type() != SQL_HANDLE_STMT) {
+                            ++badHandleCount;
+                            continue;
+                        }
+                        childHandles.push_back(std::move(handle));
                     }
-                    childHandles.push_back(std::move(handle));
                 }
             }
-        }
+            if (rollbackBeforeDisconnect) {
+                // Explicit SQL transactions need manual mode for SQLEndTran.
+                // Never turn autocommit on here: that could commit abandoned work.
+                SQLSetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT,
+                                     reinterpret_cast<SQLPOINTER>(SQL_AUTOCOMMIT_OFF), 0);
+                SQLEndTran_ptr(SQL_HANDLE_DBC, _dbcHandle->get(), SQL_ROLLBACK);
+            }
+            SQLRETURN result = SQLDisconnect_ptr(_dbcHandle->get());
+            if (SQL_SUCCEEDED(result)) {
+                // Also cover children whose weak_ptr expired as their destructor
+                // began waiting for this gate: they cannot appear in the snapshot.
+                _cleanupState->disconnected = true;
+                std::lock_guard<std::mutex> lock(_childHandlesMutex);
+                for (const auto& handle : childHandles) {
+                    handle->markImplicitlyFreed();
+                }
+                _childStatementHandles.clear();
+                _allocationsSinceCompaction = 0;
+            }
+            return result;
+        };
 
         SQLRETURN ret;
         if (hasGil) {
             py::gil_scoped_release release;
-            ret = SQLDisconnect_ptr(_dbcHandle->get());
+            ret = disconnectNative();
         } else {
-            ret = SQLDisconnect_ptr(_dbcHandle->get());
+            ret = disconnectNative();
         }
         if (!SQL_SUCCEEDED(ret)) {
             if (hasGil) {
                 checkError(ret);
             } else {
-                std::fprintf(stderr, "mssql-python: native disconnect failed (SQLRETURN %d)\n",
-                             static_cast<int>(ret));
+                std::fputs("mssql-python: native disconnect failed\n", stderr);
             }
             // Keep ownership and child-handle tracking intact for a cleanup retry.
             return;
         }
-        {
-            std::lock_guard<std::mutex> lock(_childHandlesMutex);
-            for (const auto& handle : childHandles) {
-                handle->markImplicitlyFreed();
-            }
-            _childStatementHandles.clear();
-            _allocationsSinceCompaction = 0;
-        }
-
         // Log after releasing _childHandlesMutex (#671): LOG()/LOG_ERROR() acquire
         // the GIL and must not run while a native mutex is held. Also gated on
         // hasGil so the GIL-less destructor / shutdown path never tries to log.
@@ -216,22 +223,13 @@ void Connection::disconnectNoThrow() noexcept {
         if (!_dbcHandle) {
             return;
         }
-        auto cleanup = [this]() {
-            // SQLEndTran only rolls back explicit SQL transactions after entering
-            // manual-commit mode. Never turn autocommit on here: that could commit.
-            SQLSetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT,
-                                 reinterpret_cast<SQLPOINTER>(SQL_AUTOCOMMIT_OFF), 0);
-            SQLEndTran_ptr(SQL_HANDLE_DBC, _dbcHandle->get(), SQL_ROLLBACK);
-            // Attempt disconnect even if rollback failed (e.g. a dead connection).
-            disconnect();
-        };
         // disconnect() already supports GIL-less cleanup. Drop the GIL once so
         // neither its diagnostics nor handle destruction can enter Python.
         if (PyGILState_Check()) {
             py::gil_scoped_release release;
-            cleanup();
+            disconnect(true);
         } else {
-            cleanup();
+            disconnect(true);
         }
     } catch (...) {
         std::fputs("mssql-python: unexpected failure during native connection cleanup\n", stderr);
@@ -348,7 +346,8 @@ SqlHandlePtr Connection::allocStatementHandle() {
     SQLHANDLE stmt = nullptr;
     SQLRETURN ret = SQLAllocHandle_ptr(SQL_HANDLE_STMT, _dbcHandle->get(), &stmt);
     checkError(ret);
-    auto stmtHandle = std::make_shared<SqlHandle>(static_cast<SQLSMALLINT>(SQL_HANDLE_STMT), stmt);
+    auto stmtHandle = std::make_shared<SqlHandle>(static_cast<SQLSMALLINT>(SQL_HANDLE_STMT),
+                                                stmt, _cleanupState);
 
     // THREAD-SAFETY: Lock mutex before modifying _childStatementHandles
     // This protects against concurrent disconnect() or allocStatementHandle() calls,
diff --git a/mssql_python/pybind/connection/connection.h b/mssql_python/pybind/connection/connection.h
index f14f83777..0c66aed1e 100644
--- a/mssql_python/pybind/connection/connection.h
+++ b/mssql_python/pybind/connection/connection.h
@@ -38,7 +38,7 @@ class Connection {
     void connect(const py::dict& attrs_before = py::dict());
 
     // Disconnect and free the connection handle.
-    void disconnect();
+    void disconnect(bool rollbackBeforeDisconnect = false);
 
     // Roll back and disconnect without Python callbacks or escaping exceptions.
     void disconnectNoThrow() noexcept;
@@ -136,6 +136,9 @@ class Connection {
     // Prevents data races between allocStatementHandle() and disconnect(),
     // or concurrent GC finalizers running from different threads
     mutable std::mutex _childHandlesMutex;
+    // Child wrappers retain this gate even after the Connection is destroyed.
+    const std::shared_ptr<ConnectionCleanupState> _cleanupState =
+        std::make_shared<ConnectionCleanupState>();
 };
 
 class ConnectionPool;
diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp
index 5a5991986..3aa61c24b 100644
--- a/mssql_python/pybind/ddbc_bindings.cpp
+++ b/mssql_python/pybind/ddbc_bindings.cpp
@@ -1540,7 +1540,9 @@ void DriverLoader::loadDriver() {
 }
 
 // SqlHandle definition
-SqlHandle::SqlHandle(SQLSMALLINT type, SQLHANDLE rawHandle) : _type(type), _handle(rawHandle) {}
+SqlHandle::SqlHandle(SQLSMALLINT type, SQLHANDLE rawHandle,
+                     std::shared_ptr<ConnectionCleanupState> cleanupState)
+    : _type(type), _handle(rawHandle), _cleanupState(std::move(cleanupState)) {}
 
 SqlHandle::~SqlHandle() {
     if (_handle) {
@@ -1548,6 +1550,13 @@ SqlHandle::~SqlHandle() {
     }
 }
 
+std::unique_lock<std::mutex> SqlHandle::lockForCleanup() const {
+    if (_cleanupState) {
+        return std::unique_lock<std::mutex>(_cleanupState->mutex);
+    }
+    return {};
+}
+
 SQLHANDLE SqlHandle::get() const {
     return _handle;
 }
@@ -1580,82 +1589,71 @@ void SqlHandle::markImplicitlyFreed() {
  * If you need destruction logs, use explicit close() methods instead.
  */
 void SqlHandle::free() {
-    PERF_TIMER("SqlHandle::free");
-    if (_handle && SQLFreeHandle_ptr) {
-        // GH-610: Clear describe cache to prevent memory leak.
-        describeCache.clear();
+    freeHandle();
+}
 
-        // Check if Python is shutting down using centralized helper function
-        bool pythonShuttingDown = is_python_finalizing();
-
-        // RESOURCE LEAK MITIGATION:
-        // When handles are skipped during shutdown, they are not freed, which could
-        // cause resource leaks. However, this is mitigated by:
-        // 1. Python-side atexit cleanup (in __init__.py) that explicitly closes all
-        //    connections before shutdown, ensuring handles are freed in correct order
-        // 2. OS-level cleanup at process termination recovers any remaining resources
-        // 3. This tradeoff prioritizes crash prevention over resource cleanup, which
-        //    is appropriate since we're already in shutdown sequence
-        bool skipDuringShutdown = _type == SQL_HANDLE_STMT || _type == SQL_HANDLE_DBC;
+SQLRETURN SqlHandle::freeHandle() {
+    PERF_TIMER("SqlHandle::free");
+    bool pythonShuttingDown = is_python_finalizing();
+    bool skipDuringShutdown = _type == SQL_HANDLE_STMT || _type == SQL_HANDLE_DBC;
 #ifdef _WIN32
-        // The static ENV is destroyed during DLL_PROCESS_DETACH, after Python
-        // finalization. Calling ODBC then can access already-torn-down SSPI state.
-        skipDuringShutdown = skipDuringShutdown || _type == SQL_HANDLE_ENV;
+    // The static ENV is destroyed during DLL_PROCESS_DETACH, after Python
+    // finalization. Calling ODBC then can access already-torn-down SSPI state.
+    skipDuringShutdown = skipDuringShutdown || _type == SQL_HANDLE_ENV;
 #endif
-        if (pythonShuttingDown && skipDuringShutdown) {
-            _handle = nullptr;  // Mark as freed to prevent double-free attempts
-            return;
-        }
+    if (pythonShuttingDown && skipDuringShutdown) {
+        // Do not wait for another thread's ODBC cleanup during finalization.
+        // Process teardown owns any resources not released by atexit cleanup.
+        _handle = nullptr;
+        return SQL_SUCCESS;
+    }
 
-        // CRITICAL FIX: Check if handle was already implicitly freed by parent handle
-        // When Connection::disconnect() frees the DBC handle, the ODBC driver automatically
-        // frees all child STMT handles. We track this state to avoid double-free attempts.
-        // This approach avoids calling ODBC functions on potentially-freed handles, which
-        // would cause use-after-free errors.
-        if (_implicitly_freed) {
-            _handle = nullptr;  // Just clear the pointer, don't call ODBC functions
-            return;
+    auto freeNative = [this]() -> SQLRETURN {
+        auto cleanupLock = lockForCleanup();
+        if (!_handle || !SQLFreeHandle_ptr) {
+            return SQL_INVALID_HANDLE;
         }
-
-        // Handle is valid and not implicitly freed, proceed with normal freeing.
-        // Release the GIL during the blocking ODBC call (SQLFreeHandle on a STMT
-        // with an open server-side cursor, or on a DBC, performs network I/O).
-        // This is critical when the connection is reached through an in-process
-        // Python TCP forwarder (e.g. paramiko + sshtunnel) - the forwarder
-        // thread needs the GIL to push bytes, so holding it here deadlocks
-        // (issue #565). Only release the GIL if it is actually held AND the
-        // interpreter is not finalizing - gil_scoped_release is unsafe during
-        // shutdown even if PyGILState_Check() reports the GIL as held.
-        if (!pythonShuttingDown && PyGILState_Check()) {
-            py::gil_scoped_release release;
-            SQLFreeHandle_ptr(_type, _handle);
-        } else {
-            SQLFreeHandle_ptr(_type, _handle);
+        describeCache.clear();
+        if (_implicitly_freed || (_cleanupState && _cleanupState->disconnected)) {
+            _handle = nullptr;
+            return SQL_SUCCESS;
         }
-        _handle = nullptr;
+        SQLRETURN ret = SQLFreeHandle_ptr(_type, _handle);
+        if (SQL_SUCCEEDED(ret)) {
+            _handle = nullptr;
+        }
+        return ret;
+    };
+    // The same gate is held through SQLDisconnect and child invalidation.
+    // Release the GIL before waiting, and unlock before reacquiring it.
+    if (!pythonShuttingDown && PyGILState_Check()) {
+        py::gil_scoped_release release;
+        return freeNative();
     }
+    return freeNative();
 }
 
 void SqlHandle::close_cursor() {
-    if (_type != SQL_HANDLE_STMT || !_handle) {
+    if (is_python_finalizing()) {
         return;
     }
-    if (_implicitly_freed) {
-        return;
-    }
-    if (!SQLFreeStmt_ptr) {
-        ThrowStdException("SQLFreeStmt function not loaded");
-    }
-    // Release the GIL during the blocking SQLFreeStmt(SQL_CLOSE) network
-    // round-trip; see issue #565 (in-process forwarder deadlock).
-    // Skip GIL release when the GIL isn't held or the interpreter is
-    // finalizing - gil_scoped_release is unsafe in shutdown.
+    auto closeNative = [this]() -> SQLRETURN {
+        auto cleanupLock = lockForCleanup();
+        if (_type != SQL_HANDLE_STMT || !_handle || _implicitly_freed ||
+            (_cleanupState && _cleanupState->disconnected)) {
+            return SQL_SUCCESS;
+        }
+        if (!SQLFreeStmt_ptr) {
+            ThrowStdException("SQLFreeStmt function not loaded");
+        }
+        return SQLFreeStmt_ptr(_handle, SQL_CLOSE);
+    };
     SQLRETURN ret;
-    if (!is_python_finalizing() && PyGILState_Check()) {
+    if (PyGILState_Check()) {
         py::gil_scoped_release release;
-        ret = SQLFreeStmt_ptr(_handle, SQL_CLOSE);
+        ret = closeNative();
     } else {
-        ret = SQLFreeStmt_ptr(_handle, SQL_CLOSE);
+        ret = closeNative();
     }
     if (ret != SQL_SUCCESS && ret != SQL_SUCCESS_WITH_INFO) {
         ThrowStdException("SQLFreeStmt(SQL_CLOSE) failed");
@@ -1663,44 +1661,25 @@ void SqlHandle::close_cursor() {
 }
 
 void SqlHandle::cancel() {
-    // SQLCancel is intentionally lenient: it is a no-op on non-STMT handles,
-    // already-freed handles, or if the driver does not expose it. This lets
-    // _ArrowReader.close() call it unconditionally without coordinating with
-    // the fetch thread. The GIL is released so a blocked fetch thread can
-    // observe the cancel and return.
-    //
-    // Cross-thread invariant (why no mutex is needed):
-    //   The only cross-thread pattern this driver blesses is exactly the one
-    //   ODBC blesses: cancel() may be called from a thread *other than* the
-    //   fetch thread to unblock an in-flight SQLFetch/SQLExecute on the same
-    //   HSTMT. Per the ODBC spec, SQLCancel (with the SQLGetDiagRec/Field
-    //   family) is the only entry point safe to call across threads on the
-    //   same statement handle. All other operations on a Cursor/SqlHandle
-    //   are single-owner: per DB API 2.0 and the Cursor thread-safety note
-    //   in cursor.py, callers must not share a Cursor for its lifecycle
-    //   operations (execute/fetch/close/free) across threads. Under that
-    //   contract, free() / close_cursor() / SQLFreeHandle can never be in
-    //   flight on this handle concurrently with cancel(), so the read of
-    //   _handle above and the SQLCancel_ptr(h) call below cannot race a
-    //   free() that clears _handle.
-    //
-    //   A std::mutex here would only close the cancel()-vs-free() window;
-    //   it would NOT close the (equally real) free()-vs-fetch window
-    //   without also locking every fetch — which would serialize network
-    //   I/O and defeat the whole point of cross-thread cancel. The right
-    //   place to defend against a misuse (Cursor shared across threads for
-    //   close vs. reader-cancel) is at the Python Cursor layer, not here.
-    if (_type != SQL_HANDLE_STMT || !_handle || _implicitly_freed) {
+    if (is_python_finalizing()) {
         return;
     }
-    if (!SQLCancel_ptr) {
-        return;
-    }
-    SQLHANDLE h = _handle;
+    // Fetch/execute do not take this cleanup gate, so cross-thread cancellation
+    // can still interrupt them. Reader finalizers must not cancel a freed handle.
+    auto cancelNative = [this]() -> SQLRETURN {
+        auto cleanupLock = lockForCleanup();
+        if (_type != SQL_HANDLE_STMT || !_handle || _implicitly_freed || !SQLCancel_ptr ||
+            (_cleanupState && _cleanupState->disconnected)) {
+            return SQL_SUCCESS;
+        }
+        return SQLCancel_ptr(_handle);
+    };
     SQLRETURN ret;
-    {
+    if (PyGILState_Check()) {
         py::gil_scoped_release release;
-        ret = SQLCancel_ptr(h);
+        ret = cancelNative();
+    } else {
+        ret = cancelNative();
     }
     // SQLCancel may return SQL_SUCCESS_WITH_INFO when there was nothing to
     // cancel; that is fine. We only throw on hard failure.
@@ -5962,25 +5941,16 @@ SQLRETURN SQLFreeHandle_wrap(SQLSMALLINT HandleType, SqlHandlePtr Handle) {
     LOG("SQLFreeHandle_wrap: Free SQL handle type=%d", HandleType);
     // Guard against a null/None handle being passed from Python - dereferencing
     // Handle->get() on a null shared_ptr would segfault.
-    if (!Handle || !Handle->get()) {
+    if (!Handle || HandleType != Handle->type()) {
         return SQL_INVALID_HANDLE;
     }
-    if (!SQLAllocHandle_ptr) {
+    if (!SQLFreeHandle_ptr) {
         LOG("SQLFreeHandle_wrap: Function pointer not initialized. Loading the "
             "driver.");
         DriverLoader::getInstance().loadDriver();  // Load the driver
     }
 
-    // Release the GIL during the blocking SQLFreeHandle network round-trip
-    // (see issue #565 - in-process Python TCP forwarder deadlock).
-    // Skip GIL release in shutdown paths where it would crash.
-    SQLRETURN ret;
-    if (!is_python_finalizing() && PyGILState_Check()) {
-        py::gil_scoped_release release;
-        ret = SQLFreeHandle_ptr(HandleType, Handle->get());
-    } else {
-        ret = SQLFreeHandle_ptr(HandleType, Handle->get());
-    }
+    SQLRETURN ret = Handle->freeHandle();
     if (!SQL_SUCCEEDED(ret)) {
         LOG("SQLFreeHandle_wrap: SQLFreeHandle failed with error code - %d", ret);
         return ret;
diff --git a/mssql_python/pybind/ddbc_bindings.h b/mssql_python/pybind/ddbc_bindings.h
index 00f04aa09..a4f11cabe 100644
--- a/mssql_python/pybind/ddbc_bindings.h
+++ b/mssql_python/pybind/ddbc_bindings.h
@@ -280,13 +280,20 @@ struct DescribedParamInfo {
     SQLSMALLINT decimalDigits;
 };
 
+struct ConnectionCleanupState {
+    std::mutex mutex;
+    bool disconnected = false;  // Protected by mutex, shared with every child.
+};
+
 class SqlHandle {
   public:
-    SqlHandle(SQLSMALLINT type, SQLHANDLE rawHandle);
+    SqlHandle(SQLSMALLINT type, SQLHANDLE rawHandle,
+              std::shared_ptr<ConnectionCleanupState> cleanupState = nullptr);
     ~SqlHandle();
     SQLHANDLE get() const;
     SQLSMALLINT type() const;
     void free();
+    SQLRETURN freeHandle();
     void close_cursor();
     // Cancel an in-progress statement (SQLCancel). Safe to call from a
     // thread other than the one running the fetch — this is the *only*
@@ -320,9 +327,12 @@ class SqlHandle {
     void clearDescribeCache() { describeCache.clear(); }
 
   private:
+    // The caller must release the GIL before waiting for native cleanup.
+    std::unique_lock<std::mutex> lockForCleanup() const;
     SQLSMALLINT _type;
     SQLHANDLE _handle;
     bool _implicitly_freed = false;  // Tracks if handle was freed by parent
+    std::shared_ptr<ConnectionCleanupState> _cleanupState;
 };
 using SqlHandlePtr = std::shared_ptr<SqlHandle>;
 
diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py
index 3db792abb..a032b171e 100644
--- a/tests/test_009_pooling.py
+++ b/tests/test_009_pooling.py
@@ -1276,6 +1276,246 @@ def collect_children():
     )
 
 
+@pytest.mark.parametrize("explicit_close", [False, True])
+def test_cursor_cyclic_finalizer_with_concurrent_native_disconnect(conn_str, explicit_close):
+    """Exercise real Cursor.close/free after cyclic GC removes its WeakSet entry.
+
+    The Python finalizer/WeakSet ordering is coordinated; overlap inside the
+    native cleanup calls is stress coverage, not a deterministic race trigger.
+    """
+    _run_in_subprocess(
+        f"explicit_close = {explicit_close!r}\n" + textwrap.dedent("""
+            import gc
+            import os
+            import threading
+            import weakref
+
+            import mssql_python
+            from mssql_python import connect, ddbc_bindings as ddbc, pooling
+
+            iterations = 50
+            collect_barrier = threading.Barrier(2, timeout=10)
+            cleanup_barrier = threading.Barrier(2, timeout=10)
+            free_entered = threading.Event()
+            errors = []
+
+            class FinalizerStatement:
+                # Only coordinate entry: Cursor.__del__/close and native free
+                # still run their real implementations, with a real SQL handle.
+                def __init__(self, statement):
+                    self.statement = statement
+                    self.calls = 0
+                    self.completed = False
+
+                def free(self):
+                    self.calls += 1
+                    free_entered.set()
+                    try:
+                        cleanup_barrier.wait()
+                        assert self.statement.free() is None
+                        self.completed = True
+                    except Exception as exc:
+                        errors.append(f"Cursor finalizer: {exc!r}")
+                        raise
+
+            def collect_children():
+                try:
+                    for _ in range(iterations):
+                        collect_barrier.wait()
+                        gc.collect()
+                        collect_barrier.wait()
+                except Exception as exc:
+                    errors.append(f"GC worker: {exc!r}")
+                    collect_barrier.abort()
+                    cleanup_barrier.abort()
+                    free_entered.set()
+
+            pooling(enabled=False)
+            gc.disable()
+            collector = threading.Thread(target=collect_children, daemon=True)
+            collector.start()
+            connection = None
+            native = None
+            try:
+                for _ in range(iterations):
+                    free_entered.clear()
+                    connection = connect(os.environ["DB_CONNECTION_STRING"], autocommit=True)
+                    cursor = connection.cursor()
+                    assert cursor.execute("SELECT 1").fetchall()[0][0] == 1
+                    finalizer_statement = FinalizerStatement(cursor.hstmt)
+                    cursor.hstmt = finalizer_statement
+                    cursor.cycle = cursor
+                    cursor_ref = weakref.ref(cursor)
+                    del cursor
+
+                    collect_barrier.wait()
+                    assert free_entered.wait(10), "Cursor finalizer did not enter free"
+                    assert not errors, errors
+                    assert cursor_ref() is None, "GC did not clear the cursor weakref"
+                    assert not connection._cursors, "Connection.close would still see the cursor"
+
+                    if not explicit_close:
+                        # The cursor retains its Python connection. Detach only
+                        # the native owner to exercise its destructor fallback.
+                        native = connection._conn
+                        connection._conn = None
+                        connection._closed = True
+                        mssql_python._active_connections.discard(connection)
+
+                    cleanup_barrier.wait()
+                    if explicit_close:
+                        connection.close()
+                    else:
+                        native = None
+                    collect_barrier.wait()
+
+                    assert finalizer_statement.calls == 1
+                    assert finalizer_statement.completed, errors
+                    assert not errors, errors
+                    assert finalizer_statement.statement.free() is None
+                    assert ddbc.DDBCSQLFreeHandle(3, finalizer_statement.statement) == -2
+                    assert connection.closed
+                    connection = None
+                collector.join(timeout=10)
+                assert not collector.is_alive(), "GC worker did not exit"
+            finally:
+                collect_barrier.abort()
+                cleanup_barrier.abort()
+                collector.join(timeout=10)
+                if connection is not None:
+                    connection.close()
+                native = None
+                gc.enable()
+            assert not collector.is_alive(), "GC worker did not exit"
+            assert not errors, errors
+            gc.collect()
+            """),
+        conn_str,
+    )
+
+
+def test_failed_native_disconnect_preserves_child_statement(conn_str):
+    """SQLSTATE 25000 must not irreversibly invalidate a live child handle."""
+    _run_in_subprocess(
+        """
+        import os
+        import uuid
+
+        from mssql_python import connect, ddbc_bindings as ddbc, pooling
+
+        pooling(enabled=False)
+        conn_str = os.environ["DB_CONNECTION_STRING"]
+        table = "pytest_disconnect_failure_" + uuid.uuid4().hex
+        observer = connect(conn_str, autocommit=True)
+        observer_cursor = observer.cursor()
+        native = None
+        statement = None
+        created = False
+        try:
+            observer_cursor.execute("SET LOCK_TIMEOUT 1000")
+            observer_cursor.execute(f"CREATE TABLE {table} (id INT)")
+            created = True
+            native = ddbc.Connection(conn_str, False)
+            native.set_autocommit(False)
+            statement = native.alloc_statement_handle()
+            assert ddbc.DDBCSQLExecDirect(statement, f"INSERT INTO {table} VALUES (1)") in (0, 1)
+
+            try:
+                native.close()
+            except RuntimeError as exc:
+                assert "25000" in str(exc), f"Unexpected disconnect failure: {exc}"
+            else:
+                raise AssertionError("Native disconnect accepted an uncommitted INSERT")
+
+            assert ddbc.DDBCSQLExecDirect(
+                statement, f"SELECT COUNT(*), @@TRANCOUNT FROM {table}"
+            ) in (0, 1)
+            row = []
+            assert ddbc.DDBCSQLFetchOne(statement, row) in (0, 1)
+            assert row[0] == 1 and row[1] > 0, row
+            statement._close_cursor()
+            native.rollback()
+
+            assert ddbc.DDBCSQLExecDirect(statement, f"SELECT COUNT(*) FROM {table}") in (0, 1)
+            row = []
+            assert ddbc.DDBCSQLFetchOne(statement, row) in (0, 1)
+            assert row == [0], "Failed disconnect committed the pending INSERT"
+            statement._close_cursor()
+            native.rollback()
+            native.close()
+            native = None
+
+            # Disconnect already freed the ODBC statement. The raw entry point
+            # must consume the wrapper's implicit-free state, not the stale pointer.
+            assert ddbc.DDBCSQLFreeHandle(3, statement) in (0, 1)
+            assert ddbc.DDBCSQLFreeHandle(3, statement) == -2
+            assert statement.free() is None
+            assert statement.free() is None
+            observer_cursor.execute(f"SELECT COUNT(*) FROM {table} WITH (READCOMMITTEDLOCK)")
+            assert observer_cursor.fetchone()[0] == 0
+        finally:
+            try:
+                if native is not None:
+                    try:
+                        native.rollback()
+                    finally:
+                        native.close()
+                if statement is not None:
+                    statement.free()
+            finally:
+                try:
+                    if created:
+                        observer_cursor.execute(f"DROP TABLE {table}")
+                finally:
+                    observer_cursor.close()
+                    observer.close()
+        """,
+        conn_str,
+    )
+
+
+@pytest.mark.parametrize("free_api", ["method", "raw"])
+def test_native_statement_free_entrypoints_are_idempotent(conn_str, free_api):
+    """Raw SQLRETURN and public None-returning free share one ownership state."""
+    _run_in_subprocess(
+        f"free_api = {free_api!r}\n" + textwrap.dedent("""
+            import os
+
+            from mssql_python import ddbc_bindings as ddbc
+
+            native = ddbc.Connection(os.environ["DB_CONNECTION_STRING"], False)
+            native.set_autocommit(True)
+            statement = native.alloc_statement_handle()
+            sibling = native.alloc_statement_handle()
+            try:
+                assert ddbc.DDBCSQLExecDirect(statement, "SELECT 1") in (0, 1)
+                if free_api == "raw":
+                    assert ddbc.DDBCSQLFreeHandle(3, statement) in (0, 1)
+                else:
+                    assert statement.free() is None
+                assert ddbc.DDBCSQLFreeHandle(3, statement) == -2
+                assert statement.free() is None
+                assert statement.free() is None
+
+                assert ddbc.DDBCSQLExecDirect(sibling, "SELECT 42") in (0, 1)
+                row = []
+                assert ddbc.DDBCSQLFetchOne(sibling, row) in (0, 1)
+                assert row == [42]
+                native.close()
+                native = None
+                assert ddbc.DDBCSQLFreeHandle(3, sibling) in (0, 1)
+                assert ddbc.DDBCSQLFreeHandle(3, sibling) == -2
+                assert sibling.free() is None
+            finally:
+                statement.free()
+                sibling.free()
+                if native is not None:
+                    native.close()
+            """),
+        conn_str,
+    )
+
+
 def test_pool_recovery_after_failed_connection(conn_str):
     """Test that the pool recovers after a failed connection attempt."""
     pooling(max_size=1, idle_timeout=30)

From cb46560c953916b4248028225d13a944d8219a05 Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Thu, 17 Sep 2026 09:55:23 +0100
Subject: [PATCH 10/11] FIX: serialize statement allocation with disconnect

Keep native connection ownership during allocation, register statements under the shared cleanup gate, and snapshot allocation diagnostics without Python callbacks. Cover allocation racing disconnect and allocation after failed disconnect.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 CHANGELOG.md                                  |  4 +-
 mssql_python/pybind/connection/connection.cpp | 39 ++++++----
 mssql_python/pybind/ddbc_bindings.cpp         | 17 +++--
 mssql_python/pybind/ddbc_bindings.h           |  2 +
 tests/test_009_pooling.py                     | 74 +++++++++++++++++++
 5 files changed, 112 insertions(+), 24 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 511edefa7..f28db6ce3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -98,8 +98,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
   mode before being parked. This prevents an empty transaction from remaining
   visible on an idle SQL Server session after `Connection.close()`. Abandoned
   native connections also roll back pending work before disconnecting during
-  normal object destruction. Statement-handle cleanup is synchronized with
-  disconnect, including cleanup invoked by cursor finalizers.
+  normal object destruction. Statement-handle allocation and cleanup are
+  synchronized with disconnect, including cleanup invoked by cursor finalizers.
 - **GH-769:** Corrected 11 `GetInfoConstants` IDs for scalar functions, outer
   joins, driver handles, cursor attributes, catalog support, and parameter
   descriptions. Added the ODBC name `SQL_TIMEDATE_FUNCTIONS` as an alias of
diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp
index 433bec56d..6960942fa 100644
--- a/mssql_python/pybind/connection/connection.cpp
+++ b/mssql_python/pybind/connection/connection.cpp
@@ -338,23 +338,30 @@ bool Connection::getAutocommit() const {
 
 SqlHandlePtr Connection::allocStatementHandle() {
     PERF_TIMER("Connection::allocStatementHandle");
-    if (!_dbcHandle) {
-        ThrowStdException("Connection handle not allocated");
-    }
-    updateLastUsed();
     LOG("Allocating statement handle");
-    SQLHANDLE stmt = nullptr;
-    SQLRETURN ret = SQLAllocHandle_ptr(SQL_HANDLE_STMT, _dbcHandle->get(), &stmt);
-    checkError(ret);
-    auto stmtHandle = std::make_shared<SqlHandle>(static_cast<SQLSMALLINT>(SQL_HANDLE_STMT),
-                                                stmt, _cleanupState);
-
-    // THREAD-SAFETY: Lock mutex before modifying _childStatementHandles
-    // This protects against concurrent disconnect() or allocStatementHandle() calls,
-    // or GC finalizers running from different threads
+    // Keep the wrapper outside the lock scope: unwinding a failed registration
+    // frees the statement through the same cleanup gate.
+    SqlHandlePtr stmtHandle;
     bool compacted = false;
     size_t compactBefore = 0, compactAfter = 0;
     {
+        py::gil_scoped_release release;
+        std::lock_guard<std::mutex> cleanupLock(_cleanupState->mutex);
+        if (_cleanupState->disconnected || !_dbcHandle) {
+            ThrowStdException("Connection handle not allocated");
+        }
+        updateLastUsed();
+        SQLHANDLE stmt = nullptr;
+        SQLRETURN ret = SQLAllocHandle_ptr(SQL_HANDLE_STMT, _dbcHandle->get(), &stmt);
+        if (!SQL_SUCCEEDED(ret)) {
+            // Snapshot diagnostics before disconnect can overwrite/free the DBC.
+            ErrorInfo err = SQLReadError(SQL_HANDLE_DBC, _dbcHandle->get(), ret);
+            ThrowStdException(err.sqlState.length() == 5
+                                  ? "SQLSTATE:" + err.sqlState + ":" + err.ddbcErrorMsg
+                                  : err.ddbcErrorMsg);
+        }
+        stmtHandle = std::make_shared<SqlHandle>(static_cast<SQLSMALLINT>(SQL_HANDLE_STMT),
+                                                stmt, _cleanupState);
         std::lock_guard<std::mutex> lock(_childHandlesMutex);
         
         // Track this child handle so we can mark it as implicitly freed when connection closes
@@ -809,10 +816,12 @@ bool ConnectionHandle::getAutocommit() const {
 
 SqlHandlePtr ConnectionHandle::allocStatementHandle() {
     PERF_TIMER("ConnectionHandle::allocStatementHandle");
-    if (!_conn) {
+    // close() can detach _conn while allocation waits without the GIL.
+    auto conn = _conn;
+    if (!conn) {
         ThrowStdException("Connection object is not initialized");
     }
-    return _conn->allocStatementHandle();
+    return conn->allocStatementHandle();
 }
 
 py::object Connection::getInfo(SQLUSMALLINT infoType) const {
diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp
index 3aa61c24b..5f7c7b302 100644
--- a/mssql_python/pybind/ddbc_bindings.cpp
+++ b/mssql_python/pybind/ddbc_bindings.cpp
@@ -1854,19 +1854,22 @@ SQLRETURN SQLColumns_wrap(SqlHandlePtr StatementHandle, const py::object& catalo
 ErrorInfo SQLCheckError_Wrap(SQLSMALLINT handleType, SqlHandlePtr handle, SQLRETURN retcode) {
     PERF_TIMER("SQLCheckError_Wrap");
     LOG("SQLCheckError: Checking ODBC errors - handleType=%d, retcode=%d", handleType, retcode);
+    if (retcode != SQL_INVALID_HANDLE && !SQL_SUCCEEDED(retcode) && !SQLGetDiagRec_ptr) {
+        LOG("SQLCheckError: SQLGetDiagRec function pointer not initialized, loading driver");
+        DriverLoader::getInstance().loadDriver();
+    }
+    return SQLReadError(handleType, handle ? handle->get() : nullptr, retcode);
+}
+
+ErrorInfo SQLReadError(SQLSMALLINT handleType, SQLHANDLE rawHandle, SQLRETURN retcode) {
     ErrorInfo errorInfo;
-    if (retcode == SQL_INVALID_HANDLE) {
-        LOG("SQLCheckError: SQL_INVALID_HANDLE detected - handle is invalid");
+    if (retcode == SQL_INVALID_HANDLE || !rawHandle) {
         errorInfo.ddbcErrorMsg = "Invalid handle!";
         return errorInfo;
     }
-    assert(handle != 0);
-    SQLHANDLE rawHandle = handle->get();
     if (!SQL_SUCCEEDED(retcode)) {
         if (!SQLGetDiagRec_ptr) {
-            LOG("SQLCheckError: SQLGetDiagRec function pointer not "
-                "initialized, loading driver");
-            DriverLoader::getInstance().loadDriver();  // Load the driver
+            ThrowStdException("SQLGetDiagRec function pointer not initialized");
         }
 
         SQLWCHAR sqlState[6], message[SQL_MAX_MESSAGE_LENGTH_SQLSERVER];
diff --git a/mssql_python/pybind/ddbc_bindings.h b/mssql_python/pybind/ddbc_bindings.h
index a4f11cabe..ec5c9a11f 100644
--- a/mssql_python/pybind/ddbc_bindings.h
+++ b/mssql_python/pybind/ddbc_bindings.h
@@ -343,6 +343,8 @@ struct ErrorInfo {
     std::string ddbcErrorMsg;
 };
 ErrorInfo SQLCheckError_Wrap(SQLSMALLINT handleType, SqlHandlePtr handle, SQLRETURN retcode);
+// Driver must be initialized; reads diagnostics without Python logging/callbacks.
+ErrorInfo SQLReadError(SQLSMALLINT handleType, SQLHANDLE handle, SQLRETURN retcode);
 
 // Thread-safe decimal separator accessor class
 class ThreadSafeDecimalSeparator {
diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py
index a032b171e..ef3ce1b04 100644
--- a/tests/test_009_pooling.py
+++ b/tests/test_009_pooling.py
@@ -1427,6 +1427,15 @@ def test_failed_native_disconnect_preserves_child_statement(conn_str):
             else:
                 raise AssertionError("Native disconnect accepted an uncommitted INSERT")
 
+            extra_statement = native.alloc_statement_handle()
+            try:
+                assert ddbc.DDBCSQLExecDirect(extra_statement, "SELECT 42") in (0, 1)
+                extra_row = []
+                assert ddbc.DDBCSQLFetchOne(extra_statement, extra_row) in (0, 1)
+                assert extra_row == [42]
+            finally:
+                extra_statement.free()
+
             assert ddbc.DDBCSQLExecDirect(
                 statement, f"SELECT COUNT(*), @@TRANCOUNT FROM {table}"
             ) in (0, 1)
@@ -1516,6 +1525,71 @@ def test_native_statement_free_entrypoints_are_idempotent(conn_str, free_api):
     )
 
 
+def test_native_statement_allocation_racing_disconnect(conn_str):
+    """Allocation either registers before disconnect or rejects its closed state."""
+    _run_in_subprocess(
+        """
+        import os
+        import threading
+
+        from mssql_python import ddbc_bindings as ddbc
+
+        barrier = threading.Barrier(2, timeout=10)
+        errors = []
+        statements = []
+        iterations = 100
+        native = None
+
+        def allocate():
+            try:
+                for _ in range(iterations):
+                    barrier.wait()
+                    try:
+                        statements.append(native.alloc_statement_handle())
+                    except RuntimeError as exc:
+                        assert str(exc) in (
+                            "Connection object is not initialized",
+                            "Connection handle not allocated",
+                        ), str(exc)
+                    barrier.wait()
+            except Exception as exc:
+                errors.append(repr(exc))
+                barrier.abort()
+
+        worker = threading.Thread(target=allocate, daemon=True)
+        worker.start()
+        try:
+            for _ in range(iterations):
+                native = ddbc.Connection(os.environ["DB_CONNECTION_STRING"], False)
+                native.set_autocommit(True)
+                barrier.wait()
+                native.close()
+                barrier.wait()
+                assert not errors, errors
+                for statement in statements:
+                    assert ddbc.DDBCSQLFreeHandle(3, statement) in (0, 1)
+                    assert statement.free() is None
+                statements.clear()
+                try:
+                    native.alloc_statement_handle()
+                except RuntimeError as exc:
+                    assert "Connection object is not initialized" in str(exc)
+                else:
+                    raise AssertionError("Allocation succeeded after native close")
+        finally:
+            worker.join(timeout=10)
+            if worker.is_alive():
+                barrier.abort()
+                worker.join(timeout=10)
+            for statement in statements:
+                statement.free()
+        assert not worker.is_alive(), "Allocation worker did not exit"
+        assert not errors, errors
+        """,
+        conn_str,
+    )
+
+
 def test_pool_recovery_after_failed_connection(conn_str):
     """Test that the pool recovers after a failed connection attempt."""
     pooling(max_size=1, idle_timeout=30)

From f496090d36b41da2b973c5cd41e7ad4da78b932e Mon Sep 17 00:00:00 2001
From: Sumit Sarabhai <sumitsar@microsoft.com>
Date: Thu, 17 Sep 2026 10:40:37 +0100
Subject: [PATCH 11/11] TEST: tighten KILL permission skip to exact diagnostic

Match the SQL Server 6102 permission-denied phrase in test_pool_removes_invalid_connections so an unexpected KILL failure fails the test instead of being masked as a skip, consistent with the newer sanitation test.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
---
 tests/test_009_pooling.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/test_009_pooling.py b/tests/test_009_pooling.py
index ef3ce1b04..c8cafed8f 100644
--- a/tests/test_009_pooling.py
+++ b/tests/test_009_pooling.py
@@ -912,7 +912,7 @@ def session_identity(conn):
             admin.cursor().execute(f"KILL {victim_spid}")
         except Exception as e:
             msg = str(e)
-            if "permission" in msg.lower() or "KILL" in msg:
+            if "does not have permission to use the kill statement" in msg.lower():
                 import sys as _sys
                 print(
                     f"Skipping: KILL not permitted for this login: {msg}",