diff --git a/CHANGELOG.md b/CHANGELOG.md index c4a2c48df..98db245e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -94,6 +94,12 @@ 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()`. Abandoned + native connections also roll back pending work before disconnecting during + normal object destruction. Statement-handle allocation and cleanup are + synchronized with disconnect, including cleanup invoked by cursor finalizers. - Bounded text fetched as UTF-16 now preserves leading U+FEFF and U+FFFE as payload rather than treating them as byte-order markers. This corrects row-wise `fetchone()`, `fetchmany()`, and `fetchall()` results, including diff --git a/mssql_python/connection.py b/mssql_python/connection.py index 21f564add..59496cb1f 100644 --- a/mssql_python/connection.py +++ b/mssql_python/connection.py @@ -2179,21 +2179,35 @@ def close(self) -> None: # Close the connection even if cursor cleanup had issues try: if self._conn: - if not self.autocommit: - # 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 + autocommit_error = None + rollback_error = None + 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. 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) + rollback_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(manual_commit and rollback_error is None) + except RuntimeError as e: + _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) + 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 8ccee11a7..6960942fa 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 +#include #include #include #include @@ -18,6 +19,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"); @@ -52,8 +64,8 @@ Connection::Connection(const std::u16string& conn_str, bool use_pool) allocateDbcHandle(); } -Connection::~Connection() { - disconnect(); // fallback if user forgets to disconnect +Connection::~Connection() noexcept { + disconnectNoThrow(); } // Allocates connection handle @@ -101,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 @@ -113,47 +125,75 @@ 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"); } - // 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 + std::vector childHandles; size_t originalSize = 0, afterCompactSize = 0, badHandleCount = 0; - { - std::lock_guard 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& 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 cleanupLock(_cleanupState->mutex); + { + std::lock_guard lock(_childHandlesMutex); + originalSize = _childStatementHandles.size(); + _childStatementHandles.erase( + std::remove_if(_childStatementHandles.begin(), _childStatementHandles.end(), + [](const std::weak_ptr& 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)); } + } + } + 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(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 lock(_childHandlesMutex); + for (const auto& handle : childHandles) { handle->markImplicitlyFreed(); } + _childStatementHandles.clear(); + _allocationsSinceCompaction = 0; } - _childStatementHandles.clear(); - _allocationsSinceCompaction = 0; - } // Release lock before potentially slow SQLDisconnect call + return result; + }; + SQLRETURN ret; + if (hasGil) { + py::gil_scoped_release release; + ret = disconnectNative(); + } else { + ret = disconnectNative(); + } + if (!SQL_SUCCEEDED(ret)) { + if (hasGil) { + checkError(ret); + } else { + std::fputs("mssql-python: native disconnect failed\n", stderr); + } + // Keep ownership and child-handle tracking intact for a cleanup retry. + return; + } // 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. @@ -167,26 +207,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) { @@ -194,6 +214,39 @@ void Connection::disconnect() { } } +void Connection::disconnectNoThrow() noexcept { + try { + if (isPythonFinalizing()) { + abandonDuringFinalization(); + return; + } + if (!_dbcHandle) { + return; + } + // 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; + disconnect(true); + } else { + disconnect(true); + } + } catch (...) { + std::fputs("mssql-python: unexpected failure during native connection cleanup\n", stderr); + } +} + +void Connection::abandonDuringFinalization() noexcept { + { + std::lock_guard 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 { @@ -285,22 +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(static_cast(SQL_HANDLE_STMT), stmt); - - // 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 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(static_cast(SQL_HANDLE_STMT), + stmt, _cleanupState); std::lock_guard lock(_childHandlesMutex); // Track this child handle so we can mark it as implicitly freed when connection closes @@ -564,6 +625,26 @@ bool Connection::reset() { return true; } +void Connection::prepareForPool(bool transactionAlreadyRolledBack) { + if (!_dbcHandle) { + ThrowStdException("Connection handle not allocated"); + } + + // 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); + } + 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. + setAutocommit(true); +} + void Connection::updateLastUsed() { _lastUsed = std::chrono::steady_clock::now(); } @@ -631,7 +712,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 @@ -659,17 +741,42 @@ ConnectionHandle::ConnectionHandle(const std::u16string& connStr, bool usePool, ConnectionHandle::~ConnectionHandle() { if (_conn) { - close(); + if (isPythonFinalizing()) { + _conn->abandonDuringFinalization(); + _conn = nullptr; + return; + } + try { + // Discard ends abandoned work without returning this connection to + // the pool or entering Python from a native destructor. + ConnectionPoolManager::getInstance().discardConnection(_originPool, _conn); + } catch (...) { + std::fputs("mssql-python: failed to release native connection pool capacity\n", stderr); + _conn->disconnectNoThrow(); + } } } -void ConnectionHandle::close() { +void ConnectionHandle::close(bool transactionAlreadyRolledBack) { PERF_TIMER("ConnectionHandle::close"); if (!_conn) { ThrowStdException("Connection object is not initialized"); } if (_usePool) { - ConnectionPoolManager::getInstance().returnConnection(_poolKey, _conn); + try { + _conn->prepareForPool(transactionAlreadyRolledBack); + } 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(_originPool, _conn); + } catch (...) { + } + _conn = nullptr; + throw; + } + ConnectionPoolManager::getInstance().returnConnection(_poolKey, _originPool, _conn); } else { _conn->disconnect(); } @@ -709,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/connection/connection.h b/mssql_python/pybind/connection/connection.h index d5300aff4..0c66aed1e 100644 --- a/mssql_python/pybind/connection/connection.h +++ b/mssql_python/pybind/connection/connection.h @@ -32,13 +32,19 @@ 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()); // 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; + + // Relinquish native handles without ODBC calls during interpreter finalization. + void abandonDuringFinalization() noexcept; // Commit the current transaction. void commit(); @@ -53,6 +59,7 @@ class Connection { bool getAutocommit() const; bool isAlive() const; bool reset(); + void prepareForPool(bool transactionAlreadyRolledBack = false); void updateLastUsed(); std::chrono::steady_clock::time_point lastUsed() const; @@ -129,8 +136,13 @@ 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 _cleanupState = + std::make_shared(); }; +class ConnectionPool; + class ConnectionHandle { public: ConnectionHandle(const std::u16string& connStr, bool usePool, @@ -139,7 +151,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); @@ -159,4 +171,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 _originPool; }; diff --git a/mssql_python/pybind/connection/connection_pool.cpp b/mssql_python/pybind/connection/connection_pool.cpp index 831a01db2..3ea3ccc5f 100644 --- a/mssql_python/pybind/connection/connection_pool.cpp +++ b/mssql_python/pybind/connection/connection_pool.cpp @@ -329,6 +329,15 @@ void ConnectionPool::release(std::shared_ptr conn) { } } +void ConnectionPool::discard(std::shared_ptr conn) { + { + std::lock_guard lock(_mutex); + if (_current_size > 0) + --_current_size; + } + conn->disconnectNoThrow(); +} + bool ConnectionPool::canEvict() { std::lock_guard lock(_mutex); // Never evict while any connection is checked out or in-flight. Reserved @@ -388,7 +397,9 @@ ConnectionPoolManager& ConnectionPoolManager::getInstance() { std::shared_ptr 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* + 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). @@ -450,6 +461,9 @@ std::shared_ptr 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 @@ -474,18 +488,18 @@ std::shared_ptr ConnectionPoolManager::acquireConnection(const std:: return pool->acquire(connStr, attrs_before, token_factory); } -void ConnectionPoolManager::returnConnection(const std::u16string& pool_key, - const std::shared_ptr conn) { - std::shared_ptr pool; +void ConnectionPoolManager::returnConnection( + const std::u16string& pool_key, const std::weak_ptr& originating_pool, + const std::shared_ptr conn) { + std::shared_ptr pool = originating_pool.lock(); + bool registered = false; { std::lock_guard 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 @@ -506,6 +520,20 @@ void ConnectionPoolManager::returnConnection(const std::u16string& pool_key, } } +void ConnectionPoolManager::discardConnection( + const std::weak_ptr& originating_pool, + const std::shared_ptr conn) { + if (!conn) { + return; + } + std::shared_ptr pool = originating_pool.lock(); + if (pool) { + pool->discard(conn); + } else { + conn->disconnectNoThrow(); + } +} + void ConnectionPoolManager::configure(int max_size, int idle_timeout_secs) { std::lock_guard 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..6f67eab56 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 conn); + // Permanently removes a checked-out connection and releases its capacity. + void discard(std::shared_ptr conn); + // Closes all connections in the pool, releasing resources void close(); @@ -71,7 +74,8 @@ class ConnectionPoolManager { std::shared_ptr 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* originating_pool = nullptr); // Arms (true) or disarms (false) new-pool creation. Disarming, done under // _manager_mutex, guarantees that any acquireConnection() serialized after @@ -80,7 +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 conn); + void returnConnection(const std::u16string& pool_key, + const std::weak_ptr& originating_pool, + std::shared_ptr conn); + + // Discards a connection from the exact pool generation that issued it. + void discardConnection(const std::weak_ptr& originating_pool, + std::shared_ptr conn); // Closes all pools and their connections void closePools(); diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index caaca1213..1f235f829 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -1541,7 +1541,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 cleanupState) + : _type(type), _handle(rawHandle), _cleanupState(std::move(cleanupState)) {} SqlHandle::~SqlHandle() { if (_handle) { @@ -1549,6 +1551,13 @@ SqlHandle::~SqlHandle() { } } +std::unique_lock SqlHandle::lockForCleanup() const { + if (_cleanupState) { + return std::unique_lock(_cleanupState->mutex); + } + return {}; +} + SQLHANDLE SqlHandle::get() const { return _handle; } @@ -1581,82 +1590,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"); @@ -1664,44 +1662,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) { - return; - } - if (!SQLCancel_ptr) { + if (is_python_finalizing()) { 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. @@ -1876,19 +1855,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]; @@ -5965,25 +5947,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; @@ -6103,7 +6076,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/mssql_python/pybind/ddbc_bindings.h b/mssql_python/pybind/ddbc_bindings.h index 59f26f120..11c33d8d2 100644 --- a/mssql_python/pybind/ddbc_bindings.h +++ b/mssql_python/pybind/ddbc_bindings.h @@ -281,13 +281,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 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* @@ -321,9 +328,12 @@ class SqlHandle { void clearDescribeCache() { describeCache.clear(); } private: + // The caller must release the GIL before waiting for native cleanup. + std::unique_lock lockForCleanup() const; SQLSMALLINT _type; SQLHANDLE _handle; bool _implicitly_freed = false; // Tracks if handle was freed by parent + std::shared_ptr _cleanupState; }; using SqlHandlePtr = std::shared_ptr; @@ -334,6 +344,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_006_exceptions.py b/tests/test_006_exceptions.py index b0d117761..371b9662f 100644 --- a/tests/test_006_exceptions.py +++ b/tests/test_006_exceptions.py @@ -276,6 +276,104 @@ def test_connect_runtime_error_mapped_to_correct_dbapi_exception(): assert not isinstance(exc_info.value, RuntimeError) +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 + + 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;") + + with pytest.raises(OperationalError, match="Communication link failure"): + conn.close() + + mock_conn.rollback.assert_called_once_with() + mock_conn.close.assert_called_once_with(False) + assert conn._conn is None + 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(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_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 e33f71030..c8cafed8f 100644 --- a/tests/test_009_pooling.py +++ b/tests/test_009_pooling.py @@ -127,6 +127,187 @@ def test_connection_pooling_reuse_spid(conn_str): assert spid1 == spid2, "Connections not reused - different SPIDs" +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 + import sys + + import mssql_python + + conn_str = os.environ["DB_CONNECTION_STRING"] + mssql_python.pooling(enabled=True, max_size=2, idle_timeout=30) + observer = mssql_python.connect(conn_str, autocommit=True) + 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"), + ("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: + 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}" + ) + + if open_transaction_count(subject_spid) is None: + 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() + + 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 " + f"open_transaction_count={row[0]}" + ) + + observer_cursor.close() + finally: + observer.close() + mssql_python.pooling(enabled=False) + """, + conn_str, + ) + + +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] + 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) + + 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}") + 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. @@ -708,9 +889,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 @@ -731,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}", @@ -751,8 +932,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 @@ -778,6 +963,633 @@ 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 + 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) + + 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 "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, + ) + victim.close() + admin.close() + 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() + + 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_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, + ) + + +@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, + ) + + +@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") + + 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) + 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_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)