diff --git a/docs/BACKLOG.md b/docs/BACKLOG.md index 83634517..2ef96a56 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()