fix: bound event and channel queues

docs/changelog.rst

@@ -9,6 +9,38 @@ important operational fixes.
 Recent Updates
 ==============
 
+v0.60.0 - Queue limits
+------------------------------------------------------------------------------
+
+**Added:**
+
+* The new ``listener_queue_capacity`` setting caps each PostgreSQL listener
+  queue. It works with asyncpg, psqlpy, and sync or async psycopg listeners.
+  The default has no limit. Bad values raise
+  :class:`~sqlspec.exceptions.ImproperConfigurationError`.
+* :class:`~sqlspec.extensions.litestar.channels.SQLSpecChannelsBackend` accepts
+  ``output_queue_capacity``. It caps decoded messages from any async event
+  transport, including PostgreSQL, Oracle, and polling channels. The default
+  has no limit. Bad values raise ``ValueError``.
+* ``output_queue_depth`` shows the current Litestar backlog.
+  ``dropped_message_count`` shows the total number of overflow drops.
+* PostgreSQL listener metrics now track ``events.listener.queue.depth`` and the
+  total ``events.listener.queue.dropped`` count for the hub.
+
+**Changed:**
+
+* A full capped queue drops its oldest item before it adds the new one. Each
+  PostgreSQL consumer has its own queue. Shutdown clears queued items and sets
+  depth to zero. Drop counts stay in place after a restart.
+* An overflow in PostgreSQL ``notify`` drops transient data. With
+  ``notify_queue``, it drops only a wake-up marker. The durable row stays in the
+  table and can be found by the next scan.
+* Oracle AQ and TxEventQ still read from their native queues. They do not add a
+  listener queue. Table-backed ``poll_queue`` stores check
+  ``listener_queue_capacity``, but the setting does not change how they poll.
+* Bad Litestar channel payloads are logged and acknowledged. They do not increase
+  ``dropped_message_count``.
+
 v0.59.0 - Data dictionary and loader access
 ------------------------------------------------------------------------------
 
@@ -49,6 +81,25 @@ v0.59.0 - Data dictionary and loader access
 * ADBC adapters for PostgreSQL now keep ``None`` in arrays. Each value binds as
   SQL ``NULL``. This keeps null values in place.
 
+v0.58.3 - Data and store fixes
+------------------------------------------------------------------------------
+
+**Fixed:**
+
+* Nested msgspec structs now use their own encoded field names. This works for
+  optional fields, annotations, lists, tuples, and maps. Mixed rename rules no
+  longer reuse the outer struct's rule. Keys in plain maps stay unchanged.
+* ``ensure_async_()`` now has a
+  :class:`collections.abc.Coroutine` return type. This matches the value it has
+  always returned at run time and removes the need for a cast.
+* SQLSpec Litestar session stores now inherit
+  :class:`litestar.stores.base.Store`. They keep its async context manager and
+  work with ``StoreRegistry``.
+* Each extension now loads its SQL migration query names on its own. A file
+  such as ``0001_create_table.sql`` can use ``migrate-0001-up`` and
+  ``migrate-0001-down`` with no clash across extensions. SQLSpec still stores
+  the prefixed tracker version.
+
 v0.58.2 - SQL file parameter diagnostics
 ------------------------------------------------------------------------------
 

docs/reference/extensions/events.rst

@@ -55,6 +55,7 @@ Configure the transport and durable reconciliation cadence independently:
             "events": {
                 "backend": "notify_queue",
                 "event_poll_interval": 1.0,
+                "listener_queue_capacity": 256,
             }
         },
     )
@@ -64,6 +65,15 @@ queue when no native wakeup arrives. The older ``poll_interval`` setting is a
 compatibility input; ``event_poll_interval`` takes precedence when both are
 provided.
 
+``listener_queue_capacity`` bounds pending PostgreSQL notifications separately
+for each consumer. It is unbounded when omitted. At capacity, the oldest pending
+notification is discarded so the newest notification can be retained. For
+``notify`` this intentionally loses the transient notification. For
+``notify_queue`` only the wake-up marker is lost; the durable row remains in the
+table and reconciliation recovers it. The setting is accepted by every event
+store so applications can share configuration across adapters, but it does not
+change ``poll_queue`` polling or Oracle AQ and TxEventQ native dequeue behavior.
+
 ``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.
@@ -83,6 +93,12 @@ backend owns its own listener hub that:
 * Serializes subscribe / unsubscribe under a lock so concurrent callers
   cannot race on driver-level statements that share the connection.
 
