diff --git a/CHANGELOG.md b/CHANGELOG.md index 2cac01d4d..e374d8f23 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,6 +57,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), does not change the default provider or ship any Rust driver binaries. ### Changed +- Pooled check-in skips transaction sanitation only after a prior successful + native rollback/autocommit restore and no subsequent statement allocation or + uncertain operation. Used connections still roll back explicit transactions + before parking; sanitation uses one native attribute probe and one GIL release. + Raw-handle exposure and arbitrary connection attributes disable the fast path; + successfully applied scalar login timeouts (including `connect(timeout=30)`) + do not permanently disable it. - Fetches reuse owned native metadata for stable columns within a result set; `fetchmany()` avoids the Python metadata-dictionary roundtrip. Re-execution, result transitions and statement/connection cleanup invalidate this metadata. diff --git a/eng/pipelines/pr-validation-pipeline.yml b/eng/pipelines/pr-validation-pipeline.yml index c19a3e8aa..90727218f 100644 --- a/eng/pipelines/pr-validation-pipeline.yml +++ b/eng/pipelines/pr-validation-pipeline.yml @@ -11,6 +11,41 @@ variables: arm64QemuImage: 'tonistiigi/binfmt@sha256:400a4873b838d1b89194d982c45e5fb3cda4593fbfd7e08a02e76b03b21166f0' jobs: +- job: NativePoolSanitation + displayName: 'Native pool sanitation tests' + timeoutInMinutes: 20 + pool: + vmImage: 'ubuntu-latest' + + steps: + - checkout: self + persistCredentials: false + + - task: UsePythonVersion@0 + inputs: + versionSpec: '3.13' + addToPath: true + displayName: 'Use Python 3.13 for native tests' + + - bash: | + set -euo pipefail + sudo apt-get update + sudo apt-get install -y build-essential cmake unixodbc-dev + python -m pip install --no-deps --require-hashes -r eng/requirements-build-linux.txt + displayName: 'Install native test build dependencies' + + - bash: | + set -euo pipefail + cmake -S mssql_python/pybind -B mssql_python/pybind/build/native-tests \ + -DCMAKE_BUILD_TYPE=Release -DENABLE_PROFILING=OFF \ + -DBUILD_NATIVE_POOL_TESTS=ON -DPython3_EXECUTABLE="$(command -v python)" + cmake --build mssql_python/pybind/build/native-tests \ + --config Release --target pool_sanitation_tests --parallel 2 + ctest --test-dir mssql_python/pybind/build/native-tests \ + -C Release --output-on-failure --no-tests=error \ + --timeout 120 -R '^pool_sanitation$' + displayName: 'Build and run native pool sanitation tests' + - job: CodeQLAnalysis displayName: 'CodeQL Security Analysis' pool: diff --git a/mssql_python/connection.py b/mssql_python/connection.py index 4d0b4cd72..0f5ea21e1 100644 --- a/mssql_python/connection.py +++ b/mssql_python/connection.py @@ -2223,35 +2223,14 @@ def close(self) -> None: # Close the connection even if cursor cleanup had issues try: if self._conn: - autocommit_error = None - rollback_error = None - manual_commit = False + # Native close owns transaction cleanup and fail-closed discard. + # Avoid duplicate attribute probes and untrusted rollback hints. 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: - rollback_error = e - # TODO: Check potential race conditions in case of multithreaded scenarios - # Close the connection - try: - self._conn.close(manual_commit and rollback_error is None) + self._conn.close(rollback_before_disconnect=True) 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/CMakeLists.txt b/mssql_python/pybind/CMakeLists.txt index 069a1b165..2434e09e4 100644 --- a/mssql_python/pybind/CMakeLists.txt +++ b/mssql_python/pybind/CMakeLists.txt @@ -399,3 +399,21 @@ if(APPLE) target_compile_definitions(ddbc_bindings PRIVATE MACOS_STRING_FIX) target_compile_options(ddbc_bindings PRIVATE -DAPPLE_SILICON) endif() + +# Exercise the real connection/pool/handle code against deterministic ODBC +# function pointers, without shipping test hooks in the extension. +option(BUILD_NATIVE_POOL_TESTS "Build database-free native pool sanitation tests" OFF) +if(BUILD_NATIVE_POOL_TESTS) + find_package(Python3 COMPONENTS Development REQUIRED) + get_target_property(pool_test_sources ddbc_bindings SOURCES) + add_executable(pool_sanitation_tests ${pool_test_sources} + ../../tests/native/pool_sanitation.cpp) + foreach(property INCLUDE_DIRECTORIES COMPILE_DEFINITIONS COMPILE_OPTIONS) + set_property(TARGET pool_sanitation_tests PROPERTY ${property} + $) + endforeach() + target_link_libraries(pool_sanitation_tests PRIVATE + Python3::Python simdutf::simdutf ${CMAKE_DL_LIBS}) + enable_testing() + add_test(NAME pool_sanitation COMMAND pool_sanitation_tests) +endif() diff --git a/mssql_python/pybind/connection/connection.cpp b/mssql_python/pybind/connection/connection.cpp index 9c4304d2f..4499dfb8f 100644 --- a/mssql_python/pybind/connection/connection.cpp +++ b/mssql_python/pybind/connection/connection.cpp @@ -6,6 +6,7 @@ #include "utf_utils.h" #include #include +#include #include #include #include @@ -286,6 +287,8 @@ void Connection::clearResultMetadata() { void Connection::commit() { PERF_TIMER("Connection::commit"); + _poolClean = false; + _poolSessionReset = false; if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } @@ -303,6 +306,8 @@ void Connection::commit() { void Connection::rollback() { PERF_TIMER("Connection::rollback"); + _poolClean = false; + _poolSessionReset = false; if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } @@ -320,10 +325,20 @@ void Connection::rollback() { void Connection::setAutocommit(bool enable) { PERF_TIMER("Connection::setAutocommit"); + if (!enable) { + _poolClean = false; + _poolSessionReset = false; + } if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } clearResultMetadata(); + // Only successful native sanitation with no retained statement/raw handle + // aliases proves ON. A new login or _autocommit alone is not proof. + if (enable && _dbcHandle->get() && _poolClean && !_poolProofDisabled) { + _autocommit = true; + return; + } SQLINTEGER value = enable ? SQL_AUTOCOMMIT_ON : SQL_AUTOCOMMIT_OFF; LOG("Setting autocommit=%d", enable); SQLRETURN ret; @@ -336,7 +351,10 @@ void Connection::setAutocommit(bool enable) { ret = SQLSetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT, reinterpret_cast(static_cast(value)), 0); } - checkError(ret); + if (!SQL_SUCCEEDED(ret)) { + _poolClean = false; + checkError(ret); + } if (value == SQL_AUTOCOMMIT_ON) { LOG("Autocommit enabled"); } else { @@ -354,12 +372,20 @@ bool Connection::getAutocommit() const { SQLINTEGER string_length; SQLRETURN ret = SQLGetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT, &value, sizeof(value), &string_length); - checkError(ret); + if (!SQL_SUCCEEDED(ret)) { + _poolClean = false; + checkError(ret); + } return value == SQL_AUTOCOMMIT_ON; } SqlHandlePtr Connection::allocStatementHandle() { PERF_TIMER("Connection::allocStatementHandle"); + // Invalidate before allocation (including failures), not by parsing SQL. + // Every execution/catalog/fetch path, including direct native calls, needs + // a statement handle. Retained handles also prevent re-establishing proof. + _poolClean = false; + _poolSessionReset = false; LOG("Allocating statement handle"); // Keep the wrapper outside the lock scope: unwinding a failed registration // frees the statement through the same cleanup gate. @@ -417,6 +443,16 @@ SqlHandlePtr Connection::allocStatementHandle() { } SQLRETURN Connection::setAttribute(SQLINTEGER attribute, py::object value) { + _poolClean = false; + _poolSessionReset = false; + // A scalar login timeout only bounds connection establishment; it cannot + // execute user work or retain a deferred buffer. Keep all other attributes + // conservative, and never clear an earlier permanent invalidation. + const bool scalarLoginTimeout = + attribute == SQL_ATTR_LOGIN_TIMEOUT && PyLong_CheckExact(value.ptr()); + if (!scalarLoginTimeout) { + _poolProofDisabled = true; + } clearResultMetadata(); LOG("Setting SQL attribute=%d", attribute); // SQLPOINTER ptr = nullptr; @@ -443,7 +479,21 @@ SQLRETURN Connection::setAttribute(SQLINTEGER attribute, py::object value) { if (py::isinstance(value)) { // Get the integer value - int64_t longValue = value.cast(); + int64_t longValue; + try { + longValue = value.cast(); + } catch (const py::cast_error&) { + _poolProofDisabled = true; + throw; + } catch (const py::error_already_set&) { + _poolProofDisabled = true; + throw; + } + if (scalarLoginTimeout && + (longValue < 0 || + static_cast(longValue) > std::numeric_limits::max())) { + _poolProofDisabled = true; + } SQLRETURN ret; { @@ -456,6 +506,7 @@ SQLRETURN Connection::setAttribute(SQLINTEGER attribute, py::object value) { } if (!SQL_SUCCEEDED(ret)) { + _poolProofDisabled = true; LOG("Failed to set integer attribute=%d, ret=%d", attribute, ret); } else { LOG("Set integer attribute=%d successfully", attribute); @@ -604,6 +655,16 @@ bool Connection::reset() { if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } + if (_cleanupState->childFreeFailed) { + _poolClean = false; + _poolSessionReset = false; + LOG("Cannot reuse connection after a statement handle failed to free"); + return false; + } + if (_poolClean && _poolSessionReset && !_poolProofDisabled) { + updateLastUsed(); + return true; + } clearResultMetadata(); LOG("Resetting connection via SQL_ATTR_RESET_CONNECTION"); // NOTE: SQL_ATTR_RESET_CONNECTION is a pool-checkin reset: it asks the @@ -627,6 +688,7 @@ bool Connection::reset() { (SQLPOINTER)SQL_RESET_CONNECTION_YES, SQL_IS_INTEGER); } if (!SQL_SUCCEEDED(ret)) { + _poolClean = false; LOG("Failed to reset connection (ret=%d). Marking as dead.", ret); return false; } @@ -641,32 +703,117 @@ bool Connection::reset() { (SQLPOINTER)SQL_TXN_READ_COMMITTED, SQL_IS_INTEGER); } if (!SQL_SUCCEEDED(ret)) { + _poolClean = false; LOG("Failed to reset transaction isolation level (ret=%d). Marking as dead.", ret); return false; } + // A retained child can finish cleanup while either attribute call releases the GIL. + if (_cleanupState->childFreeFailed) { + _poolClean = false; + _poolSessionReset = false; + LOG("Cannot reuse connection after a statement handle failed to free"); + return false; + } updateLastUsed(); return true; } -void Connection::prepareForPool(bool transactionAlreadyRolledBack) { +void Connection::prepareForPool() { + PERF_TIMER("Connection::prepareForPool"); if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } + if (_cleanupState->childFreeFailed) { + ThrowStdException("Cannot pool connection after a statement handle failed to free"); + } + if (_poolClean && !_poolProofDisabled) { + PERF_TIMER("Connection::prepareForPool::clean"); + return; + } - // 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); + _poolClean = false; + _poolSessionReset = false; + clearResultMetadata(); + SQLRETURN ret; + std::string statementError; + bool sessionReset = false; + { + // One GIL release and metadata invalidation for the whole sequence. + // Do not trust Python mode/rollback hints: native callers and set_attr + // can change the state, and autocommit permits explicit BEGIN TRAN. + py::gil_scoped_release release; + PERF_TIMER("Connection::prepareForPool::sanitize"); + SQLINTEGER mode = SQL_AUTOCOMMIT_OFF; + SQLINTEGER length = 0; + { + PERF_TIMER("Connection::prepareForPool::get_autocommit"); + ret = SQLGetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT, + &mode, sizeof(mode), &length); + } + if (SQL_SUCCEEDED(ret) && mode == SQL_AUTOCOMMIT_ON) { + // SQLEndTran is a no-op in autocommit mode, but SQL Server still + // permits an explicit BEGIN TRANSACTION. Schedule the driver's + // session reset first so states such as SET NOEXEC ON cannot suppress + // the rollback batch that triggers it. + PERF_TIMER("Connection::prepareForPool::rollback_autocommit"); + ret = SQLSetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_RESET_CONNECTION, + reinterpret_cast(SQL_RESET_CONNECTION_YES), + SQL_IS_INTEGER); + if (SQL_SUCCEEDED(ret)) { + SQLHANDLE statement = nullptr; + ret = SQLAllocHandle_ptr(SQL_HANDLE_STMT, _dbcHandle->get(), &statement); + if (SQL_SUCCEEDED(ret)) { + const std::u16string rollbackQuery = + u"IF @@TRANCOUNT > 0 ROLLBACK TRANSACTION; " + u"SET TRANSACTION ISOLATION LEVEL READ COMMITTED"; + ret = SQLExecDirect_ptr(statement, + reinterpretU16stringAsSqlWChar(rollbackQuery), SQL_NTS); + if (!SQL_SUCCEEDED(ret)) { + ErrorInfo error = SQLReadError(SQL_HANDLE_STMT, statement, ret); + statementError = error.sqlState.length() == 5 + ? "SQLSTATE:" + error.sqlState + ":" + error.ddbcErrorMsg + : error.ddbcErrorMsg; + } + SQLRETURN freeRet = SQLFreeHandle_ptr(SQL_HANDLE_STMT, statement); + if (SQL_SUCCEEDED(ret) && !SQL_SUCCEEDED(freeRet)) { + ErrorInfo error = SQLReadError(SQL_HANDLE_STMT, statement, freeRet); + statementError = error.sqlState.length() == 5 + ? "SQLSTATE:" + error.sqlState + ":" + error.ddbcErrorMsg + : error.ddbcErrorMsg; + ret = freeRet; + } + sessionReset = SQL_SUCCEEDED(ret); + } + } + } else if (SQL_SUCCEEDED(ret)) { + PERF_TIMER("Connection::prepareForPool::rollback_manual"); + ret = SQLEndTran_ptr(SQL_HANDLE_DBC, _dbcHandle->get(), SQL_ROLLBACK); + // Never enable autocommit after a failed rollback: it could commit + // abandoned work. Manual mode can leave even an empty transaction open. + if (SQL_SUCCEEDED(ret)) { + PERF_TIMER("Connection::prepareForPool::autocommit_on"); + ret = SQLSetConnectAttr_ptr(_dbcHandle->get(), SQL_ATTR_AUTOCOMMIT, + reinterpret_cast(SQL_AUTOCOMMIT_ON), 0); + } + } } - if (!transactionAlreadyRolledBack) { - rollback(); + if (!statementError.empty()) { + ThrowStdException(statementError); + } + checkError(ret); + // A weak_ptr expires before its destructor finishes. Read the count before + // the failure latch so a completed destructor cannot hide a failed native free. + const bool noChildren = _cleanupState->childStatementCount == 0; + if (_cleanupState->childFreeFailed) { + ThrowStdException("Cannot pool connection after a statement handle failed to free"); } - // 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); + _autocommit = true; + _poolSessionReset = sessionReset; + updateLastUsed(); + // A native statement alias can execute again without another allocation, + // even in a later lease. All wrappers must finish cleanup before the fast path. + _poolClean = !_poolProofDisabled && noChildren; } void Connection::updateLastUsed() { @@ -781,28 +928,38 @@ ConnectionHandle::~ConnectionHandle() { } } -void ConnectionHandle::close(bool transactionAlreadyRolledBack) { +void ConnectionHandle::close(bool rollbackBeforeDisconnect) { PERF_TIMER("ConnectionHandle::close"); if (!_conn) { ThrowStdException("Connection object is not initialized"); } - if (_usePool) { - 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 (...) { + try { + if (_usePool) { + _conn->prepareForPool(); + } else { + // Preserve unpooled close semantics without doing pool sanitation. + if (rollbackBeforeDisconnect && !_conn->getAutocommit()) { + _conn->rollback(); } - _conn = nullptr; + _conn->disconnect(); + } + } catch (...) { + // A low-level unpooled close still allows the caller to recover from + // SQLDisconnect failure using the existing connection and children. + if (!_usePool && !rollbackBeforeDisconnect) { throw; } + // Never retain a connection whose transaction state could not be + // sanitized. Release capacity and preserve the original cleanup error. + try { + ConnectionPoolManager::getInstance().discardConnection(_originPool, _conn); + } catch (...) { + } + _conn = nullptr; + throw; + } + if (_usePool) { ConnectionPoolManager::getInstance().returnConnection(_poolKey, _originPool, _conn); - } else { - _conn->disconnect(); } _conn = nullptr; } @@ -849,6 +1006,12 @@ SqlHandlePtr ConnectionHandle::allocStatementHandle() { } py::object Connection::getInfo(SQLUSMALLINT infoType) const { + _poolClean = false; + _poolSessionReset = false; + if (infoType == SQL_DRIVER_HDBC || infoType == SQL_DRIVER_HENV || + infoType == SQL_DRIVER_HSTMT || infoType == SQL_DRIVER_HLIB) { + _poolProofDisabled = true; + } if (!_dbcHandle) { ThrowStdException("Connection handle not allocated"); } diff --git a/mssql_python/pybind/connection/connection.h b/mssql_python/pybind/connection/connection.h index f43613842..c9dc6fab5 100644 --- a/mssql_python/pybind/connection/connection.h +++ b/mssql_python/pybind/connection/connection.h @@ -59,7 +59,7 @@ class Connection { bool getAutocommit() const; bool isAlive() const; bool reset(); - void prepareForPool(bool transactionAlreadyRolledBack = false); + void prepareForPool(); void updateLastUsed(); std::chrono::steady_clock::time_point lastUsed() const; @@ -108,6 +108,14 @@ class Connection { std::u16string _connStr; bool _fromPool = false; bool _autocommit = true; + // Only successful rollback + AUTOCOMMIT_ON establishes this proof. A new + // login or deferred SQL_ATTR_RESET_CONNECTION alone does not establish it. + mutable bool _poolClean = false; + // True only after check-in has triggered the driver reset and restored the + // default isolation level. This permits checkout to skip a duplicate reset. + mutable bool _poolSessionReset = false; + // Escaped raw handles and arbitrary/deferred attributes cannot be tracked. + mutable bool _poolProofDisabled = false; SqlHandlePtr _dbcHandle; std::chrono::steady_clock::time_point _lastUsed; // POSIX-epoch expiry (seconds) of the access token this connection last @@ -152,7 +160,7 @@ class ConnectionHandle { const py::object& tokenFactory = py::object()); ~ConnectionHandle(); - void close(bool transactionAlreadyRolledBack = false); + void close(bool rollbackBeforeDisconnect = 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 4e4523955..b536fc2d7 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -1590,12 +1590,19 @@ void DriverLoader::loadDriver() { // SqlHandle definition SqlHandle::SqlHandle(SQLSMALLINT type, SQLHANDLE rawHandle, std::shared_ptr cleanupState) - : _type(type), _handle(rawHandle), _cleanupState(std::move(cleanupState)) {} + : _type(type), _handle(rawHandle), _cleanupState(std::move(cleanupState)) { + if (_type == SQL_HANDLE_STMT && _cleanupState) { + ++_cleanupState->childStatementCount; + } +} SqlHandle::~SqlHandle() { if (_handle) { free(); } + if (_type == SQL_HANDLE_STMT && _cleanupState) { + --_cleanupState->childStatementCount; + } } std::unique_lock SqlHandle::lockForCleanup() const { @@ -1660,7 +1667,7 @@ SQLRETURN SqlHandle::freeHandle() { auto freeNative = [this]() -> SQLRETURN { auto cleanupLock = lockForCleanup(); - if (!_handle || !SQLFreeHandle_ptr) { + if (!_handle) { return SQL_INVALID_HANDLE; } describeCache.clear(); @@ -1668,9 +1675,11 @@ SQLRETURN SqlHandle::freeHandle() { _handle = nullptr; return SQL_SUCCESS; } - SQLRETURN ret = SQLFreeHandle_ptr(_type, _handle); + SQLRETURN ret = SQLFreeHandle_ptr ? SQLFreeHandle_ptr(_type, _handle) : SQL_INVALID_HANDLE; if (SQL_SUCCEEDED(ret)) { _handle = nullptr; + } else if (_type == SQL_HANDLE_STMT && _cleanupState) { + _cleanupState->childFreeFailed = true; } return ret; }; @@ -2025,10 +2034,22 @@ py::list SQLGetAllDiagRecords(SqlHandlePtr handle) { return records; } +static bool FetchDiagnosticsAreEmpty(SQLHSTMT hStmt) { + if (!SQLGetDiagField_ptr) + return false; + // Diagnostic header reads preserve the originating records. An unsupported + // or uncertain count must retain the original record enumeration. + SQLINTEGER count = -1; + SQLRETURN ret = SQLGetDiagField_ptr(SQL_HANDLE_STMT, hStmt, 0, SQL_DIAG_NUMBER, + &count, 0, nullptr); + return ret == SQL_SUCCESS && count == 0; +} + // Called only with the GIL held, immediately after the originating ODBC call. static void CaptureFetchDiagnostics(SQLHSTMT hStmt, SQLRETURN ret, py::handle messages, bool internalTruncation = false) { - if ((ret == SQL_SUCCESS_WITH_INFO || ret == SQL_NO_DATA) && messages && !messages.is_none()) + if ((ret == SQL_SUCCESS_WITH_INFO || ret == SQL_NO_DATA) && messages && !messages.is_none() && + !FetchDiagnosticsAreEmpty(hStmt)) AppendDiagRecords(hStmt, SQL_HANDLE_STMT, messages, internalTruncation); } @@ -6471,7 +6492,7 @@ PYBIND11_MODULE(ddbc_bindings, m) { 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, - py::arg("transaction_already_rolled_back") = false, "Close the connection") + py::arg("rollback_before_disconnect") = 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 748ecc5ca..8fc5f9d27 100644 --- a/mssql_python/pybind/ddbc_bindings.h +++ b/mssql_python/pybind/ddbc_bindings.h @@ -13,6 +13,7 @@ #include // Add this line for datetime support #include +#include #include #include #include @@ -291,6 +292,9 @@ struct DescribedParamInfo { struct ConnectionCleanupState { std::mutex mutex; bool disconnected = false; // Protected by mutex, shared with every child. + // Unlike weak_ptr expiration, this count includes destructors still freeing handles. + std::atomic childStatementCount{0}; + std::atomic childFreeFailed{false}; // Sticky for this physical connection. }; class SqlHandle { diff --git a/tests/native/pool_sanitation.cpp b/tests/native/pool_sanitation.cpp new file mode 100644 index 000000000..a866c1044 --- /dev/null +++ b/tests/native/pool_sanitation.cpp @@ -0,0 +1,1177 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +#include "connection/connection.h" +#include "connection/connection_pool.h" +#include "logger_bridge.hpp" +#include +#include +#include +#include +#include +#include + +SQLSMALLINT SQLNumResultCols_wrap(SqlHandlePtr statementHandle, py::handle messages); +SQLRETURN FetchOne_wrap(SqlHandlePtr statementHandle, py::list& row, + const std::string& charEncoding, const std::string& wcharEncoding, + int charCtype, py::handle messages); +py::list SQLGetAllDiagRecords(SqlHandlePtr handle); + +namespace { +struct Handle { + SQLSMALLINT type; + SQLHANDLE parent; + bool autocommit = true; + bool transaction = false; + bool pendingWork = false; + bool resetPending = false; + SQLULEN loginTimeout = 0; +}; + +std::unordered_map> handles; +std::vector calls; +std::string failNext; +int commits = 0; +int logins = 0; +std::vector loginTimeouts; +std::function duringFree; +std::function duringRollbackBatch; +std::function duringReset; + +struct CallbackScope { + std::function& hook; + CallbackScope(std::function& hook, std::function action) : hook(hook) { + hook = std::move(action); + } + ~CallbackScope() { hook = {}; } +}; + +void require(bool condition, const std::string& message) { + if (!condition) { + throw std::runtime_error(message); + } +} + +SQLRETURN record(const std::string& operation) { + calls.push_back(operation); + if (operation == failNext) { + failNext.clear(); + return SQL_ERROR; + } + return SQL_SUCCESS; +} + +SQLRETURN SQL_API allocate(SQLSMALLINT type, SQLHANDLE parent, SQLHANDLE* out) { + auto ret = record(type == SQL_HANDLE_STMT ? "allocate_statement" : "allocate"); + if (!SQL_SUCCEEDED(ret)) { + return ret; + } + auto handle = std::make_unique(Handle{type, parent}); + *out = handle.get(); + handles.emplace(*out, std::move(handle)); + return SQL_SUCCESS; +} + +SQLRETURN SQL_API freeHandle(SQLSMALLINT, SQLHANDLE handle) { + if (duringFree) duringFree(); + auto ret = record("free"); + if (SQL_SUCCEEDED(ret)) { + handles.erase(handle); + } + return ret; +} + +SQLRETURN SQL_API setEnv(SQLHENV, SQLINTEGER, SQLPOINTER, SQLINTEGER) { + return SQL_SUCCESS; +} + +SQLRETURN SQL_API login(SQLHDBC dbc, SQLHWND, SQLWCHAR*, SQLSMALLINT, + SQLWCHAR*, SQLSMALLINT, SQLSMALLINT*, SQLUSMALLINT) { + ++logins; + loginTimeouts.push_back(handles.at(dbc)->loginTimeout); + return record("login"); +} + +SQLRETURN SQL_API setAttr(SQLHDBC dbc, SQLINTEGER attribute, SQLPOINTER value, SQLINTEGER length) { + std::string operation = "attribute"; + if (attribute == SQL_ATTR_AUTOCOMMIT) { + operation = value ? "on" : "off"; + } else if (attribute == SQL_ATTR_RESET_CONNECTION) { + operation = "reset"; + } else if (attribute == SQL_ATTR_TXN_ISOLATION) { + operation = "isolation"; + } + auto ret = record(operation); + if (!SQL_SUCCEEDED(ret)) { + return ret; + } + auto& handle = *handles.at(dbc); + if (attribute == SQL_ATTR_AUTOCOMMIT) { + if (value) { + if (handle.pendingWork) { + ++commits; + } + handle.transaction = handle.pendingWork = false; + } + handle.autocommit = value != nullptr; + } else if (attribute == SQL_ATTR_RESET_CONNECTION) { + if (duringReset) duringReset(); + handle.resetPending = true; + } else if (attribute == SQL_ATTR_LOGIN_TIMEOUT && length == SQL_IS_INTEGER) { + handle.loginTimeout = reinterpret_cast(value); + } + // A reset is deferred until execution. It must not establish clean proof. + return SQL_SUCCESS; +} + +SQLRETURN SQL_API getAttr(SQLHDBC dbc, SQLINTEGER attribute, SQLPOINTER value, + SQLINTEGER, SQLINTEGER*) { + auto ret = record(attribute == SQL_ATTR_AUTOCOMMIT ? "get" : "alive"); + if (SQL_SUCCEEDED(ret)) { + *static_cast(value) = attribute == SQL_ATTR_AUTOCOMMIT + ? (handles.at(dbc)->autocommit ? SQL_AUTOCOMMIT_ON : SQL_AUTOCOMMIT_OFF) + : SQL_CD_FALSE; + } + return ret; +} + +SQLRETURN SQL_API endTran(SQLSMALLINT, SQLHANDLE dbc, SQLSMALLINT completion) { + auto ret = record(completion == SQL_ROLLBACK ? "rollback" : "commit"); + if (SQL_SUCCEEDED(ret) && !handles.at(dbc)->autocommit) { + auto& handle = *handles.at(dbc); + if (completion == SQL_COMMIT && handle.pendingWork) { + ++commits; + } + handle.pendingWork = false; + // Model the empty manual-mode transaction that motivated PR #777. + handle.transaction = true; + } + return ret; +} + +SQLRETURN SQL_API executeDirect(SQLHSTMT statement, SQLWCHAR*, SQLINTEGER) { + if (duringRollbackBatch) duringRollbackBatch(); + auto& handle = *handles.at(handles.at(statement)->parent); + if (handle.resetPending) { + handle.transaction = handle.pendingWork = false; + handle.resetPending = false; + } + auto ret = record("rollback_batch"); + if (SQL_SUCCEEDED(ret)) { + handle.transaction = handle.pendingWork = false; + } + return ret; +} + +SQLRETURN SQL_API disconnect(SQLHDBC dbc) { + record("disconnect"); + if (handles.at(dbc)->pendingWork) { + return SQL_ERROR; + } + handles.at(dbc)->transaction = handles.at(dbc)->pendingWork = false; + for (auto it = handles.begin(); it != handles.end();) { + if (it->second->parent == dbc) { + it = handles.erase(it); + } else { + ++it; + } + } + return SQL_SUCCESS; +} + +SQLRETURN SQL_API getInfo(SQLHDBC dbc, SQLUSMALLINT, SQLPOINTER value, + SQLSMALLINT bufferLength, SQLSMALLINT* length) { + const auto required = static_cast(sizeof(dbc)); + if (length) { + *length = required; + } + if (bufferLength < 0 || (value && bufferLength < required)) { + return SQL_ERROR; + } + if (value) { + std::copy_n(reinterpret_cast(&dbc), sizeof(dbc), + static_cast(value)); + } + return SQL_SUCCESS; +} + +SQLRETURN SQL_API diagnostic(SQLSMALLINT, SQLHANDLE, SQLSMALLINT recordNumber, + SQLWCHAR* state, SQLINTEGER* native, SQLWCHAR* message, + SQLSMALLINT, SQLSMALLINT* length) { + if (recordNumber != 1) { + return SQL_NO_DATA; + } + const SQLWCHAR sqlstate[] = {'H', 'Y', '0', '0', '0', 0}; + const SQLWCHAR text[] = {'f', 'a', 'i', 'l', 'e', 'd', 0}; + std::copy(std::begin(sqlstate), std::end(sqlstate), state); + std::copy(std::begin(text), std::end(text), message); + *native = 0; + *length = 6; + return SQL_SUCCESS; +} + +void expectCalls(std::initializer_list expected) { + std::vector wanted(expected.begin(), expected.end()); + if (calls != wanted) { + std::string actual; + for (const auto& call : calls) { + actual += call + " "; + } + throw std::runtime_error("Unexpected ODBC calls: " + actual); + } +} + +SQLRETURN SQL_API freeStatement(SQLHSTMT, SQLUSMALLINT option) { + require(option == SQL_UNBIND, "Unexpected statement cleanup"); + return record("statement_unbind"); +} + +struct DiagnosticRecord { + std::u16string state; + std::u16string message; + SQLRETURN result = SQL_SUCCESS; +}; +std::vector fetchRecords; +SQLRETURN fetchResult = SQL_SUCCESS_WITH_INFO; +SQLRETURN numberResult = SQL_SUCCESS; +SQLINTEGER diagnosticCount = 0; +bool writeDiagnosticCount = true; +SQLRETURN stateResult = SQL_SUCCESS; + +SQLRETURN SQL_API fetchDiagnostic(SQLSMALLINT type, SQLHANDLE, SQLSMALLINT number, + SQLWCHAR* state, SQLINTEGER* native, SQLWCHAR* message, + SQLSMALLINT capacity, SQLSMALLINT* length) { + require(type == SQL_HANDLE_STMT && number > 0, "Invalid record lookup"); + record("diag_record"); + if (static_cast(number) > fetchRecords.size()) return SQL_NO_DATA; + const auto& item = fetchRecords[number - 1]; + require(item.state.size() == 5, "Invalid test SQLSTATE"); + std::copy(item.state.begin(), item.state.end(), state); + state[5] = 0; + *native = 42; + const auto size = std::min(item.message.size(), static_cast(capacity - 1)); + std::copy_n(item.message.begin(), size, message); + message[size] = 0; + *length = static_cast(item.message.size()); + return item.result; +} + +SQLRETURN SQL_API fetchDiagnosticField(SQLSMALLINT type, SQLHANDLE, SQLSMALLINT number, + SQLSMALLINT identifier, SQLPOINTER value, + SQLSMALLINT capacity, SQLSMALLINT* length) { + require(type == SQL_HANDLE_STMT, "Wrong diagnostic handle type"); + if (identifier == SQL_DIAG_NUMBER) { + record("diag_number"); + require(number == 0 && capacity == 0 && length == nullptr, + "SQL_DIAG_NUMBER must read the numeric header field"); + if (writeDiagnosticCount) *static_cast(value) = diagnosticCount; + return numberResult; + } + record("diag_state"); + require(identifier == SQL_DIAG_SQLSTATE && number > 0, "Unexpected diagnostic field"); + if (stateResult != SQL_SUCCESS) return stateResult; + if (static_cast(number) > fetchRecords.size()) return SQL_NO_DATA; + const auto& state = fetchRecords[number - 1].state; + require(capacity >= static_cast(6 * sizeof(SQLWCHAR)), "SQLSTATE buffer too small"); + std::copy(state.begin(), state.end(), static_cast(value)); + static_cast(value)[5] = 0; + return SQL_SUCCESS; +} + +SQLRETURN SQL_API fetchColumnCount(SQLHSTMT, SQLSMALLINT* count) { + record("column_count"); + *count = 0; + return fetchResult; +} + +SQLRETURN SQL_API fetchNullLob(SQLHANDLE, SQLUSMALLINT, SQLSMALLINT, SQLPOINTER, SQLLEN, + SQLLEN* length) { + record("get_lob"); + *length = SQL_NULL_DATA; + return fetchResult; +} + +SQLRETURN SQL_API fetchNoData(SQLHANDLE) { + record("fetch_no_data"); + return SQL_NO_DATA; +} + +struct FetchDiagnosticScope { + SQLGetDiagRecFunc oldRecord = SQLGetDiagRec_ptr; + SQLGetDiagFieldFunc oldField = SQLGetDiagField_ptr; + SQLNumResultColsFunc oldCount = SQLNumResultCols_ptr; + SQLGetDataFunc oldData = SQLGetData_ptr; + SQLFetchFunc oldFetch = SQLFetch_ptr; + SQLFreeStmtFunc oldFreeStmt = SQLFreeStmt_ptr; + + FetchDiagnosticScope() { + fetchRecords.clear(); + fetchResult = SQL_SUCCESS_WITH_INFO; + numberResult = stateResult = SQL_SUCCESS; + diagnosticCount = 0; + writeDiagnosticCount = true; + SQLGetDiagRec_ptr = fetchDiagnostic; + SQLGetDiagField_ptr = fetchDiagnosticField; + SQLNumResultCols_ptr = fetchColumnCount; + SQLGetData_ptr = fetchNullLob; + SQLFetch_ptr = fetchNoData; + SQLFreeStmt_ptr = freeStatement; + } + ~FetchDiagnosticScope() { + SQLGetDiagRec_ptr = oldRecord; + SQLGetDiagField_ptr = oldField; + SQLNumResultCols_ptr = oldCount; + SQLGetData_ptr = oldData; + SQLFetch_ptr = oldFetch; + SQLFreeStmt_ptr = oldFreeStmt; + } +}; + +void expectMessage(const py::list& messages, size_t index, const std::string& state, + const std::string& text) { + auto message = messages[index].cast(); + require(message[0].cast() == "[" + state + "] (42)" && + message[1].cast() == text, "Diagnostic message changed"); +} + +void expectFailure(const std::function& action) { + try { + action(); + } catch (const std::runtime_error&) { + return; + } + throw std::runtime_error("Expected native failure"); +} + +std::unique_ptr acquire(bool pooled = true, + const py::dict& attrs = py::dict()) { + return std::make_unique(u"native pool test", pooled, attrs); +} + +void checkParked() { + require(commits == 0, "Cleanup committed pending work"); + for (const auto& item : handles) { + if (item.second->type == SQL_HANDLE_DBC) { + require(item.second->autocommit && !item.second->transaction, + "Pooled DBC retained a transaction or manual mode"); + } + } +} + +void warm() { + auto connection = acquire(); + connection->close(); + checkParked(); +} + +void startWork(const SqlHandlePtr& statement) { + auto dbc = handles.at(statement->get())->parent; + handles.at(dbc)->transaction = handles.at(dbc)->pendingWork = true; +} + +void resetScenario() { + failNext.clear(); + auto& manager = ConnectionPoolManager::getInstance(); + manager.closePools(); + manager.configure(1, 30); + manager.setAccepting(true); + calls.clear(); + commits = logins = 0; + loginTimeouts.clear(); +} +} // namespace + +int main() { + py::scoped_interpreter interpreter; + mssql_python::logging::LoggerBridge::updateLevel(1000); + SQLAllocHandle_ptr = allocate; + SQLFreeHandle_ptr = freeHandle; + SQLSetEnvAttr_ptr = setEnv; + SQLDriverConnect_ptr = login; + SQLSetConnectAttr_ptr = setAttr; + SQLGetConnectAttr_ptr = getAttr; + SQLExecDirect_ptr = executeDirect; + SQLEndTran_ptr = endTran; + SQLDisconnect_ptr = disconnect; + SQLGetInfo_ptr = getInfo; + SQLGetDiagRec_ptr = diagnostic; + + int passed = 0; + auto run = [&](const char* name, const std::function& test) { + resetScenario(); + test(); + ++passed; + std::cout << "PASS " << name << '\n'; + }; + try { + run("getInfo size queries allow null output and length pointers", [] { + Handle handle{SQL_HANDLE_DBC, nullptr}; + SQLSMALLINT length = 0; + require(getInfo(&handle, SQL_DRIVER_HDBC, nullptr, 0, &length) == SQL_SUCCESS, + "Size query failed"); + require(length == static_cast(sizeof(SQLHDBC)), + "Size query returned the wrong handle size"); + require(getInfo(&handle, SQL_DRIVER_HDBC, nullptr, 0, nullptr) == SQL_SUCCESS, + "Optional length pointer was dereferenced"); + }); + run("getInfo rejects short buffers without writing", [] { + Handle handle{SQL_HANDLE_DBC, nullptr}; + std::array buffer; + for (SQLSMALLINT capacity = -1; + capacity < static_cast(buffer.size()); ++capacity) { + buffer.fill(0xa5); + SQLSMALLINT length = 0; + require(getInfo(&handle, SQL_DRIVER_HDBC, buffer.data(), capacity, &length) == SQL_ERROR, + "Short buffer was accepted"); + require(length == static_cast(sizeof(SQLHDBC)), + "Short buffer did not report the required size"); + require(std::all_of(buffer.begin(), buffer.end(), + [](unsigned char byte) { return byte == 0xa5; }), + "Short buffer was modified"); + } + }); + run("getInfo copies handle bytes into an unaligned exact-size buffer", [] { + Handle handle{SQL_HANDLE_DBC, nullptr}; + SQLHDBC dbc = &handle; + alignas(SQLHDBC) std::array buffer; + buffer.fill(0xa5); + require(getInfo(dbc, SQL_DRIVER_HDBC, buffer.data() + 1, + static_cast(sizeof(dbc)), nullptr) == SQL_SUCCESS, + "Exact-size buffer was rejected"); + require(std::equal(buffer.begin() + 1, buffer.end() - 1, + reinterpret_cast(&dbc)), + "Copied handle representation differs"); + require(buffer.front() == 0xa5 && buffer.back() == 0xa5, + "Copy wrote outside the advertised buffer"); + }); + for (SQLRETURN origin : {SQLRETURN(SQL_SUCCESS_WITH_INFO), SQLRETURN(SQL_NO_DATA)}) { + run("successful zero-record header avoids enumeration without clearing messages", [origin] { + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + FetchDiagnosticScope diagnostics; + fetchResult = origin; + py::list messages; + messages.append(py::make_tuple("existing", "preserved")); + calls.clear(); + require(SQLNumResultCols_wrap(statement, messages) == 0, "Column count changed"); + expectCalls({"column_count", "diag_number"}); + require(messages.size() == 1, "Zero-record gate cleared existing messages"); + statement.reset(); + connection->close(); + }); + } + for (int scenario = 0; scenario < 9; ++scenario) { + run("missing unsupported info unknown and nonzero headers retain enumeration", [scenario] { + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + FetchDiagnosticScope diagnostics; + fetchRecords = {{u"01000", u"PRINT message"}, + {u"01004", u"visible truncation", SQL_SUCCESS_WITH_INFO}}; + switch (scenario) { + case 0: SQLGetDiagField_ptr = nullptr; break; + case 1: numberResult = SQL_ERROR; break; + case 2: numberResult = SQL_SUCCESS_WITH_INFO; break; + case 3: diagnosticCount = -1; break; + case 4: writeDiagnosticCount = false; break; + case 5: diagnosticCount = 2; break; + case 6: numberResult = SQL_NO_DATA; break; + case 7: numberResult = SQL_INVALID_HANDLE; break; + case 8: diagnosticCount = 1; break; // Never cap enumeration at the header. + } + py::list messages; + calls.clear(); + SQLNumResultCols_wrap(statement, messages); + if (scenario == 0) { + expectCalls({"column_count", "diag_record", "diag_record", "diag_record"}); + } else { + expectCalls({"column_count", "diag_number", "diag_record", "diag_record", + "diag_record"}); + } + require(messages.size() == 2, "Fallback lost warning or PRINT records"); + expectMessage(messages, 0, "01000", "PRINT message"); + expectMessage(messages, 1, "01004", "visible truncation"); + statement.reset(); + connection->close(); + }); + } + for (bool empty : {false, true}) { + run("actual SQLFetch NO_DATA preserves return and any messages", [empty] { + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + FetchDiagnosticScope diagnostics; + if (!empty) { + diagnosticCount = 1; + fetchRecords = {{u"01000", u"final PRINT"}}; + } + py::list rows, messages; + calls.clear(); + auto ret = FetchOne_wrap(statement, rows, "utf-16le", "utf-16le", + SQL_C_WCHAR, messages); + require(ret == SQL_NO_DATA && rows.empty(), "NO_DATA result changed"); + if (empty) { + expectCalls({"statement_unbind", "fetch_no_data", "diag_number"}); + require(messages.empty(), "Empty diagnostics created a message"); + } else { + expectCalls({"statement_unbind", "fetch_no_data", "diag_number", + "diag_record", "diag_record"}); + require(messages.size() == 1, "NO_DATA lost its PRINT record"); + expectMessage(messages, 0, "01000", "final PRINT"); + } + statement.reset(); + connection->close(); + }); + } + for (int scenario = 0; scenario < 4; ++scenario) { + run("LOB continuation filters only internal truncation and retains unrelated warnings", [scenario] { + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + FetchDiagnosticScope diagnostics; + diagnosticCount = 3; + fetchRecords = {{u"01004", u"internal truncation"}, + {u"01000", u"PRINT message"}, + {u"01S02", u"option changed", SQL_SUCCESS_WITH_INFO}}; + if (scenario == 1) SQLGetDiagField_ptr = nullptr; + if (scenario == 2) stateResult = SQL_ERROR; + if (scenario == 3) stateResult = SQL_SUCCESS_WITH_INFO; + py::list messages; + calls.clear(); + auto value = FetchLobColumnData(statement->get(), 1, SQL_C_WCHAR, + true, false, "utf-16le", messages); + require(value.is_none() && messages.size() == 2, + "LOB continuation lost warnings or exposed internal truncation"); + expectMessage(messages, 0, "01000", "PRINT message"); + expectMessage(messages, 1, "01S02", "option changed"); + require(std::count(calls.begin(), calls.end(), "diag_number") == + (scenario == 1 ? 0 : 1), + "LOB header probe count changed"); + statement.reset(); + connection->close(); + }); + } + run("no probe for success absent messages or explicit diagnostic enumeration", [] { + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + FetchDiagnosticScope diagnostics; + py::list messages; + fetchResult = SQL_SUCCESS; + calls.clear(); + SQLNumResultCols_wrap(statement, messages); + expectCalls({"column_count"}); + fetchResult = SQL_SUCCESS_WITH_INFO; + calls.clear(); + SQLNumResultCols_wrap(statement, {}); + SQLNumResultCols_wrap(statement, py::none()); + expectCalls({"column_count", "column_count"}); + fetchRecords = {{u"01000", u"explicit diagnostic"}}; + calls.clear(); + messages = SQLGetAllDiagRecords(statement); + expectCalls({"diag_record", "diag_record"}); + expectMessage(messages, 0, "01000", "explicit diagnostic"); + statement.reset(); + connection->close(); + }); + run("new login is not clean proof; repeated empty leases skip all sanitation", [] { + auto connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + for (int i = 0; i < 100; ++i) { + calls.clear(); + connection = acquire(); + expectCalls({"alive"}); + calls.clear(); + connection->setAutocommit(true); + expectCalls({}); + calls.clear(); + connection->close(true); + expectCalls({}); + checkParked(); + } + require(logins == 1, "Empty leases did not reuse the physical connection"); + }); + run("new login and repeated setters do not establish clean proof", [] { + auto connection = acquire(); + calls.clear(); + connection->setAutocommit(true); + connection->setAutocommit(true); + expectCalls({"on", "on"}); + connection->close(); + connection = acquire(); + calls.clear(); + connection->setAutocommit(true); + expectCalls({}); + connection->close(); + }); + run("manual mode is never elided; only successful sanitation restores proof", [] { + warm(); + auto connection = acquire(); + calls.clear(); + connection->setAutocommit(false); + connection->setAutocommit(false); + connection->setAutocommit(true); + connection->setAutocommit(true); + expectCalls({"off", "off", "on", "on"}); + connection->close(); + connection = acquire(); + calls.clear(); + connection->setAutocommit(true); + expectCalls({}); + connection->close(); + }); + run("failed setter preserves metadata invalidation and sanitation requirement", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + auto generation = statement->resultMetadata.snapshot().generation; + statement->resultMetadata.publish(generation, std::make_shared()); + calls.clear(); + failNext = "on"; + expectFailure([&] { connection->setAutocommit(true); }); + expectCalls({"on"}); + auto snapshot = statement->resultMetadata.snapshot(); + require(!snapshot.metadata && snapshot.generation == generation + 1, + "Setter did not invalidate metadata before ODBC failure"); + statement.reset(); + calls.clear(); + connection->setAutocommit(true); + expectCalls({"on"}); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + run("timeout=30 is applied before login and preserves repeated empty-lease fast path", [] { + py::dict attrs; + attrs[py::int_(SQL_ATTR_LOGIN_TIMEOUT)] = py::int_(30); + auto connection = acquire(true, attrs); + connection->setAutocommit(true); + require(loginTimeouts == std::vector{30}, + "Login timeout was not applied as 30 before SQLDriverConnect"); + require(std::count(calls.begin(), calls.end(), "attribute") == 1, + "Login timeout was not applied exactly once"); + calls.clear(); + connection->close(true); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + for (int i = 0; i < 100; ++i) { + connection = acquire(true, attrs); + connection->setAutocommit(true); + calls.clear(); + connection->close(true); + expectCalls({}); + checkParked(); + } + require(logins == 1 && loginTimeouts == std::vector{30}, + "Timed leases did not reuse the physical connection with timeout 30"); + }); + run("valid scalar timeout cannot make dirty work clean", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + statement.reset(); + connection->setAttr(SQL_ATTR_LOGIN_TIMEOUT, py::int_(30)); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + checkParked(); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({}); + }); + for (bool timeoutFirst : {true, false}) { + run("scalar timeout never clears an unknown attribute's permanent invalidation", [timeoutFirst] { + py::dict attrs; + auto setTimeout = [&] { + attrs[py::int_(SQL_ATTR_LOGIN_TIMEOUT)] = py::int_(30); + }; + if (timeoutFirst) { + setTimeout(); + } + attrs[py::int_(SQL_ATTR_AUTOCOMMIT)] = py::int_(SQL_AUTOCOMMIT_ON); + if (!timeoutFirst) { + setTimeout(); + } + auto connection = acquire(true, attrs); + require(loginTimeouts == std::vector{30}, + "Timeout was lost when combined with another attribute"); + connection->close(); + connection = acquire(); + connection->setAttr(SQL_ATTR_LOGIN_TIMEOUT, py::int_(30)); + connection->close(); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + for (const py::object& value : std::vector{ + py::str("30"), py::bytes("30"), py::bool_(true), + py::int_(-1), py::int_(uint64_t{1} << 32)}) { + run("nonscalar and out-of-range login timeouts permanently disable proof", [&value] { + warm(); + auto connection = acquire(); + connection->setAttr(SQL_ATTR_LOGIN_TIMEOUT, value); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + for (const py::object& value : std::vector{ + py::float_(30.0), py::eval("1 << 100")}) { + run("unsupported or overflowing timeout conversion fails closed", [&value] { + warm(); + auto connection = acquire(); + expectFailure([&] { connection->setAttr(SQL_ATTR_LOGIN_TIMEOUT, value); }); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + run("failed login-timeout application releases capacity without login", [] { + py::dict attrs; + attrs[py::int_(SQL_ATTR_LOGIN_TIMEOUT)] = py::int_(30); + failNext = "attribute"; + expectFailure([&] { acquire(true, attrs); }); + require(logins == 0, "Connected despite failed login-timeout application"); + auto connection = acquire(true, attrs); + require(loginTimeouts == std::vector{30}, + "Failed timeout application did not release pool capacity"); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + run("manual mode rolls back once and parks in autocommit", [] { + warm(); + auto connection = acquire(); + connection->setAutocommit(false); + calls.clear(); + connection->close(); + expectCalls({"get", "rollback", "on"}); + checkParked(); + }); + run("statement allocation invalidates even without execution", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + statement.reset(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({}); + }); + run("failed statement allocation invalidates clean proof", [] { + warm(); + auto connection = acquire(); + failNext = "allocate_statement"; + expectFailure([&] { connection->allocStatementHandle(); }); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + run("explicit transaction and retained statement alias across leases", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + auto generation = statement->resultMetadata.snapshot().generation; + statement->resultMetadata.publish(generation, std::make_shared()); + connection->close(); + auto metadata = statement->resultMetadata.snapshot(); + require(!metadata.metadata && metadata.generation == generation + 1, + "Pool sanitation must invalidate result metadata exactly once"); + checkParked(); + connection = acquire(); + calls.clear(); + connection->setAutocommit(true); + expectCalls({"on"}); + startWork(statement); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + checkParked(); + statement.reset(); + connection = acquire(); + connection->close(); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({}); + }); + for (bool fail : {false, true}) { + run("child lifetime count outlives weak pointer expiration", [fail] { + auto cleanup = std::make_shared(); + SQLHANDLE nativeStatement = nullptr; + require(SQL_SUCCEEDED(allocate(SQL_HANDLE_STMT, nullptr, &nativeStatement)), + "Test statement allocation failed"); + auto statement = std::make_shared( + static_cast(SQL_HANDLE_STMT), nativeStatement, cleanup); + std::weak_ptr alias = statement; + bool sawExpiredWithPendingCleanup = false; + { + CallbackScope callback(duringFree, [&] { + sawExpiredWithPendingCleanup = + alias.expired() && cleanup->childStatementCount == 1; + }); + if (fail) failNext = "free"; + statement.reset(); + } + require(sawExpiredWithPendingCleanup, + "Lifetime count did not cover the expired wrapper's destructor"); + require(cleanup->childStatementCount == 0, "Completed destructor retained its count"); + require(cleanup->childFreeFailed == fail, "Destructor lost the native free outcome"); + if (fail) { + require(SQL_SUCCEEDED(freeHandle(SQL_HANDLE_STMT, nativeStatement)), + "Test orphan cleanup failed"); + } + }); + } + run("failed cursor free discards the connection and invalidates surviving aliases", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + const auto nativeStatement = statement->get(); + const auto dbc = handles.at(nativeStatement)->parent; + failNext = "free"; + require(statement->freeHandle() == SQL_ERROR, "Expected failed free"); + calls.clear(); + expectFailure([&] { connection->close(); }); + expectCalls({"off", "rollback", "disconnect", "free"}); + require(handles.count(nativeStatement) == 0 && handles.count(dbc) == 0, + "Discard did not reclaim the failed statement and its connection"); + require(statement->isImplicitlyFreed(), "Surviving alias was not invalidated"); + calls.clear(); + require(statement->freeHandle() == SQL_SUCCESS, "Implicitly freed alias failed cleanup"); + statement.reset(); + expectCalls({}); + require(commits == 0, "Discard committed pending work"); + connection = acquire(); + require(logins == 2, "Discard did not release capacity / replace DBC"); + connection->close(); + checkParked(); + }); + for (bool autocommit : {true, false}) { + for (bool explicitFailure : {true, false}) { + run("expired alias after failed destruction cannot return an orphan to the pool", + [autocommit, explicitFailure] { + warm(); + auto connection = acquire(); + connection->setAutocommit(autocommit); + auto statement = connection->allocStatementHandle(); + startWork(statement); + const auto nativeStatement = statement->get(); + const auto dbc = handles.at(nativeStatement)->parent; + std::weak_ptr alias = statement; + if (explicitFailure) { + failNext = "free"; + require(statement->freeHandle() == SQL_ERROR, "Expected explicit free failure"); + } + failNext = "free"; + statement.reset(); + require(alias.expired(), "Wrapper did not expire"); + require(handles.count(nativeStatement) == 1, "Native free unexpectedly succeeded"); + calls.clear(); + expectFailure([&] { connection->close(); }); + expectCalls({"off", "rollback", "disconnect", "free"}); + require(handles.count(nativeStatement) == 0 && handles.count(dbc) == 0, + "Discard did not reclaim the orphan and its connection"); + require(commits == 0, "Orphan cleanup committed pending work"); + connection = acquire(); + require(logins == 2, "Orphan cleanup did not release the size-one pool capacity"); + connection->close(); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({}); + require(logins == 2, "Replacement connection did not retain the clean fast path"); + checkParked(); + }); + } + } + run("successful retry cannot clear a previous child-free failure", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + failNext = "free"; + require(statement->freeHandle() == SQL_ERROR, "Expected explicit free failure"); + require(statement->freeHandle() == SQL_SUCCESS, "Expected successful retry"); + statement.reset(); + calls.clear(); + expectFailure([&] { connection->close(); }); + expectCalls({"off", "rollback", "disconnect", "free"}); + connection = acquire(); + require(logins == 2, "Successful retry cleared the failure latch"); + connection->close(); + checkParked(); + }); + run("child-free failure while parked rejects the connection at checkout", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + const auto nativeStatement = statement->get(); + const auto dbc = handles.at(nativeStatement)->parent; + connection->close(); + failNext = "free"; + statement.reset(); + require(handles.count(nativeStatement) == 1, "Expected orphaned statement"); + connection = acquire(); + require(logins == 2, "Checkout reused a connection with failed child cleanup"); + require(handles.count(nativeStatement) == 0 && handles.count(dbc) == 0, + "Checkout did not reclaim the orphan and its connection"); + connection->close(); + checkParked(); + }); + run("child-free failure during sanitation still discards the connection", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + const auto nativeStatement = statement->get(); + const auto dbc = handles.at(nativeStatement)->parent; + { + CallbackScope callback(duringRollbackBatch, [&] { + failNext = "free"; + statement.reset(); + }); + expectFailure([&] { connection->close(); }); + } + require(handles.count(nativeStatement) == 0 && handles.count(dbc) == 0, + "Sanitation hid a failed destructor's orphan"); + require(commits == 0, "Sanitation failure committed pending work"); + connection = acquire(); + require(logins == 2, "Sanitation failure did not release capacity"); + connection->close(); + checkParked(); + }); + run("child-free failure during checkout reset rejects the candidate", [] { + warm(); + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + const auto nativeStatement = statement->get(); + const auto dbc = handles.at(nativeStatement)->parent; + connection->close(); + { + CallbackScope callback(duringReset, [&] { + failNext = "free"; + statement.reset(); + }); + connection = acquire(); + } + require(logins == 2, "Checkout reset hid a failed child free"); + require(handles.count(nativeStatement) == 0 && handles.count(dbc) == 0, + "Checkout reset did not reclaim the orphan and its connection"); + connection->close(); + checkParked(); + }); + for (const auto* operation : {"get", "reset", "allocate_statement", "rollback_batch", "free"}) { + run(operation, [operation] { + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + statement.reset(); + calls.clear(); + failNext = operation; + expectFailure([&] { connection->close(); }); + require(std::find(calls.begin(), calls.end(), "disconnect") != calls.end(), + "Failed sanitation did not disconnect"); + require(commits == 0, "Failure cleanup committed work"); + connection = acquire(); + require(logins == 2, "Discard did not release capacity / replace DBC"); + connection->close(); + checkParked(); + }); + } + for (const auto* operation : {"rollback", "on"}) { + run("manual-mode sanitation failure", [operation] { + auto connection = acquire(); + connection->setAutocommit(false); + calls.clear(); + failNext = operation; + expectFailure([&] { connection->close(); }); + require(std::find(calls.begin(), calls.end(), "disconnect") != calls.end(), + "Failed manual-mode sanitation did not disconnect"); + if (std::string(operation) == "rollback") { + require(std::find(calls.begin(), calls.end(), "on") == calls.end(), + "Enabled autocommit after failed rollback"); + } + require(commits == 0, "Failure cleanup committed work"); + connection = acquire(); + require(logins == 2, "Discard did not release capacity / replace DBC"); + connection->close(); + checkParked(); + }); + } + for (auto info : {SQL_DRIVER_HDBC, SQL_DRIVER_HENV, SQL_DRIVER_HSTMT, SQL_DRIVER_HLIB}) { + run("raw handle exposure permanently disables proof", [info] { + warm(); + auto connection = acquire(); + connection->getInfo(static_cast(info)); + connection->close(); + connection = acquire(); + calls.clear(); + connection->setAutocommit(true); + expectCalls({"on"}); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + run("ordinary getinfo invalidates the current lease only", [] { + warm(); + auto connection = acquire(); + connection->getInfo(SQL_DBMS_NAME); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({}); + }); + run("arbitrary attrs_before disable proof", [] { + py::dict attrs; + attrs[py::int_(SQL_ATTR_AUTOCOMMIT)] = py::int_(SQL_AUTOCOMMIT_OFF); + auto connection = acquire(true, attrs); + connection->close(); + connection = acquire(); + calls.clear(); + connection->setAutocommit(true); + expectCalls({"on"}); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + for (bool fail : {false, true}) { + run("generic set_attr including failure disables proof", [fail] { + warm(); + auto connection = acquire(); + if (fail) { + failNext = "attribute"; + expectFailure([&] { connection->setAttr(SQL_ATTR_LOGIN_TIMEOUT, py::int_(30)); }); + } else { + connection->setAttr(SQL_ATTR_AUTOCOMMIT, py::int_(SQL_AUTOCOMMIT_OFF)); + } + calls.clear(); + connection->close(); + if (fail) { + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + } else { + expectCalls({"get", "rollback", "on"}); + } + connection = acquire(); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + for (const auto* operation : {"commit", "rollback", "on", "off", "get"}) { + run("failed native operation invalidates clean proof", [operation] { + warm(); + auto connection = acquire(); + if (std::string(operation) == "on") { + // A proven ON->ON is deliberately skipped; inject failure + // only after an operation invalidates that proof. + auto statement = connection->allocStatementHandle(); + } + failNext = operation; + expectFailure([&] { + if (std::string(operation) == "commit") { + connection->commit(); + } else if (std::string(operation) == "rollback") { + connection->rollback(); + } else if (std::string(operation) == "get") { + connection->getAutocommit(); + } else { + connection->setAutocommit(std::string(operation) == "on"); + } + }); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + for (const auto* operation : {"reset", "isolation"}) { + run("failed deferred reset replaces the physical connection", [operation] { + auto connection = acquire(); + connection->setAutocommit(false); + connection->close(); + failNext = operation; + connection = acquire(); + require(logins == 2, "Failed reset was reused"); + calls.clear(); + connection->close(); + expectCalls({"get", "reset", "allocate_statement", "rollback_batch", "free"}); + }); + } + run("abandonment rolls back and releases capacity", [] { + auto connection = acquire(); + auto statement = connection->allocStatementHandle(); + startWork(statement); + connection.reset(); + require(commits == 0, "Abandoned connection committed work"); + connection = acquire(); + require(logins == 2, "Abandonment did not release capacity"); + connection->close(); + checkParked(); + }); + for (bool autocommit : {true, false}) { + run("unpooled close does not run pooled sanitation", [autocommit] { + auto connection = acquire(false); + connection->setAutocommit(autocommit); + calls.clear(); + connection->close(true); + if (autocommit) { + expectCalls({"get", "disconnect", "free"}); + } else { + expectCalls({"get", "rollback", "disconnect", "free"}); + } + }); + } + for (const auto* operation : {"get", "rollback"}) { + run("unpooled failure discards without pool sanitation", [operation] { + auto connection = acquire(false); + connection->setAutocommit(false); + failNext = operation; + expectFailure([&] { connection->close(true); }); + require(std::find(calls.begin(), calls.end(), "disconnect") != calls.end(), + "Unpooled failure did not disconnect"); + require(commits == 0, "Unpooled error cleanup committed work"); + }); + } + run("raw unpooled disconnect failure preserves connection and children", [] { + auto connection = acquire(false); + connection->setAutocommit(false); + auto statement = connection->allocStatementHandle(); + startWork(statement); + calls.clear(); + expectFailure([&] { connection->close(); }); + expectCalls({"disconnect"}); + require(handles.count(statement->get()) == 1, "Live child was freed"); + auto sibling = connection->allocStatementHandle(); + connection->rollback(); + connection->close(); + require(statement->isImplicitlyFreed() && sibling->isImplicitlyFreed(), + "Successful disconnect did not invalidate children"); + }); + ConnectionPoolManager::getInstance().closePools(); + std::cout << passed << " native pool tests passed\n"; + return 0; + } catch (const std::exception& error) { + std::cerr << "FAIL after " << passed << " tests: " << error.what() << '\n'; + failNext.clear(); + ConnectionPoolManager::getInstance().closePools(); + return 1; + } +} diff --git a/tests/test_006_exceptions.py b/tests/test_006_exceptions.py index 371b9662f..318e3b0e6 100644 --- a/tests/test_006_exceptions.py +++ b/tests/test_006_exceptions.py @@ -282,7 +282,7 @@ def test_close_cleans_up_after_rollback_failure(): mock_conn = MagicMock() mock_conn.get_autocommit.return_value = False - mock_conn.rollback.side_effect = RuntimeError("SQLSTATE:08S01:Communication link failure") + mock_conn.close.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;") @@ -290,8 +290,8 @@ def test_close_cleans_up_after_rollback_failure(): 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) + mock_conn.rollback.assert_not_called() + mock_conn.close.assert_called_once_with(rollback_before_disconnect=True) assert conn._conn is None assert conn.closed @@ -301,7 +301,7 @@ def test_close_cleans_up_after_autocommit_read_failure(): from unittest.mock import MagicMock, patch mock_conn = MagicMock() - mock_conn.get_autocommit.side_effect = RuntimeError("SQLSTATE:08S01:Communication link failure") + mock_conn.close.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;") @@ -310,13 +310,14 @@ def test_close_cleans_up_after_autocommit_read_failure(): conn.close() mock_conn.rollback.assert_not_called() - mock_conn.close.assert_called_once_with(False) + mock_conn.get_autocommit.assert_not_called() + mock_conn.close.assert_called_once_with(rollback_before_disconnect=True) 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.""" +def test_close_delegates_manual_transaction_cleanup_to_native(): + """Only native close probes the mode and rolls back, without Python hints.""" from unittest.mock import MagicMock, patch mock_conn = MagicMock() @@ -327,8 +328,9 @@ def test_close_reports_successful_rollback_to_native_cleanup(): conn.close() - mock_conn.rollback.assert_called_once_with() - mock_conn.close.assert_called_once_with(True) + mock_conn.get_autocommit.assert_not_called() + mock_conn.rollback.assert_not_called() + mock_conn.close.assert_called_once_with(rollback_before_disconnect=True) def test_autocommit_close_delegates_transaction_cleanup_to_native(): @@ -347,12 +349,13 @@ def test_autocommit_close_delegates_transaction_cleanup_to_native(): conn.close() mock_conn.rollback.assert_not_called() - mock_conn.close.assert_called_once_with(False) + mock_conn.get_autocommit.assert_not_called() + mock_conn.close.assert_called_once_with(rollback_before_disconnect=True) @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.""" +def test_native_close_error_does_not_need_python_preclose_calls(preclose_failure): + """Native cleanup is authoritative even if Python probes would have failed.""" from unittest.mock import MagicMock, patch mock_conn = MagicMock() @@ -369,11 +372,33 @@ def test_native_close_error_takes_precedence_over_preclose_failure(preclose_fail with pytest.raises(OperationalError, match="Native close failed"): conn.close() - mock_conn.close.assert_called_once_with(False) + mock_conn.close.assert_called_once_with(rollback_before_disconnect=True) + mock_conn.get_autocommit.assert_not_called() + mock_conn.rollback.assert_not_called() assert conn._conn is None assert conn.closed +def test_cursor_cleanup_failure_still_calls_native_close(): + from unittest.mock import MagicMock, patch + + native = MagicMock() + with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=native): + conn = connect("Server=testserver;Database=mydb;Trusted_Connection=yes;") + broken = MagicMock(closed=False) + broken.close.side_effect = RuntimeError("cursor cleanup failed") + healthy = MagicMock(closed=False) + conn._cursors = {broken, healthy} + + conn.close() + + broken.close.assert_called_once_with() + healthy.close.assert_called_once_with() + native.close.assert_called_once_with(rollback_before_disconnect=True) + assert conn.closed + assert conn._conn is None + + 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 28b7465bc..ff9c7e317 100644 --- a/tests/test_009_pooling.py +++ b/tests/test_009_pooling.py @@ -83,6 +83,33 @@ def reset_pooling_state(): # ============================================================================= +def test_pooled_positive_login_timeout_reaches_native_attrs(): + """Exercise the Python boundary paired with native timeout call-count tests.""" + from unittest.mock import MagicMock, patch + from mssql_python.constants import ConstantsDDBC + + pooling(enabled=True, max_size=2, idle_timeout=30) + native = MagicMock() + with patch("mssql_python.connection.ddbc_bindings.Connection", return_value=native) as create: + for _ in range(3): + connection = connect( + "Server=testserver;Database=mydb;Trusted_Connection=yes;", + timeout=30, + autocommit=True, + ) + assert create.call_args.args[1] is True + assert create.call_args.args[2] == {ConstantsDDBC.SQL_ATTR_LOGIN_TIMEOUT.value: 30} + assert connection.timeout == 0 # Query timeout is independent. + connection.close() + assert create.call_count == 3 + assert native.set_autocommit.call_count == 3 + native.set_autocommit.assert_called_with(True) + assert native.close.call_count == 3 + native.close.assert_called_with(rollback_before_disconnect=True) + native.get_autocommit.assert_not_called() + native.rollback.assert_not_called() + + def test_connection_pooling_basic(conn_str): """Test basic connection pooling functionality with multiple connections.""" # Enable pooling with small pool size @@ -172,6 +199,13 @@ def open_transaction_count(session_id): None, None, ), + ( + "explicit transaction with noexec", + True, + "BEGIN TRANSACTION; SELECT 1; SET NOEXEC ON", + None, + None, + ), ) expected_spid = None for name, autocommit, sql, params, action in scenarios: @@ -224,7 +258,8 @@ def open_transaction_count(session_id): ) -def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str): +@pytest.mark.parametrize("execution", ["execute", "executemany", "failed_execute"]) +def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str, execution): """Autocommit normalization must not commit an explicit SQL transaction.""" _run_in_subprocess( """ @@ -233,7 +268,8 @@ def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str import mssql_python conn_str = os.environ["DB_CONNECTION_STRING"] - table = "pytest_pool_explicit_autocommit_transaction" + execution = EXECUTION_MODE + table = f"pytest_pool_explicit_autocommit_transaction_{execution}" mssql_python.pooling(enabled=True, max_size=2, idle_timeout=30) observer = mssql_python.connect(conn_str, autocommit=True) try: @@ -271,7 +307,38 @@ def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str ) sys.exit(77) - subject_cursor.execute(f"BEGIN TRANSACTION; INSERT INTO {table} VALUES (1)") + subject_cursor.close() + subject.close() + # Exercise proven-clean, no-statement leases before reintroducing + # explicit work. Inspect the parked session without borrowing it. + for _ in range(3): + empty = mssql_python.connect(conn_str, autocommit=True) + empty.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 and row[0] == 0 + + subject = mssql_python.connect(conn_str, autocommit=True) + subject_cursor = subject.cursor() + subject_cursor.execute("BEGIN TRANSACTION") + if execution == "executemany": + subject_cursor.executemany( + f"INSERT INTO {table} VALUES (?)", [(1,), (2,)] + ) + elif execution == "failed_execute": + subject_cursor.execute(f"INSERT INTO {table} VALUES (1)") + try: + subject_cursor.execute(f"INSERT INTO {table} VALUES (1)") + except mssql_python.IntegrityError: + pass + else: + raise AssertionError("Expected duplicate-key execution failure") + else: + subject_cursor.execute(f"INSERT INTO {table} VALUES (1)") subject_cursor.close() subject.close() @@ -303,7 +370,7 @@ def test_autocommit_explicit_transaction_is_rolled_back_on_pool_checkin(conn_str finally: observer.close() mssql_python.pooling(enabled=False) - """, + """.replace("EXECUTION_MODE", repr(execution)), conn_str, ) diff --git a/tests/test_fetch_settings_cache.py b/tests/test_fetch_settings_cache.py index 78ad5229b..4cec693d3 100644 --- a/tests/test_fetch_settings_cache.py +++ b/tests/test_fetch_settings_cache.py @@ -943,8 +943,8 @@ def _check_native_mixed_fetch_diagnostics(mode, expected_native): ] all_records = [(f"[{state}] ({number})", message) for state, number, message in records] wanted = [all_records[index] for index in (0, 2, 4)] - rec_calls, field_calls, callback_errors = [], [], [] - observed_handles, delegated_records = [], [] + rec_calls, field_calls, header_calls, callback_errors = [], [], [], [] + observed_handles, delegated_records, delegated_headers = [], [], [] phase, target_handle = "observe", None def guarded(callback_type): @@ -994,13 +994,31 @@ def read_record(handle_type, handle, number, state, native_error, message, capac return success @guarded(field_type) - def read_state(handle_type, handle, number, identifier, output, capacity, length): + def read_field(handle_type, handle, number, identifier, output, capacity, length): if phase != "inject" or (handle_type, handle) != target_handle: - if original_read_state is None: + if original_read_field is None: return error - return original_read_state( + result = original_read_field( handle_type, handle, number, identifier, output, capacity, length ) + if identifier == 2 and output: # SQL_DIAG_NUMBER is a SQLINTEGER header field. + count = ctypes.cast(output, ctypes.POINTER(ctypes.c_int32))[0] + delegated_headers.append((handle_type, handle, number, result, count)) + return result + if identifier == 2: + header_calls.append(number) + if ( + not handle + or handle_type != ConstantsDDBC.SQL_HANDLE_STMT.value + or number != 0 + or not output + or capacity != 0 + or length + ): + callback_errors.append("invalid diagnostic-count header lookup") + return error + ctypes.cast(output, ctypes.POINTER(ctypes.c_int32))[0] = len(records) + return success field_calls.append(number) # SQL_DIAG_SQLSTATE uses bytes, including the sixth SQLWCHAR terminator. if ( @@ -1036,7 +1054,7 @@ def read_state(handle_type, handle, number, identifier, output, capacity, length original_rec, original_field = rec_pointer.value, field_pointer.value assert original_rec, "Driver diagnostic records must be available" original_read_record = rec_type(original_rec) - original_read_state = field_type(original_field) if original_field else None + original_read_field = field_type(original_field) if original_field else None try: rec_pointer.value = ctypes.cast(read_record, ctypes.c_void_p).value # Learn this cursor's raw handle while forwarding the diagnostic call unchanged. @@ -1047,7 +1065,7 @@ def read_state(handle_type, handle, number, identifier, output, capacity, length assert target_handle[0] == ConstantsDDBC.SQL_HANDLE_STMT.value and target_handle[1] phase = "inject" field_pointer.value = ( - None if mode == "missing" else ctypes.cast(read_state, ctypes.c_void_p).value + None if mode == "missing" else ctypes.cast(read_field, ctypes.c_void_p).value ) delegated_records.clear() assert mssql_python.ddbc_bindings.DDBCSQLGetAllDiagRecords(other_cursor.hstmt) == [] @@ -1057,6 +1075,7 @@ def read_state(handle_type, handle, number, identifier, output, capacity, length assert tuple(cursor.fetchone()) == (b"x" * 8193,) assert not callback_errors, callback_errors assert cursor.messages == wanted + assert header_calls == ([] if mode == "missing" else [0]) assert field_calls == ([] if mode == "missing" else [1, 2, 3, 4, 5, 6]) if mode == "available": assert rec_calls == [1, 3, 5] @@ -1065,15 +1084,28 @@ def read_state(handle_type, handle, number, identifier, output, capacity, length rec_calls.clear() field_calls.clear() + header_calls.clear() assert mssql_python.ddbc_bindings.DDBCSQLGetAllDiagRecords(cursor.hstmt) == all_records assert not callback_errors, callback_errors assert rec_calls == [1, 2, 3, 4, 5, 6] assert field_calls == [] + assert header_calls == [] delegated_records.clear() + delegated_headers.clear() phase = "delegate" assert cursor.fetchone() is None assert not callback_errors, callback_errors - assert delegated_records == [(*target_handle, 1)] + if mode == "missing": + assert delegated_headers == [] + assert delegated_records == [(*target_handle, 1)] + else: + assert len(delegated_headers) == 1 + handle_type, handle, number, result, count = delegated_headers[0] + assert (handle_type, handle, number) == (*target_handle, 0) + if result == success and count == 0: + assert delegated_records == [] + else: + assert delegated_records == [(*target_handle, 1)] assert cursor.messages == wanted finally: rec_pointer.value, field_pointer.value = original_rec, original_field