From 383383164704e06d6cae93241881a880a9e0bcdd Mon Sep 17 00:00:00 2001 From: wshallwshall Date: Sat, 8 Aug 2026 16:36:00 -0500 Subject: [PATCH] fix(transports): close DATABASE cursors before releasing the connection BACKLOG #1104. database.py opened a cursor at five sites and closed it at none -- cur.close() appeared nowhere in the file. aioodbc/pyodbc keep the ODBC statement handle open until the cursor is closed, so each of those connections returned to the pool BUSY and the next caller's first command failed with HY000 Connection is busy with results for another command This is delivery semantics, not tidiness. An UPDATE leaves a row count pending, so the DATABASE source's mark is the usual victim, and _poll_once treats a failed mark as at-least-once: the row is left unmarked and RE-EMITTED AS A DUPLICATE. Observed on main against a real SQL Server 2022 container: DATABASE source mark failed (row will re-emit, a duplicate): ('HY000', ...) FAILED tests/test_database_source_integration.py::test_source_polls_and_marks_rows assert [(1, 1)] == [(0, 2)] THE ERROR APPEARS ON THE INNOCENT STATEMENT. The command that fails is not the one that left the handle open; it is whatever next draws that connection. That misdirection is why this survived, and it is why the fix is at all five sites rather than only the one that was seen to fail. _close_cursor never raises: a close failure must not mask the caller's real error and must not skip the release that follows it, because leaking a pooled connection to save a cursor is the worse trade. EVIDENCE, and the integration half of it is weak on its own. On the container the unfixed tree failed 1 run in 10 and the fixed tree 0 in 10 -- at a ~10% base rate that is well inside chance and proves nothing. It is recorded as the reproduction that FOUND the defect, not as proof it is fixed. The proof is tests/test_database_cursor_close.py, which asserts close-before-release deterministically against a fake pool, and which was verified to go RED on a mutant with the closes removed (2 of 3 failed; the third covers _close_cursor's own contract and correctly did not). Verified: ruff, ruff format, mypy strict (263 files) all clean; 236 tests pass against a live SQL Server 2022 container, including the previously-failing test_source_polls_and_marks_rows; backlog hygiene 19 passed; the repo-wide link gate reports 5,359 links across 347 files all resolving. Found while triaging PR #253's red SQL Server leg. #253 was exonerated by measurement -- the same test fails identically on main -- after first being exonerated by mechanism. Verified against the container only after confirming the host reaches it and not the native MSSQLSERVER service also running on this box: both listeners on 1433 were Docker processes and SERVERPROPERTY('MachineName') returned the container's own hostname. --- docs/BACKLOG.md | 62 +++++++++++ messagefoundry/transports/database.py | 32 ++++++ tests/test_database_cursor_close.py | 141 ++++++++++++++++++++++++++ 3 files changed, 235 insertions(+) create mode 100644 tests/test_database_cursor_close.py diff --git a/docs/BACKLOG.md b/docs/BACKLOG.md index 66c059b4..47b8d8c4 100644 --- a/docs/BACKLOG.md +++ b/docs/BACKLOG.md @@ -6955,3 +6955,65 @@ covered -- the comment quoted above is that shape in prose). `#1096` cap change. Diagnosed by tracing the port number through `_free_port` and the `api_port + step` call site rather than by re-running: the failing port lies outside the inbound window, which is what rules out the family `#1014` already fixed and points at the one it did not. + +## 1104. the DATABASE connector never closes its cursors, so a pooled connection is returned busy and the source's mark fails, emitting a duplicate + +> ✅ **SHIPPED 2026-08-08 — found and fixed in the same pass; reproduced against a real SQL Server 2022 container, not inferred.** Value **6/10** · Difficulty **2/10**. `messagefoundry/transports/database.py` opened a cursor at **five** sites and closed it at **none** — `cur.close()` appeared nowhere in the file. aioodbc/pyodbc keep the ODBC statement handle open until the cursor is closed, so every one of those connections went back to the pool **busy**, and the next caller's first command failed with `HY000 Connection is busy with results for another command`. **This is delivery semantics, not tidiness:** the usual victim is the DATABASE source's `mark`, and `_poll_once` treats a failed mark as at-least-once — the row is left unmarked and **re-emitted as a DUPLICATE**. + +**Cluster:** Connectors / delivery semantics. **Priority:** P2. **Verdict:** built. **Severity:** no PHI +effect. A shipped connector emits duplicate messages on SQL Server whenever the pool hands back a dirty +connection at the wrong moment. Per CLAUDE.md §0 this is stated in the conditional: **a deploying site +running a DATABASE source against SQL Server would see duplicates**, at a rate set by pool reuse. + +**Observed on `main`**, not on a branch: + +``` +DATABASE source mark failed (row will re-emit, a duplicate): + ('HY000', '[Microsoft][ODBC Driver 18 for SQL Server]Connection is busy with + results for another command (0) (SQLExecDirectW)') +FAILED tests/test_database_source_integration.py::test_source_polls_and_marks_rows +assert [(1, 1)] == [(0, 2)] # 1 row left unmarked -> it re-emits +``` + +**Mechanism.** `_select` runs the poll and `_mark` runs an `UPDATE`; both release the connection in a +`finally` without closing the cursor. An `UPDATE` leaves a row count pending on the statement handle, +so the connection is dirty when it returns to the pool. The failure then lands on **whatever statement +next draws that connection**, which is why it reads as unrelated and intermittent. + +**⚠️ THE ERROR APPEARS ON THE INNOCENT STATEMENT.** The command that fails is not the one that left the +handle open. Triaging the reported statement leads nowhere; the cause is one connection-checkout +earlier. That misdirection is the whole reason this survived. + +**Why CI never caught it on `main`.** The `sql server (store + connector)` leg is gated on server-DB and +docker path changes, so it is **skipped on every `main` push** — measured across the five most recent. +It runs only on PRs that touch those paths, which is how a real defect sat on `main` while the leg that +detects it stayed green-by-absence. That is the #1000 shape at the workflow level: a check whose silence +is mistaken for a pass. **Filing this does not fix that**; the leg's `main` coverage is a separate +question and is NOT addressed here. + +**The fix.** A `_close_cursor` helper, called before `pool.release` at all five sites. It never raises: +a close failure must not mask the caller's real error, and must not skip the release that follows — +leaking a pooled connection to save a cursor is the worse trade. + +**⚠️ BE HONEST ABOUT THE INTEGRATION EVIDENCE — IT IS WEAK ON ITS OWN.** Measured on the container: +**1 failure in 10 runs** on the unfixed tree, **0 in 10** with the fix. At a ~10% base rate that +difference is **well inside chance** and proves nothing by itself. It is recorded as the reproduction +that found the defect, not as the evidence that it is fixed. The evidence is +`tests/test_database_cursor_close.py`, which asserts the ordering **deterministically** against a fake +pool and was **verified to go RED on a mutant** with the closes removed (2 of 3 tests failed; the third +covers `_close_cursor`'s own contract and correctly did not). A guard with a 10% detection rate is not +a guard. + +**Related:** #1000 (a control green because its evidence could not see the class it covered — both the +skipped CI leg and the racy integration test are that shape), #1103 (found the same day, also a harness +/ connector defect whose error message points away from the cause), ADR 0003 (the aioodbc choice this +rides on). + +**Source:** found 2026-08-08 while triaging PR #253's red SQL Server leg. #253 was exonerated **by +measurement** — the same test fails identically on `main` — after first being exonerated by mechanism +(that step runs an explicit path list, so `testpaths` cannot reach it). The two reds on #253 were two +*different* unrelated failures, which is why "it failed twice, so it is real" would have been the wrong +read. Verified against a Docker SQL Server 2022 container after first confirming the host actually +reaches the container and not the native `MSSQLSERVER` service also running on that box: both listeners +on 1433 were Docker processes, and `SERVERPROPERTY('MachineName')` returned the container's own +hostname. That check is not optional on this machine. diff --git a/messagefoundry/transports/database.py b/messagefoundry/transports/database.py index 1d761cf7..bcc80079 100644 --- a/messagefoundry/transports/database.py +++ b/messagefoundry/transports/database.py @@ -478,6 +478,28 @@ async def _acquire(pool: Any, timeout: float) -> Any: ) from exc +async def _close_cursor(cur: Any) -> None: + """Close a cursor BEFORE its connection goes back to the pool (BACKLOG #1104). + + aioodbc/pyodbc keep the ODBC statement handle open until the cursor is closed, and a connection + released with an open handle is handed to the NEXT caller still busy. That caller's first command + then fails with ``HY000 Connection is busy with results for another command`` — a failure with no + relationship to the statement it lands on, which is what made this hard to attribute. An ``UPDATE`` + leaves a row count pending, so the DATABASE source's ``mark`` is the usual victim, and a failed + mark **re-emits the row as a DUPLICATE** (see ``_poll_once``). So this is delivery semantics, not + tidiness. + + Never raises. A close failure must not mask the caller's real error, and must not skip the pool + release that follows it — leaking a connection to save a cursor would be the worse trade. + """ + if cur is None: + return + try: + await cur.close() + except Exception as exc: # noqa: BLE001 - hygiene only; the caller's outcome always wins + logger.debug("DATABASE: cursor close failed, ignored: %s", exc) + + async def _probe_db( get_pool: Callable[[], Any], *, timeout: float = _DEFAULT_DB_ACQUIRE_TIMEOUT ) -> None: @@ -496,6 +518,7 @@ async def _probe_db( if state else DeliveryError(f"DATABASE connect failed: {exc}") ) from exc + cur: Any = None try: cur = await conn.cursor() await cur.execute("SELECT 1") @@ -507,6 +530,7 @@ async def _probe_db( else DeliveryError(f"DATABASE probe failed: {exc}") ) from exc finally: + await _close_cursor(cur) await pool.release(conn) @@ -591,6 +615,7 @@ async def send( params = _bind_params(payload, self._param_names) # NegativeAckError(permanent) on bad data pool = await self._get_pool() conn = await _acquire(pool, self._acquire_timeout) + cur: Any = None try: cur = await conn.cursor() try: @@ -607,6 +632,7 @@ async def send( raise # not a DB driver error → an internal/code error, let the runner handle it raise _classify_db_error(state, str(exc)) from exc finally: + await _close_cursor(cur) await pool.release(conn) return captured @@ -890,12 +916,14 @@ async def _select(self) -> tuple[list[str], list[Any]]: hostage to downstream store I/O.""" pool = await self._get_pool() conn = await _acquire(pool, self._acquire_timeout) + cur: Any = None try: cur = await conn.cursor() await cur.execute(self._poll_sql) columns = [d[0] for d in cur.description] rows = list(await cur.fetchall()) finally: + await _close_cursor(cur) await pool.release(conn) return columns, rows @@ -928,10 +956,12 @@ async def _mark(self, record: dict[str, Any]) -> None: return pool = await self._get_pool() conn = await _acquire(pool, self._acquire_timeout) + cur: Any = None try: cur = await conn.cursor() await cur.execute(self._mark_sql, params) finally: + await _close_cursor(cur) await pool.release(conn) async def test_connection(self) -> None: @@ -1038,6 +1068,7 @@ async def query( # Map the transient pool-timeout onto the lookup's own PHI-free error type so the transform # worker dead-letters/errors this message consistently with other lookup failures. raise DbLookupError(f"db_lookup on {connection!r}: {exc}") from exc + cur: Any = None try: cur = await conn.cursor() await cur.execute(sql, bound) @@ -1052,6 +1083,7 @@ async def query( f"db_lookup query on {connection!r} failed" + (f" [{state}]" if state else "") ) from exc finally: + await _close_cursor(cur) await pool.release(conn) return [dict(zip(columns, row)) for row in rows] # noqa: B905 diff --git a/tests/test_database_cursor_close.py b/tests/test_database_cursor_close.py new file mode 100644 index 00000000..7f050807 --- /dev/null +++ b/tests/test_database_cursor_close.py @@ -0,0 +1,141 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Organization and contributors +"""BACKLOG #1104: a pooled connection must never go back to the pool with an open cursor. + +aioodbc/pyodbc keep the ODBC statement handle open until the cursor is closed. A connection released +with an open handle is handed to the NEXT caller still busy, and that caller's first command fails +with ``HY000 Connection is busy with results for another command`` -- attributed to whatever statement +happens to land on the recycled connection, not to the one that left it dirty. + +**Why this is delivery semantics, not tidiness.** An ``UPDATE`` leaves a row count pending, so the +DATABASE source's ``mark`` is the usual victim, and ``_poll_once`` treats a failed mark as +at-least-once: the row is left unmarked and **re-emitted as a duplicate** on the next poll. Observed +on a real SQL Server 2022 container against ``main``: + + DATABASE source mark failed (row will re-emit, a duplicate): + ('HY000', '[Microsoft][ODBC Driver 18 for SQL Server]Connection is busy with results + for another command') + assert [(1, 1)] == [(0, 2)] # one row unmarked -> it will re-emit + +**These tests are deterministic ON PURPOSE.** The integration test that first exposed this is racy -- +it depends on pool reuse order, and measured 1 failure in 10 runs on the unfixed tree. A guard with a +~10% detection rate is not a guard, and "0 failures in 10 runs after the fix" is not evidence either: +at that base rate the difference is well inside chance. So the invariant is asserted directly, by +recording the ORDER of operations against a fake pool, where it holds or fails every time. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from messagefoundry.transports import database as db + + +class _FakeCursor: + def __init__(self, log: list[str], tag: str) -> None: + self._log, self._tag = log, tag + self.description: list[tuple[str, Any]] = [("id", None)] + + async def execute(self, *_a: Any, **_k: Any) -> None: + self._log.append(f"execute:{self._tag}") + + async def fetchall(self) -> list[Any]: + return [] + + async def close(self) -> None: + self._log.append(f"close:{self._tag}") + + +class _FakeConn: + def __init__(self, log: list[str], tag: str) -> None: + self._log, self._tag = log, tag + + async def cursor(self) -> _FakeCursor: + return _FakeCursor(self._log, self._tag) + + async def commit(self) -> None: + self._log.append("commit") + + async def rollback(self) -> None: # pragma: no cover - not exercised here + self._log.append("rollback") + + +class _FakePool: + """Records acquire/release so the ORDER of close vs release is checkable.""" + + def __init__(self, log: list[str], tag: str) -> None: + self._log, self._tag = log, tag + + async def acquire(self) -> _FakeConn: + self._log.append("acquire") + return _FakeConn(self._log, self._tag) + + async def release(self, _conn: Any) -> None: + self._log.append("release") + + +def _assert_closed_before_released(log: list[str]) -> None: + """Every release must be immediately preceded by a close of that connection's cursor.""" + assert "release" in log, f"the connection was never released; log={log}" + for i, op in enumerate(log): + if op == "release": + assert i > 0 and log[i - 1].startswith("close:"), ( + "a pooled connection was released WITHOUT closing its cursor -- the next caller " + f"receives it busy and fails with HY000. log={log}" + ) + + +@pytest.mark.asyncio +async def test_source_select_closes_its_cursor_before_release() -> None: + src = object.__new__(db.DatabaseSource) + log: list[str] = [] + pool = _FakePool(log, "select") + src._get_pool = lambda: _pool_coro(pool) # type: ignore[method-assign,assignment] + src._acquire_timeout = 5.0 # type: ignore[attr-defined] + src._poll_sql = "SELECT 1" # type: ignore[attr-defined] + + await src._select() + assert "close:select" in log, f"the poll cursor was never closed; log={log}" + _assert_closed_before_released(log) + + +@pytest.mark.asyncio +async def test_source_mark_closes_its_cursor_before_release() -> None: + """The mark is the site that actually bites: an UPDATE leaves a row count pending.""" + src = object.__new__(db.DatabaseSource) + log: list[str] = [] + pool = _FakePool(log, "mark") + src._get_pool = lambda: _pool_coro(pool) # type: ignore[method-assign,assignment] + src._acquire_timeout = 5.0 # type: ignore[attr-defined] + src._mark_sql = "UPDATE t SET done=1 WHERE id=?" # type: ignore[attr-defined] + src._mark_names = ("id",) # type: ignore[attr-defined] + + await src._mark({"id": 1}) + assert "close:mark" in log, f"the mark cursor was never closed; log={log}" + _assert_closed_before_released(log) + + +@pytest.mark.asyncio +async def test_close_cursor_never_raises_and_never_skips_the_release() -> None: + """A close failure must not mask the caller's error, nor strand the connection. + + Leaking a pooled connection to save a cursor would be the worse trade, so ``_close_cursor`` + swallows. This pins that it swallows -- if it ever propagates, the ``finally`` above it stops + releasing and the pool drains to nothing under a flapping driver. + """ + + class _Boom: + async def close(self) -> None: + raise RuntimeError("driver went away mid-close") + + await db._close_cursor(_Boom()) # must not raise + await db._close_cursor(None) # the never-opened case + + +def _pool_coro(pool: _FakePool) -> Any: + async def _c() -> _FakePool: + return pool + + return _c()