Pin AsyncAdaptedConnection.close OS-level suppression branches

The narrow try/except on close() rollback suppresses transport
failures so a leader flip or peer reset does not leak out of the
pool. The tuple covered the dbapi-level types (OperationalError,
InterfaceError, DqliteConnectionError) plus OSError, TimeoutError,
and ConnectionError — but only the dbapi-level types had direct
tests, leaving the three OS-level branches unpinned. A refactor
narrowing the tuple (e.g. dropping OSError) would silently re-raise
real transport failures and leak the underlying AsyncConnection.

Add a parametrized test across OSError, BrokenPipeError,
ConnectionError, ConnectionResetError, and TimeoutError that
asserts the DEBUG log breadcrumb fires and close() still runs. Add
an inverse test that ValueError (outside the tuple) propagates.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

tests/test_aio_close_rollback_log.py

@@ -124,3 +124,79 @@ def _fake_await_only(coro: object) -> object:
     ]
     assert matching
     assert fake.close_calls == 1
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [
+        OSError(32, "broken pipe"),
+        BrokenPipeError(32, "broken pipe"),
+        ConnectionError("peer went away"),
+        ConnectionResetError(104, "connection reset by peer"),
+        TimeoutError("read timed out"),
+    ],
+)
+def test_close_suppresses_os_level_rollback_errors(
+    caplog: pytest.LogCaptureFixture, exc: BaseException
+) -> None:
+    """The narrow suppression tuple on ``close()`` includes OSError,
+    TimeoutError, and ConnectionError alongside the dbapi-level
+    types. Pin each OS-level branch so a refactor dropping any one
+    of them would fail this test rather than silently re-raising
+    the exception and leaking the underlying AsyncConnection.
+    """
+    fake = _FakeAsyncConn(exc)
+    adapter = AsyncAdaptedConnection(fake)  # type: ignore[arg-type]
+
+    from sqlalchemydqlite import aio as aio_module
+
+    def _fake_await_only(coro: object) -> object:
+        import asyncio
+
+        return asyncio.new_event_loop().run_until_complete(coro)  # type: ignore[arg-type]
+
+    orig = aio_module.await_only
+    aio_module.await_only = _fake_await_only  # type: ignore[assignment]
+    try:
+        with caplog.at_level(logging.DEBUG, logger="sqlalchemydqlite.aio"):
+            adapter.close()
+    finally:
+        aio_module.await_only = orig  # type: ignore[assignment]
+
+    matching = [
+        r
+        for r in caplog.records
+        if r.levelno == logging.DEBUG and "rollback failed" in r.getMessage()
+    ]
+    assert matching, f"no DEBUG log captured for {type(exc).__name__}: {caplog.records!r}"
+    # Log line carries the exception type (so operators triaging a
+    # noisy pool can correlate the cause without enabling exc_info
+    # rendering). ``%s`` with ``type(exc).__name__`` in the format.
+    assert type(exc).__name__ in matching[0].getMessage()
+    # close() ran regardless of which transport error fired the
+    # suppression branch.
+    assert fake.close_calls == 1
+
+
+def test_close_propagates_value_error_out_of_tuple() -> None:
+    """Inverse pin: a ``ValueError`` from rollback is not in the
+    narrow tuple and must propagate. Guards against a refactor
+    widening the suppression back to ``except Exception``.
+    """
+    fake = _FakeAsyncConn(ValueError("parameter out of range"))
+    adapter = AsyncAdaptedConnection(fake)  # type: ignore[arg-type]
+
+    from sqlalchemydqlite import aio as aio_module
+
+    def _fake_await_only(coro: object) -> object:
+        import asyncio
+
+        return asyncio.new_event_loop().run_until_complete(coro)  # type: ignore[arg-type]
+
+    orig = aio_module.await_only
+    aio_module.await_only = _fake_await_only  # type: ignore[assignment]
+    try:
+        with pytest.raises(ValueError, match="parameter out of range"):
+            adapter.close()
+    finally:
+        aio_module.await_only = orig  # type: ignore[assignment]