fix: preserve stream and event configuration contracts

Author: cofin

docs/changelog.rst

@@ -21,6 +21,18 @@ Unreleased
   return result envelopes, object DDL lookups return ``DDLResult`` directly, and
   callers should inspect capability or DDL status instead of treating empty
   lists as unsupported metadata.
+* Standardized event transport configuration on ``notify``, ``notify_queue``,
+  ``poll_queue``, ``aq``, and ``txeventq``. Retired transport names now raise
+  an explicit configuration error with the canonical replacement.
+
+**Fixed:**
+
+* Event channels now honor adapter ``events_backend`` driver features when no
+  extension-level backend is configured.
+* Early mysql-connector row-stream cleanup now consumes unread results without
+  reconnecting underneath an active transaction.
+* Public row streams continue to clean up duck-typed sources whose ``close()``
+  method uses the original no-argument contract.
 
 **Docs:**
 

docs/reference/extensions/events.rst

@@ -22,6 +22,11 @@ The durable queue is the source of truth for ``notify_queue``; native
 notifications only prompt consumers to check it. Durable event queues are not
 browser fan-out transports.
 
+Set ``extension_config["events"]["backend"]`` to select the transport. The
+adapter ``driver_features["events_backend"]`` value is used only when the
+extension setting is absent. Retired transport names fail with an explicit
+canonical replacement instead of silently changing delivery semantics.
+
 ``polling`` is not a SQLSpec backend name. Litestar Queues uses it for the
 fallback worker mode where no push wakeup transport is available and the
 worker waits for its configured polling interval.

sqlspec/adapters/mysqlconnector/core.py

@@ -232,16 +232,15 @@ def fetch_chunk(self) -> "list[dict[str, Any]]":
     def close(self, error: bool = False) -> None:
         cursor = self._cursor
         self._cursor = None
-        if cursor is not None:
-            with contextlib.suppress(Exception):
-                cursor.close()
         connection = self._driver.connection
         raw_connection = getattr(connection, "_cnx", None) or connection
-        if getattr(raw_connection, "unread_result", False):
-            with contextlib.suppress(Exception):
-                raw_connection.shutdown()
-                raw_connection.unread_result = False
-                raw_connection.reconnect()
+        try:
+            if getattr(raw_connection, "unread_result", False):
+                raw_connection.consume_results()
+        finally:
+            if cursor is not None:
+                with contextlib.suppress(Exception):
+                    cursor.close()
 
 
 class MysqlConnectorAsyncStreamSource:
@@ -302,16 +301,15 @@ async def fetch_chunk(self) -> "list[dict[str, Any]]":
     async def close(self, error: bool = False) -> None:
         cursor = self._cursor
         self._cursor = None
-        if cursor is not None:
-            with contextlib.suppress(Exception):
-                await cursor.close()
         connection = self._driver.connection
         raw_connection = getattr(connection, "_cnx", None) or connection
-        if getattr(raw_connection, "unread_result", False):
-            with contextlib.suppress(Exception):
-                await raw_connection.shutdown()
-                raw_connection.unread_result = False
-                await raw_connection.reconnect()
+        try:
+            if getattr(raw_connection, "unread_result", False):
+                await raw_connection.consume_results()
+        finally:
+            if cursor is not None:
+                with contextlib.suppress(Exception):
+                    await cursor.close()
 
 
 def normalize_execute_many_parameters(parameters: Any) -> Any:

sqlspec/driver/_stream.py

@@ -11,6 +11,7 @@
 
 import builtins
 import contextlib
+import inspect
 from typing import TYPE_CHECKING, Any, Generic, Protocol, TypeVar, cast, overload
 
 from typing_extensions import Self
@@ -43,7 +44,7 @@ def start(self) -> None: ...
 
     def fetch_chunk(self) -> "list[dict[str, Any]]": ...
 
-    def close(self, error: bool = False) -> None: ...
+    def close(self) -> None: ...
 
 
 class AsyncRowSource(Protocol):
@@ -53,7 +54,29 @@ async def start(self) -> None: ...
 
     async def fetch_chunk(self) -> "list[dict[str, Any]]": ...
 
-    async def close(self, error: bool = False) -> None: ...
+    async def close(self) -> None: ...
+
+
+def _close_sync_source(source: SyncRowSource, error: bool) -> None:
+    """Close a source while preserving the original no-argument contract."""
+    close = source.close
+    try:
+        inspect.signature(close).bind(error=error)
+    except (TypeError, ValueError):
+        close()
+        return
+    cast("Any", close)(error=error)
+
+
+async def _close_async_source(source: AsyncRowSource, error: bool) -> None:
+    """Close an async source while preserving the original no-argument contract."""
+    close = source.close
+    try:
+        inspect.signature(close).bind(error=error)
+    except (TypeError, ValueError):
+        await close()
+        return
+    await cast("Any", close)(error=error)
 
 
 def rows_to_dicts(rows: "list[Any]", column_names: "list[str]") -> "list[dict[str, Any]]":
@@ -142,7 +165,7 @@ def _close(self, error: bool = False) -> None:
         self._buffer = []
         self._buffer_index = 0
         with contextlib.suppress(Exception):
-            self._source.close(error=error)
+            _close_sync_source(self._source, error)
 
 
 class AsyncRowStream(Generic[RowT]):
@@ -224,7 +247,7 @@ async def _aclose(self, error: bool = False) -> None:
         self._buffer = []
         self._buffer_index = 0
         with contextlib.suppress(Exception):
-            await self._source.close(error=error)
+            await _close_async_source(self._source, error)
 
 
 class EagerSyncRowSource:

sqlspec/extensions/events/_channel.py

@@ -92,6 +92,12 @@ def _resolve_event_type(payload: "dict[str, Any]", metadata: "dict[str, Any] | N
 
 
 _POSTGRES_ADAPTERS = frozenset({"asyncpg", "psycopg", "psqlpy"})
+_EVENT_BACKENDS = frozenset({"notify", "notify_queue", "poll_queue", "aq", "txeventq"})
+_RETIRED_EVENT_BACKENDS = {
+    "listen_notify": "notify",
+    "listen_notify_durable": "notify_queue",
+    "table_queue": "poll_queue",
+}
 
 
 def _get_default_backend(adapter_name: "str | None") -> str:
@@ -101,6 +107,26 @@ def _get_default_backend(adapter_name: "str | None") -> str:
     return "poll_queue"
 
 
+def _resolve_backend_name(config: Any, extension_settings: "dict[str, Any]", adapter_name: "str | None") -> str:
+    """Resolve and validate event backend configuration."""
+    backend_name = extension_settings.get("backend")
+    if backend_name is None:
+        driver_features = getattr(config, "driver_features", {})
+        if isinstance(driver_features, dict):
+            backend_name = driver_features.get("events_backend")
+    if backend_name is None:
+        return _get_default_backend(adapter_name)
+    if backend_name in _RETIRED_EVENT_BACKENDS:
+        replacement = _RETIRED_EVENT_BACKENDS[backend_name]
+        msg = f"Event backend {backend_name!r} was removed; use {replacement!r}"
+        raise ImproperConfigurationError(msg)
+    if backend_name not in _EVENT_BACKENDS:
+        valid = ", ".join(sorted(_EVENT_BACKENDS))
+        msg = f"Unknown event backend {backend_name!r}; expected one of: {valid}"
+        raise ImproperConfigurationError(msg)
+    return cast("str", backend_name)
+
+
 def load_native_backend(
     config: Any, backend_name: str | None, extension_settings: "dict[str, Any]", adapter_name: "str | None" = None
 ) -> Any | None:
@@ -363,7 +389,7 @@ def __init__(self, config: "SyncDatabaseConfig[Any, Any, Any]") -> None:
         hints = get_runtime_hints(self._adapter_name, config)
         self._poll_interval_default = float(extension_settings.get("poll_interval") or hints.poll_interval)
         queue_backend = build_queue_backend(config, extension_settings, adapter_name=self._adapter_name, hints=hints)
-        backend_name = extension_settings.get("backend") or _get_default_backend(self._adapter_name)
+        backend_name = _resolve_backend_name(config, extension_settings, self._adapter_name)
         native_backend = load_native_backend(config, backend_name, extension_settings, adapter_name=self._adapter_name)
         if native_backend is None:
             if backend_name not in {None, "poll_queue"}:
@@ -591,7 +617,7 @@ def __init__(self, config: "AsyncDatabaseConfig[Any, Any, Any]") -> None:
         hints = get_runtime_hints(self._adapter_name, config)
         self._poll_interval_default = float(extension_settings.get("poll_interval") or hints.poll_interval)
         queue_backend = build_queue_backend(config, extension_settings, adapter_name=self._adapter_name, hints=hints)
-        backend_name = extension_settings.get("backend") or _get_default_backend(self._adapter_name)
+        backend_name = _resolve_backend_name(config, extension_settings, self._adapter_name)
         native_backend = load_native_backend(config, backend_name, extension_settings, adapter_name=self._adapter_name)
         if native_backend is None:
             if backend_name not in {None, "poll_queue"}:

tests/unit/adapters/test_mysqlconnector/test_config.py

@@ -289,43 +289,76 @@ async def test_async_dispatch_select_stream_uses_driver_cursor_options(monkeypat
     connection.cursor.assert_awaited_once_with(prepared=True, buffered=False)
 
 
-def test_sync_stream_close_reconnects_instead_of_draining_unread_rows() -> None:
-    """Early close discards the dirty protocol session without fetching remaining rows."""
+def test_sync_stream_close_consumes_unread_rows_without_reconnecting() -> None:
+    """Early close preserves the current session and any active transaction."""
     connection = MagicMock()
     connection.unread_result = True
     connection._cnx = None
     cursor = MagicMock()
-    cursor.close.side_effect = RuntimeError("Unread result found")
     driver = MagicMock(connection=connection, driver_features={})
     source = MysqlConnectorSyncStreamSource(driver, "SELECT 1", None, 10, set())
     source._cursor = cursor
 
     source.close()
 
-    cursor.fetchall.assert_not_called()
-    connection.shutdown.assert_called_once_with()
-    connection.reconnect.assert_called_once_with()
+    connection.consume_results.assert_called_once_with()
+    cursor.close.assert_called_once_with()
+    connection.shutdown.assert_not_called()
+    connection.reconnect.assert_not_called()
+
+
+def test_sync_stream_close_closes_cursor_when_consuming_unread_rows_fails() -> None:
+    connection = MagicMock(unread_result=True, _cnx=None)
+    connection.consume_results.side_effect = RuntimeError("consume failed")
+    cursor = MagicMock()
+    source = MysqlConnectorSyncStreamSource(MagicMock(connection=connection), "SELECT 1", None, 10, set())
+    source._cursor = cursor
+
+    with pytest.raises(RuntimeError, match="consume failed"):
+        source.close()
+
+    cursor.close.assert_called_once_with()
+    connection.reconnect.assert_not_called()
 
 
 @pytest.mark.anyio
-async def test_async_stream_close_reconnects_instead_of_draining_unread_rows() -> None:
-    """Async early close replaces the dirty protocol session without fetching remaining rows."""
+async def test_async_stream_close_consumes_unread_rows_without_reconnecting() -> None:
+    """Async early close preserves the current session and active transaction."""
     connection = MagicMock()
     connection.unread_result = True
     connection._cnx = None
+    connection.consume_results = AsyncMock()
     connection.shutdown = AsyncMock()
     connection.reconnect = AsyncMock()
     cursor = MagicMock()
-    cursor.close = AsyncMock(side_effect=RuntimeError("Unread result found"))
+    cursor.close = AsyncMock()
     driver = MagicMock(connection=connection, driver_features={})
     source = MysqlConnectorAsyncStreamSource(driver, "SELECT 1", None, 10, set())
     source._cursor = cursor
 
     await source.close()
 
-    cursor.fetchall.assert_not_called()
-    connection.shutdown.assert_awaited_once_with()
-    connection.reconnect.assert_awaited_once_with()
+    connection.consume_results.assert_awaited_once_with()
+    cursor.close.assert_awaited_once_with()
+    connection.shutdown.assert_not_awaited()
+    connection.reconnect.assert_not_awaited()
+
+
+@pytest.mark.anyio
+async def test_async_stream_close_closes_cursor_when_consuming_unread_rows_fails() -> None:
+    connection = MagicMock(unread_result=True, _cnx=None)
+    connection.consume_results = AsyncMock(side_effect=RuntimeError("consume failed"))
+    connection.reconnect = AsyncMock()
+    cursor = MagicMock()
+    cursor.close = AsyncMock()
+    source = MysqlConnectorAsyncStreamSource(MagicMock(connection=connection), "SELECT 1", None, 10, set())
+    source._cursor = cursor
+
+    with pytest.raises(RuntimeError, match="consume failed"):
+        await source.close()
+
+    cursor.close.assert_awaited_once_with()
+    connection.reconnect.assert_not_awaited()
 
 
 def test_sync_create_connection_forwards_local_infile_gate(monkeypatch: pytest.MonkeyPatch) -> None:

tests/unit/driver/test_stream.py

@@ -176,6 +176,26 @@ def close(self, error: bool = False) -> None:
     assert source.close_calls == 1
 
 
+def test_sync_close_supports_legacy_no_argument_source() -> None:
+    class LegacyCloseSource:
+        def __init__(self) -> None:
+            self.close_calls = 0
+
+        def start(self) -> None:
+            pass
+
+        def fetch_chunk(self) -> "list[dict[str, Any]]":
+            return []
+
+        def close(self) -> None:
+            self.close_calls += 1
+
+    source = LegacyCloseSource()
+    _sync_stream(source).close()
+
+    assert source.close_calls == 1
+
+
 # --------------------------------------------------------------------------- #
 # AsyncRowStream
 # --------------------------------------------------------------------------- #
@@ -279,6 +299,26 @@ async def close(self, error: bool = False) -> None:
     assert source.close_calls == 1
 
 
+async def test_async_close_supports_legacy_no_argument_source() -> None:
+    class LegacyCloseSource:
+        def __init__(self) -> None:
+            self.close_calls = 0
+
+        async def start(self) -> None:
+            pass
+
+        async def fetch_chunk(self) -> "list[dict[str, Any]]":
+            return []
+
+        async def close(self) -> None:
+            self.close_calls += 1
+
+    source = LegacyCloseSource()
+    await _async_stream(source).aclose()
+
+    assert source.close_calls == 1
+
+
 # --------------------------------------------------------------------------- #
 # Eager sources
 # --------------------------------------------------------------------------- #