+Listener hubs report aggregate pending depth as
+``events.listener.queue.depth`` and increment the cumulative
+``events.listener.queue.dropped`` metric whenever overflow evicts an item.
+Shutdown clears pending payloads and records depth zero; a restarted hub keeps
+the runtime's cumulative dropped count.
+
 The listener lease is held for the backend lifetime. Publishers use separate,
 short-lived pooled sessions, so a shared PostgreSQL pool must configure at
 least two connections: ``max_size >= 2`` for asyncpg/psycopg and
@@ -93,6 +109,8 @@ the listener.
 The Oracle native backends (``aq`` and
 ``txeventq``) use an analogous pattern: a per-channel
 queue-handle cache backed by a single dedicated session per backend instance.
+They dequeue directly and do not add the PostgreSQL listener buffer described
+above.
 ``dequeue`` honors ``min(poll_interval, aq_wait_seconds)`` as its wait bound so
 the caller's polling cadence is respected.
 

docs/reference/extensions/litestar.rst

@@ -22,6 +22,16 @@ Configuration
 Channels Backend
 ================
 
+``SQLSpecChannelsBackend`` buffers decoded output from every asynchronous
+``EventChannel`` transport. Pass ``output_queue_capacity`` to bound that buffer;
+the default ``None`` remains unbounded. When full, the backend discards the
+oldest decoded message before acknowledging and retaining the newest one.
+``output_queue_depth`` reports the current pending count and
+``dropped_message_count`` reports cumulative overflow drops for the backend
+instance. Malformed payloads are acknowledged and logged without increasing the
+overflow count. Shutdown clears pending output while preserving the cumulative
+drop diagnostic for lifecycle reuse.
+
 .. autoclass:: sqlspec.extensions.litestar.SQLSpecChannelsBackend
    :members:
    :show-inheritance:

pyproject.toml

@@ -24,7 +24,7 @@ maintainers = [{ name = "Litestar Developers", email = "hello@litestar.dev" }]
 name = "sqlspec"
 readme = "README.md"
 requires-python = ">=3.10, <4.0"
-version = "0.59.0"
+version = "0.60.0"
 
 [project.urls]
 Discord = "https://discord.gg/litestar"
@@ -331,7 +331,7 @@ opt_level = "3"   # Maximum optimization (0-3)
 allow_dirty = true
 commit = false
 commit_args = "--no-verify"
-current_version = "0.59.0"
+current_version = "0.60.0"
 ignore_missing_files = false
 ignore_missing_version = false
 message = "chore(release): bump to v{new_version}"

sqlspec/adapters/asyncpg/events/_hub.py

@@ -17,6 +17,7 @@
 
 from sqlspec.exceptions import EventChannelError
 from sqlspec.extensions.events import normalize_event_channel_name
+from sqlspec.extensions.events._buffer import enqueue_with_capacity, resolve_listener_queue_capacity
 from sqlspec.utils.logging import get_logger, log_with_context
 from sqlspec.utils.type_guards import has_add_listener
 
@@ -39,6 +40,7 @@ class AsyncpgListenerHub:
         "_config",
         "_connection",
         "_connection_cm",
+        "_listener_queue_capacity",
         "_lock",
         "_pool_destroying_registered",
         "_queues",
@@ -49,6 +51,7 @@ def __init__(self, config: "AsyncpgConfig", backend_name: str = "notify") -> Non
         self._backend_name = backend_name
         self._config = config
         self._lock = asyncio.Lock()
+        self._listener_queue_capacity = resolve_listener_queue_capacity(config)
         self._queues: dict[str, WeakKeyDictionary[asyncio.Task[Any], asyncio.Queue[str]]] = {}
         self._callbacks: dict[str, Callable[..., None]] = {}
         self._connection_cm: Any | None = None
@@ -106,9 +109,12 @@ async def dequeue(self, channel: str, poll_interval: float) -> "str | None":
         if queue is None:
             return None
         try:
-            return await asyncio.wait_for(queue.get(), timeout=poll_interval)
+            payload = await asyncio.wait_for(queue.get(), timeout=poll_interval)
         except asyncio.TimeoutError:
             return None
+        else:
+            self._record_queue_depth()
+            return payload
 
     async def shutdown(self) -> None:
         async with self._lock:
@@ -123,6 +129,7 @@ async def shutdown(self) -> None:
             self._callbacks.clear()
             self._connection = None
             self._connection_cm = None
+            self._record_queue_depth()
         if connection is not None:
             for channel in channels:
                 callback = callbacks.get(channel)
@@ -151,7 +158,7 @@ def _get_consumer_queue(self, channel: str) -> "asyncio.Queue[str] | None":
             return None
         queue = queues.get(task)
         if queue is None:
