fix(transports): close DATABASE cursors before releasing the connection (BACKLOG #1104)

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.

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
 

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()