-            queue = asyncio.Queue()
+            queue = asyncio.Queue(maxsize=self._listener_queue_capacity or 0)
             queues[task] = queue
         return queue
 
@@ -207,7 +214,13 @@ def _dispatch(self, channel: str, payload: str) -> None:
         if queues is None:
             return
         for queue in list(queues.values()):
-            queue.put_nowait(payload)
+            if enqueue_with_capacity(queue, payload, self._listener_queue_capacity, empty_error=asyncio.QueueEmpty):
+                self._config.get_observability_runtime().increment_metric("events.listener.queue.dropped")
+        self._record_queue_depth()
+
+    def _record_queue_depth(self) -> None:
+        depth = sum(queue.qsize() for queues in self._queues.values() for queue in queues.values())
+        self._config.get_observability_runtime().record_metric("events.listener.queue.depth", depth)
 
 
 def _record_listener_lifecycle(config: "AsyncpgConfig", backend_name: str, status: str) -> None:

sqlspec/adapters/psqlpy/events/_hub.py

@@ -18,6 +18,7 @@
 
 from sqlspec.core import SQL
 from sqlspec.extensions.events import normalize_event_channel_name
+from sqlspec.extensions.events._buffer import enqueue_with_capacity, resolve_listener_queue_capacity
 from sqlspec.utils.logging import get_logger, log_with_context
 from sqlspec.utils.type_guards import is_notification
 from sqlspec.utils.uuids import uuid4
@@ -79,6 +80,7 @@ class PsqlpyListenerHub:
         "_callbacks",
         "_config",
         "_listener",
+        "_listener_queue_capacity",
         "_listener_started",
         "_lock",
         "_pool_destroying_registered",
@@ -91,6 +93,7 @@ def __init__(self, config: "PsqlpyConfig", backend_name: str = "notify") -> None
         self._backend_name = backend_name
         self._config = config
         self._lock = asyncio.Lock()
+        self._listener_queue_capacity = resolve_listener_queue_capacity(config)
         self._queues: dict[str, WeakKeyDictionary[asyncio.Task[Any], asyncio.Queue[str]]] = {}
         self._callbacks: dict[str, _PsqlpyHubCallback] = {}
         self._ready_events: dict[str, dict[str, asyncio.Event]] = {}
@@ -134,9 +137,12 @@ async def dequeue(self, channel: str, poll_interval: float) -> "str | None":
         if queue is None:
             return None
         try:
-            return await asyncio.wait_for(queue.get(), timeout=poll_interval)
+            payload = await asyncio.wait_for(queue.get(), timeout=poll_interval)
         except asyncio.TimeoutError:
             return None
+        else:
+            self._record_queue_depth()
+            return payload
 
     async def shutdown(self) -> None:
         async with self._lock:
@@ -149,6 +155,7 @@ async def shutdown(self) -> None:
             self._queues.clear()
             self._callbacks.clear()
             self._ready_events.clear()
+            self._record_queue_depth()
             self._listener = None
             self._listener_started = False
         if listener is not None:
@@ -178,7 +185,7 @@ def _get_consumer_queue(
             return None
         queue = queues.get(task)
         if queue is None:
-            queue = asyncio.Queue()
+            queue = asyncio.Queue(maxsize=self._listener_queue_capacity or 0)
             queues[task] = queue
         return queue
 
@@ -194,7 +201,7 @@ async def _subscribe_locked(
         queues: WeakKeyDictionary[asyncio.Task[Any], asyncio.Queue[str]] = WeakKeyDictionary()
         consumer_queue: asyncio.Queue[str] | None = None
         if consumer_task is not None:
-            consumer_queue = asyncio.Queue()
+            consumer_queue = asyncio.Queue(maxsize=self._listener_queue_capacity or 0)
             queues[consumer_task] = consumer_queue
         self._queues[channel] = queues
         self._callbacks[channel] = callback
@@ -284,7 +291,13 @@ def _dispatch(self, channel: str, payload: str) -> None:
         if queues is None:
             return
         for queue in list(queues.values()):
-            queue.put_nowait(payload)
+            if enqueue_with_capacity(queue, payload, self._listener_queue_capacity, empty_error=asyncio.QueueEmpty):
+                self._config.get_observability_runtime().increment_metric("events.listener.queue.dropped")
+        self._record_queue_depth()
+
+    def _record_queue_depth(self) -> None:
+        depth = sum(queue.qsize() for queues in self._queues.values() for queue in queues.values())
+        self._config.get_observability_runtime().record_metric("events.listener.queue.depth", depth)
 
 
 def _record_listener_lifecycle(config: "PsqlpyConfig", backend_name: str, status: str) -> None: