From 6c5a93b53fa858a169bf625fc0e096549636ba6d Mon Sep 17 00:00:00 2001 From: "sergei.romanchuk" Date: Sat, 18 Jul 2026 14:07:27 +0200 Subject: [PATCH] fix(broker): recover asyncpg listener after disconnect Recreate terminated listener connections with bounded backoff and reconcile durable messages missed during the LISTEN gap. Preserve atomic claims and shutdown-safe lifecycle handling. Add regression coverage for listener termination, reconnect races, missed notifications, and reconciliation pagination. Refs taskiq-python/taskiq#633 --- README.md | 12 + taskiq_postgresql/abc/driver.py | 22 ++ taskiq_postgresql/abc/query.py | 39 +++ taskiq_postgresql/broker.py | 464 +++++++++++++++++++++++--- taskiq_postgresql/drivers/_asyncpg.py | 240 +++++++++++-- taskiq_postgresql/exceptions.py | 8 + tests/test_asyncpg_listener.py | 171 ++++++++++ tests/test_broker.py | 220 ++++++++++++ tests/test_listener_recovery.py | 422 +++++++++++++++++++++++ tests/test_reconciliation_queries.py | 59 ++++ uv.lock | 130 ++++---- 11 files changed, 1632 insertions(+), 155 deletions(-) create mode 100644 tests/test_asyncpg_listener.py create mode 100644 tests/test_listener_recovery.py create mode 100644 tests/test_reconciliation_queries.py diff --git a/README.md b/README.md index 32a06eb..5f33f44 100644 --- a/README.md +++ b/README.md @@ -157,6 +157,18 @@ async def setup_schedule(): | `driver` | `Literal["asyncpg", "psycopg", "psqlpy"]` | `"asyncpg"` | Database driver | | `**connect_kwargs` | `Any` | - | Additional driver-specific connection parameters | +#### Listener recovery + +After an established asyncpg listener is disconnected, the broker creates a +fresh listener with capped exponential backoff. Once `LISTEN` is registered, +it reconciles ready rows through the same atomic claim path, so notifications +lost during the connection gap do not strand tasks. Initial connection and +configuration failures remain fail-fast. Driver-specific disconnect recovery +for psycopg and psqlpy is not yet guaranteed. Reconciliation is scoped by an +internal channel marker stored in the database; it does not mutate task labels. +Rows written by earlier versions without the marker are not auto-reconciled, +because their original channel cannot be determined safely. + ### PostgresqlResultBackend | Parameter | Type | Default | Description | diff --git a/taskiq_postgresql/abc/driver.py b/taskiq_postgresql/abc/driver.py index 69d47c8..b83b52c 100644 --- a/taskiq_postgresql/abc/driver.py +++ b/taskiq_postgresql/abc/driver.py @@ -15,6 +15,8 @@ DeleteReturningQuery, InsertOrUpdateQuery, InsertQuery, + MaxValueQuery, + SelectAvailableIdsQuery, SelectQuery, ) @@ -56,6 +58,8 @@ def __init__( self.select_query = SelectQuery( self.table_name, ) + self.max_value_query = MaxValueQuery(self.table_name) + self.select_available_ids_query = SelectAvailableIdsQuery(self.table_name) self.insert_or_update_query = InsertOrUpdateQuery(self.table_name) self.run_migrations = run_migrations @@ -135,6 +139,24 @@ async def select( ) -> list[dict[str, Any]]: """Select a row from a table.""" + async def max_value(self, column: Column) -> int: + """Disable reconciliation for legacy custom drivers by default.""" + return 0 + + async def select_available_ids( + self, + *, + primary_key: Column, + labels: Column, + created_at: Column, + channel_name: str, + after_id: int, + through_id: int, + limit: int, + ) -> list[int]: + """Return no reconciliation rows for legacy custom drivers.""" + return [] + @abstractmethod async def exists(self, id: Any) -> bool: """Check if a row exists in a table.""" diff --git a/taskiq_postgresql/abc/query.py b/taskiq_postgresql/abc/query.py index d5ea680..2436f8e 100644 --- a/taskiq_postgresql/abc/query.py +++ b/taskiq_postgresql/abc/query.py @@ -1,6 +1,8 @@ from abc import ABC, abstractmethod from typing import Any, Literal, Optional, Sequence +BROKER_CHANNEL_LABEL = "__taskiq_postgresql_channel__" + class QueryBase(ABC): """Base class for all queries.""" @@ -219,6 +221,43 @@ def make_query( ) +class MaxValueQuery(QueryBase): + """Query the greatest value of a column.""" + + def make_query(self, column: Column) -> str: + """Return a MAX query that uses zero for an empty table.""" + return f"SELECT COALESCE(MAX({column.name}), 0) FROM {self.table_name}" # noqa: S608 + + +class SelectAvailableIdsQuery(QueryBase): + """Query available message IDs within a fixed high-watermark.""" + + def make_query( + self, + primary_key: Column, + labels: Column, + created_at: Column, + ) -> str: + """Return the bounded keyset query for ready messages.""" + return ( + f"SELECT {primary_key.name} FROM {self.table_name} " # noqa: S608 + f"WHERE {primary_key.name} > $1 AND {primary_key.name} <= $2 " + f"AND {labels.name}->>'{BROKER_CHANNEL_LABEL}' = $3 " + f"AND ({labels.name}->>'delay' IS NULL " + "OR EXTRACT(EPOCH FROM " + f"(CURRENT_TIMESTAMP - {created_at.name})) >= CASE " + f"WHEN jsonb_typeof({labels.name}->'delay') = 'number' " + f"THEN trunc(({labels.name}->>'delay')::numeric) " + f"WHEN jsonb_typeof({labels.name}->'delay') = 'string' " + f"AND btrim({labels.name}->>'delay') ~ '^[+-]?[0-9]+$' " + f"THEN btrim({labels.name}->>'delay')::numeric " + f"WHEN jsonb_typeof({labels.name}->'delay') = 'boolean' " + f"THEN CASE WHEN ({labels.name}->>'delay')::boolean THEN 1 ELSE 0 END " + "END) " + f"ORDER BY {primary_key.name} LIMIT $4" + ) + + class CreatedAtColumn(Column): """Column for the created at timestamp.""" diff --git a/taskiq_postgresql/broker.py b/taskiq_postgresql/broker.py index f69bbe2..5e07d0f 100644 --- a/taskiq_postgresql/broker.py +++ b/taskiq_postgresql/broker.py @@ -1,22 +1,43 @@ from __future__ import annotations -from asyncio import Queue as AsyncQueue -from asyncio import Task, get_running_loop +from asyncio import ( + CancelledError, + Event, + Lock, + Task, + TimeoutError, + create_task, + get_running_loop, + wait_for, +) from dataclasses import dataclass from logging import getLogger +from random import uniform +from time import monotonic from typing import TYPE_CHECKING, Any, Callable, Literal, Optional, TypeVar, Union from taskiq import AckableMessage, AsyncBroker, AsyncResultBackend, BrokerMessage from taskiq_postgresql.abc.driver import ListenDriver -from taskiq_postgresql.abc.query import Column, CreatedAtColumn, PrimaryKeyColumn +from taskiq_postgresql.abc.query import ( + BROKER_CHANNEL_LABEL, + Column, + CreatedAtColumn, + PrimaryKeyColumn, +) +from taskiq_postgresql.exceptions import ( + ListenerDisconnectedError, + TransientDatabaseConnectionError, +) from taskiq_postgresql.utils import get_db_driver, get_db_listen_driver if TYPE_CHECKING: - from collections.abc import AsyncGenerator + from collections.abc import AsyncGenerator, AsyncIterator _T = TypeVar("_T") +_RECONCILIATION_BATCH_SIZE = 100 +_MAX_LISTENER_RETRY_DELAY = 30.0 logger = getLogger("taskiq.asyncpg_broker") @@ -105,7 +126,17 @@ def __init__( ) self.pool_kwargs: dict[str, Any] = pool_kwargs if pool_kwargs else {} self.max_retry_attempts: int = max_retry_attempts - self._queue: AsyncQueue[str] | None = None + self._driver_name = driver + self._shutdown_event = Event() + self._lifecycle_lock = Lock() + self._listener_recovery_lock = Lock() + self._shutdown_requests = 0 + self._base_started = False + self._query_driver_started = False + self._listener_started = False + self._listener_generation = 0 + self._pending_listen_driver: ListenDriver | None = None + self._listener_start_task: Task[None] | None = None self.columns = Table(task_id=Column(name="task_id", type_=field_for_task_id)) @@ -124,7 +155,12 @@ def __init__( run_migrations=run_migrations, **self.connection_kwargs, ) - self.listen_driver: ListenDriver = get_db_listen_driver(driver)( + self._listen_driver_factory = get_db_listen_driver(driver) + self.listen_driver: ListenDriver = self._new_listen_driver() + + def _new_listen_driver(self) -> ListenDriver: + """Create an isolated listener generation.""" + return self._listen_driver_factory( connection_string=self.dsn, channel_name=self.channel_name, **self.connection_kwargs, @@ -143,16 +179,145 @@ def dsn(self) -> str: async def startup(self) -> None: """Initialize the broker.""" - await super().startup() - - await self.driver.on_startup() - await self.listen_driver.on_startup() + async with self._lifecycle_lock: + if self._shutdown_requests: + raise RuntimeError("Broker shutdown is in progress.") + if self._base_started: + return + self._shutdown_event.clear() + try: + self._base_started = True + await super().startup() + self._raise_if_shutdown_requested() + + self._query_driver_started = True + await self.driver.on_startup() + self._raise_if_shutdown_requested() + + self._listener_started = True + startup_task = create_task(self.listen_driver.on_startup()) + self._listener_start_task = startup_task + try: + await startup_task + finally: + if self._listener_start_task is startup_task: + self._listener_start_task = None + self._raise_if_shutdown_requested() + self._listener_generation = 1 + except BaseException: + self._shutdown_event.set() + await self._shutdown_active_listener() + await self._shutdown_query_driver() + await self._shutdown_base_broker() + raise + + def _raise_if_shutdown_requested(self) -> None: + """Cancel startup when shutdown won the lifecycle race.""" + if self._shutdown_event.is_set(): + raise CancelledError async def shutdown(self) -> None: """Close all connections on shutdown.""" - await super().shutdown() - await self.driver.on_shutdown() - await self.listen_driver.on_shutdown() + self._shutdown_requests += 1 + self._shutdown_event.set() + startup_task = self._listener_start_task + if startup_task is not None and not startup_task.done(): + startup_task.cancel() + try: + async with self._lifecycle_lock: + errors = [ + await self._cancel_listener_startup(), + await self._shutdown_pending_listener(), + await self._shutdown_base_broker(), + await self._shutdown_query_driver(), + await self._shutdown_active_listener(), + ] + first_error = next( + (error for error in errors if error is not None), + None, + ) + if first_error is not None: + raise first_error + finally: + self._shutdown_requests -= 1 + + async def _cancel_listener_startup(self) -> Exception | None: + """Cancel a listener generation that is still connecting.""" + startup_task, self._listener_start_task = self._listener_start_task, None + if startup_task is None: + return None + if not startup_task.done(): + startup_task.cancel() + try: + await startup_task + except CancelledError: + return None + except Exception as error: # pragma: no cover - defensive cleanup + return error + return None + + async def _shutdown_pending_listener(self) -> Exception | None: + """Close a listener generation that has not been promoted.""" + candidate, self._pending_listen_driver = self._pending_listen_driver, None + if candidate is None: + return None + return await self._shutdown_listener(candidate) + + async def _shutdown_active_listener(self) -> Exception | None: + """Close the active listener generation once.""" + if not self._listener_started: + return None + self._listener_started = False + return await self._shutdown_listener(self.listen_driver) + + async def _shutdown_listener( + self, + listen_driver: ListenDriver, + ) -> Exception | None: + """Close a listener without allowing cleanup to hide a primary error.""" + try: + await listen_driver.on_shutdown() + except Exception as error: # pragma: no cover - defensive cleanup + logger.warning( + "PostgreSQL listener cleanup failed " + "(driver=%s channel=%s exception=%s)", + self._driver_name, + self.channel_name, + type(error).__name__, + ) + return error + return None + + async def _shutdown_query_driver(self) -> Exception | None: + """Close the query driver and record its lifecycle state.""" + if not self._query_driver_started: + return None + self._query_driver_started = False + try: + await self.driver.on_shutdown() + except Exception as error: # pragma: no cover - defensive cleanup + logger.warning( + "PostgreSQL query driver cleanup failed (driver=%s exception=%s)", + self._driver_name, + type(error).__name__, + ) + return error + return None + + async def _shutdown_base_broker(self) -> Exception | None: + """Close Taskiq-owned resources and record their lifecycle state.""" + if not self._base_started: + return None + self._base_started = False + try: + await super().shutdown() + except Exception as error: # pragma: no cover - defensive cleanup + logger.warning( + "Taskiq broker cleanup failed (exception=%s)", + type(error).__name__, + ) + return error + return None async def kick(self, message: BrokerMessage) -> None: """ @@ -162,6 +327,10 @@ async def kick(self, message: BrokerMessage) -> None: :param message: Message to send. """ + stored_labels = { + **message.labels, + BROKER_CHANNEL_LABEL: self.channel_name, + } message_inserted_id = await self.driver.insert( [ self.columns.task_id, @@ -173,7 +342,7 @@ async def kick(self, message: BrokerMessage) -> None: message.task_id, message.task_name, message.message, - message.labels, + stored_labels, ], [ self.columns.primary_key, @@ -222,46 +391,233 @@ async def listen(self) -> AsyncGenerator[AckableMessage, None]: :yields: AckableMessage instances. """ - while True: + query_failure_attempt = 0 + while not self._shutdown_event.is_set(): + listen_driver = self.listen_driver + listener_generation = self._listener_generation try: - async for message_id in self.listen_driver: - # Normalize payload to integer ID (psycopg may yield string payloads). - try: - normalized_id = int(message_id) # type: ignore[arg-type] - except (TypeError, ValueError): - logger.warning( - "Invalid NOTIFY payload %r on channel %s", - message_id, - self.channel_name, - ) - continue - - # Atomically claim the message row. If None is returned, another - # worker has already claimed it. - row = await self.driver.delete_returning( - self.columns.primary_key, - normalized_id, - [self.columns.message], - ) - - if row is None: - # Claimed elsewhere or missing; skip. - continue - - message: Optional[bytes] = row.get(self.columns.message.name) - - if message is None: - logger.warning( - "Message with id %s has no payload.", - message_id, - ) - continue - - async def ack(*, _message_id: int = message_id) -> None: # noqa: ARG001 - # No-op: the row was already deleted when claimed. - return None - - yield AckableMessage(data=message, ack=ack) + async for message in self._messages_from_ids(self._reconcile_ids()): + yield message + query_failure_attempt = 0 + + async for message in self._messages_from_ids(listen_driver): + yield message + + if self._shutdown_event.is_set(): + return + self._handle_listener_completion() + except ListenerDisconnectedError: + if not await self._recover_listener( + listen_driver, + listener_generation, + ): + return + except TransientDatabaseConnectionError as error: + query_failure_attempt += 1 + delay = self._listener_retry_delay(query_failure_attempt) + logger.warning( + "PostgreSQL query connection unavailable " + "(driver=%s attempt=%s delay=%.3f exception=%s)", + self._driver_name, + query_failure_attempt, + delay, + type(error).__name__, + ) + if await self._wait_for_shutdown(delay): + return except Exception as error: - logger.exception("Error processing message: %s", error) + self._handle_unexpected_listener_error(error) + + def _handle_listener_completion(self) -> None: + """Keep legacy drivers unchanged and recover asyncpg completion.""" + if self._driver_name == "asyncpg": + raise ListenerDisconnectedError( + "PostgreSQL listener stopped unexpectedly.", + ) + + def _handle_unexpected_listener_error(self, error: Exception) -> None: + """Preserve legacy retries while surfacing asyncpg contract failures.""" + if self._driver_name == "asyncpg": + raise error + logger.exception("Error processing message: %s", error) + + async def _reconcile_ids(self) -> AsyncGenerator[int, None]: + """Yield ready row IDs that may have missed a transient notification.""" + through_id = await self.driver.max_value(self.columns.primary_key) + after_id = 0 + while after_id < through_id: + message_ids = await self.driver.select_available_ids( + primary_key=self.columns.primary_key, + labels=self.columns.labels, + created_at=self.columns.created_at, + channel_name=self.channel_name, + after_id=after_id, + through_id=through_id, + limit=_RECONCILIATION_BATCH_SIZE, + ) + if not message_ids: + return + for message_id in message_ids: + yield message_id + after_id = message_ids[-1] + + async def _messages_from_ids( + self, + message_ids: AsyncIterator[Any], + ) -> AsyncGenerator[AckableMessage, None]: + """Claim and decode IDs without advancing past database failures.""" + async for message_id in message_ids: + message = await self._claim_message(message_id) + if message is not None: + yield message + + async def _claim_message(self, message_id: Any) -> AckableMessage | None: + """Atomically claim one database row and build its Taskiq message.""" + try: + normalized_id = int(message_id) + except (TypeError, ValueError): + logger.warning( + "Invalid NOTIFY payload on channel %s", + self.channel_name, + ) + return None + + row = await self.driver.delete_returning( + self.columns.primary_key, + normalized_id, + [self.columns.message], + ) + if row is None: + return None + + message: Optional[bytes] = row.get(self.columns.message.name) + if message is None: + logger.warning("Message with id %s has no payload.", normalized_id) + return None + + async def ack() -> None: + return None + + return AckableMessage(data=message, ack=ack) + + async def _recover_listener( + self, + failed_driver: ListenDriver, + failed_generation: int, + ) -> bool: + """Recover one failed generation exactly once across consumers.""" + async with self._listener_recovery_lock: + if self._shutdown_event.is_set(): + return False + if ( + failed_driver is not self.listen_driver + or failed_generation != self._listener_generation + ): + return True + return await self._recover_current_listener(failed_generation) + + async def _recover_current_listener(self, generation: int) -> bool: + """Replace a disconnected listener using bounded exponential backoff.""" + disconnected_at = monotonic() + logger.warning( + "PostgreSQL listener disconnected (driver=%s channel=%s generation=%s)", + self._driver_name, + self.channel_name, + generation, + ) + + if self._listener_started: + self._listener_started = False + await self._shutdown_listener(self.listen_driver) + + attempt = 0 + while not self._shutdown_event.is_set(): + attempt += 1 + delay = self._listener_retry_delay(attempt) + logger.info( + "PostgreSQL listener reconnect attempt " + "(driver=%s channel=%s attempt=%s delay=%.3f)", + self._driver_name, + self.channel_name, + attempt, + delay, + ) + if await self._wait_for_shutdown(delay): + return False + + candidate = self._new_listen_driver() + if not await self._start_listener_candidate(candidate, attempt): continue + + self.listen_driver = candidate + self._listener_started = True + self._listener_generation = generation + 1 + logger.info( + "PostgreSQL listener reconnected " + "(driver=%s channel=%s attempts=%s downtime=%.3f generation=%s)", + self._driver_name, + self.channel_name, + attempt, + monotonic() - disconnected_at, + self._listener_generation, + ) + return True + return False + + async def _start_listener_candidate( + self, + candidate: ListenDriver, + attempt: int, + ) -> bool: + """Start and validate one listener reconnect candidate.""" + self._pending_listen_driver = candidate + startup_task = create_task(candidate.on_startup()) + self._listener_start_task = startup_task + try: + await startup_task + except CancelledError: + await self._shutdown_listener(candidate) + if self._shutdown_event.is_set(): + return False + raise + except TransientDatabaseConnectionError as error: + await self._shutdown_listener(candidate) + logger.warning( + "PostgreSQL listener reconnect failed " + "(driver=%s channel=%s attempt=%s exception=%s)", + self._driver_name, + self.channel_name, + attempt, + type(error).__name__, + ) + return False + except BaseException: + await self._shutdown_listener(candidate) + raise + finally: + if self._listener_start_task is startup_task: + self._listener_start_task = None + if self._pending_listen_driver is candidate: + self._pending_listen_driver = None + + if self._shutdown_event.is_set(): + await self._shutdown_listener(candidate) + return False + return True + + def _listener_retry_delay(self, attempt: int) -> float: + """Return capped exponential backoff with bounded jitter.""" + exponent = max(0, min(attempt - 1, 6)) + base_delay = min(0.5 * (2**exponent), _MAX_LISTENER_RETRY_DELAY) + return min( + base_delay * uniform(0.8, 1.2), # noqa: S311 + _MAX_LISTENER_RETRY_DELAY, + ) + + async def _wait_for_shutdown(self, delay: float) -> bool: + """Wait for either shutdown or the reconnect delay.""" + try: + await wait_for(self._shutdown_event.wait(), timeout=delay) + except TimeoutError: + return False + return True diff --git a/taskiq_postgresql/drivers/_asyncpg.py b/taskiq_postgresql/drivers/_asyncpg.py index 1ca242d..469d152 100644 --- a/taskiq_postgresql/drivers/_asyncpg.py +++ b/taskiq_postgresql/drivers/_asyncpg.py @@ -1,17 +1,46 @@ -from asyncio import Queue as AsyncQueue -from contextlib import asynccontextmanager +from asyncio import Event +from asyncio import TimeoutError as AsyncioTimeoutError +from collections import deque +from collections.abc import Callable +from contextlib import asynccontextmanager, suppress from datetime import date, datetime from types import TracebackType from typing import Any, AsyncIterator, Optional, Sequence, Union from uuid import UUID from asyncpg import Connection, Pool, connect, create_pool +from asyncpg.exceptions import ( + AdminShutdownError, + CannotConnectNowError, + CrashShutdownError, + IdleSessionTimeoutError, + InterfaceError, + PostgresConnectionError, + PostgresError, + TooManyConnectionsError, +) from asyncpg.transaction import Transaction from taskiq.compat import IS_PYDANTIC2 from taskiq_postgresql.abc.driver import ListenDriver, QueryDriver from taskiq_postgresql.abc.query import Column -from taskiq_postgresql.exceptions import DatabaseConnectionError +from taskiq_postgresql.exceptions import ( + DatabaseConnectionError, + ListenerDisconnectedError, + TransientDatabaseConnectionError, +) + +_TRANSIENT_CONNECTION_ERRORS = ( + OSError, + TimeoutError, + AsyncioTimeoutError, + PostgresConnectionError, + AdminShutdownError, + CrashShutdownError, + CannotConnectNowError, + IdleSessionTimeoutError, + TooManyConnectionsError, +) if IS_PYDANTIC2: from pydantic_core import to_json @@ -87,8 +116,20 @@ def __parser_query( @asynccontextmanager async def connection(self) -> AsyncIterator[Connection]: - async with self.pool.acquire() as connection: - yield connection + connection: Optional[Connection] = None + try: + async with self.pool.acquire() as connection: + yield connection + except _TRANSIENT_CONNECTION_ERRORS: + raise TransientDatabaseConnectionError( + "PostgreSQL query connection failed.", + ) from None + except InterfaceError: + if connection is not None and connection.is_closed(): + raise TransientDatabaseConnectionError( + "PostgreSQL query connection failed.", + ) from None + raise async def __aenter__(self) -> Connection: """Enter the context manager.""" @@ -199,6 +240,38 @@ async def select( {column.name: row[column.name] for column in columns} for row in rows ] + async def max_value(self, column: Column) -> int: + """Return the greatest value in a column.""" + async with self, self.connection() as connection: + value = await connection.fetchval(self.max_value_query.make_query(column)) + return int(value) + + async def select_available_ids( + self, + *, + primary_key: Column, + labels: Column, + created_at: Column, + channel_name: str, + after_id: int, + through_id: int, + limit: int, + ) -> list[int]: + """Return a bounded page of message IDs that are ready to run.""" + async with self, self.connection() as connection: + rows = await connection.fetch( + self.select_available_ids_query.make_query( + primary_key, + labels, + created_at, + ), + after_id, + through_id, + channel_name, + limit, + ) + return [int(row[primary_key.name]) for row in rows] + async def exists(self, id: Any) -> bool: """Check if a row exists in a table.""" async with self, self.connection() as connection: @@ -222,17 +295,23 @@ async def delete_by_date( async def on_startup(self) -> None: """On startup.""" - if self.run_migrations: - async with self, self.connection() as connection: - transaction = connection.transaction() - await transaction.start() - await self.create_table() - await self.create_index() - await transaction.commit() + try: + if self.run_migrations: + async with self, self.connection() as connection: + transaction = connection.transaction() + await transaction.start() + await self.create_table() + await self.create_index() + await transaction.commit() + except BaseException: + with suppress(Exception): + await self.on_shutdown() + raise async def on_shutdown(self) -> None: """On shutdown.""" - await self.pool.close() + if self.pool is not None: + await self.pool.close() self.pool = None async def execute(self, query: str, *values: Any) -> str: @@ -252,26 +331,114 @@ def __init__( ) -> None: """Initialize the listen driver.""" super().__init__(connection_string, channel_name, **connection_kwargs) - self._queue: AsyncQueue[int] = AsyncQueue() + self.connection: Optional[Connection] = None + self._messages: deque[int] = deque() + self._state_changed = Event() + self._stopping = False + self._disconnected = False + self._termination_callback: Callable[[Connection], None] = ( + self._termination_handler + ) async def on_startup(self) -> None: """On startup.""" - self.connection = await connect( - self.connection_string, - **self.connection_kwargs, - ) - await self.connection.add_listener( - self.channel_name, - self._notification_handler, - ) + self._messages.clear() + self._state_changed.clear() + self._stopping = False + self._disconnected = False + + connection: Optional[Connection] = None + termination_registered = False + listener_registered = False + try: + connection = await connect( + self.connection_string, + **self.connection_kwargs, + ) + self.connection = connection + connection.add_termination_listener(self._termination_callback) + termination_registered = True + await connection.add_listener( + self.channel_name, + self._notification_handler, + ) + listener_registered = True + except BaseException as error: + self._stopping = True + await self._close_connection( + connection, + termination_registered=termination_registered, + listener_registered=listener_registered, + propagate_errors=False, + ) + if self.connection is connection: + self.connection = None + connection_closed = connection is not None and connection.is_closed() + if isinstance(error, _TRANSIENT_CONNECTION_ERRORS) or ( + isinstance(error, InterfaceError) and connection_closed + ): + raise TransientDatabaseConnectionError( + "PostgreSQL listener connection failed.", + ) from None + if isinstance( + error, + (PostgresError, InterfaceError, ValueError, TypeError), + ): + raise DatabaseConnectionError( + "PostgreSQL listener configuration failed.", + ) from None + raise async def on_shutdown(self) -> None: """On shutdown.""" - await self.connection.remove_listener( - self.channel_name, - self._notification_handler, + self._stopping = True + self._state_changed.set() + connection, self.connection = self.connection, None + await self._close_connection( + connection, + termination_registered=True, + listener_registered=True, ) - await self.connection.close() + + async def _close_connection( + self, + connection: Optional[Connection], + *, + termination_registered: bool, + listener_registered: bool, + propagate_errors: bool = True, + ) -> None: + """Close a listener connection without skipping later cleanup steps.""" + if connection is None: + return + + first_error: Optional[Exception] = None + if termination_registered: + try: + connection.remove_termination_listener(self._termination_callback) + except Exception as error: # pragma: no cover - defensive cleanup + first_error = error + if listener_registered: + try: + await connection.remove_listener( + self.channel_name, + self._notification_handler, + ) + except Exception as error: # pragma: no cover - defensive cleanup + first_error = first_error or error + try: + await connection.close() + except Exception as error: # pragma: no cover - defensive cleanup + first_error = first_error or error + + if first_error is not None and propagate_errors: + raise first_error + + def _termination_handler(self, connection: Connection) -> None: + """Wake consumers when the active connection terminates unexpectedly.""" + if connection is self.connection and not self._stopping: + self._disconnected = True + self._state_changed.set() def _notification_handler( self, @@ -291,11 +458,22 @@ def _notification_handler( **channel**: name of the channel the notification was sent to; **payload**: the payload. """ - if self._queue is not None: - self._queue.put_nowait(int(payload)) + if con_ref is self.connection and not self._stopping: + self._messages.append(int(payload)) + self._state_changed.set() async def __aiter__(self) -> AsyncIterator[Any]: """Iterate over the queue.""" - while not self.connection.is_closed(): - message_id = await self._queue.get() - yield message_id + while True: + if self._stopping: + return + if self._disconnected: + raise ListenerDisconnectedError( + "PostgreSQL listener connection was lost.", + ) + if self._messages: + yield self._messages.popleft() + continue + + self._state_changed.clear() + await self._state_changed.wait() diff --git a/taskiq_postgresql/exceptions.py b/taskiq_postgresql/exceptions.py index 43b7ab8..8991074 100644 --- a/taskiq_postgresql/exceptions.py +++ b/taskiq_postgresql/exceptions.py @@ -6,5 +6,13 @@ class DatabaseConnectionError(BaseTaskiqAsyncpgError): """Error if cannot connect to PostgreSQL.""" +class TransientDatabaseConnectionError(DatabaseConnectionError): + """Error for a temporary PostgreSQL connection failure.""" + + +class ListenerDisconnectedError(TransientDatabaseConnectionError): + """Error raised when an established listener connection is lost.""" + + class ResultIsMissingError(BaseTaskiqAsyncpgError): """Error if cannot retrieve result from PostgreSQL.""" diff --git a/tests/test_asyncpg_listener.py b/tests/test_asyncpg_listener.py new file mode 100644 index 0000000..c411b45 --- /dev/null +++ b/tests/test_asyncpg_listener.py @@ -0,0 +1,171 @@ +import asyncio +from collections.abc import Callable +from unittest.mock import AsyncMock, MagicMock + +import pytest + +pytest.importorskip("asyncpg") + +from taskiq_postgresql.drivers import _asyncpg +from taskiq_postgresql.drivers._asyncpg import AsyncpgListenDriver +from taskiq_postgresql.exceptions import ( + DatabaseConnectionError, + ListenerDisconnectedError, + TransientDatabaseConnectionError, +) + +pytestmark = pytest.mark.anyio +_MESSAGE_ID = 42 + + +def make_connection() -> MagicMock: + """Create the asyncpg connection boundary used by listener tests.""" + connection = MagicMock() + connection.add_listener = AsyncMock() + connection.remove_listener = AsyncMock() + connection.close = AsyncMock() + connection.is_closed.return_value = False + return connection + + +async def test_termination_wakes_pending_listener( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An unexpected connection close must be observable without a notification.""" + connection = make_connection() + termination_callback: Callable[[object], None] | None = None + + def capture_termination_callback(callback: Callable[[object], None]) -> None: + nonlocal termination_callback + termination_callback = callback + + connection.add_termination_listener.side_effect = capture_termination_callback + monkeypatch.setattr(_asyncpg, "connect", AsyncMock(return_value=connection)) + driver = AsyncpgListenDriver("postgresql://unused", "taskiq") + await driver.on_startup() + + pending_message = asyncio.create_task(driver.__aiter__().__anext__()) + await asyncio.sleep(0) + assert termination_callback is not None + termination_callback(connection) + + with pytest.raises(ListenerDisconnectedError): + await asyncio.wait_for(pending_message, timeout=0.1) + + +async def test_shutdown_wakes_listener_and_closes_connection_once( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Intentional shutdown must stop iteration and remain idempotent.""" + connection = make_connection() + monkeypatch.setattr(_asyncpg, "connect", AsyncMock(return_value=connection)) + driver = AsyncpgListenDriver("postgresql://unused", "taskiq") + await driver.on_startup() + pending_message = asyncio.create_task(driver.__aiter__().__anext__()) + await asyncio.sleep(0) + + await driver.on_shutdown() + await driver.on_shutdown() + + with pytest.raises(StopAsyncIteration): + await asyncio.wait_for(pending_message, timeout=0.1) + connection.remove_termination_listener.assert_called_once() + connection.remove_listener.assert_awaited_once() + connection.close.assert_awaited_once() + + +async def test_partial_listener_startup_closes_connection( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failed LISTEN registration must not leak the connected session.""" + connection = make_connection() + connection.add_listener.side_effect = OSError("connection lost") + monkeypatch.setattr(_asyncpg, "connect", AsyncMock(return_value=connection)) + driver = AsyncpgListenDriver("postgresql://unused", "taskiq") + + with pytest.raises(TransientDatabaseConnectionError): + await driver.on_startup() + + assert driver.connection is None + connection.close.assert_awaited_once() + + +async def test_listener_timeout_is_transient( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Asyncio timeouts remain retryable on Python versions before 3.11.""" + monkeypatch.setattr( + _asyncpg, + "connect", + AsyncMock(side_effect=asyncio.TimeoutError), + ) + driver = AsyncpgListenDriver("postgresql://unused", "taskiq") + + with pytest.raises(TransientDatabaseConnectionError): + await driver.on_startup() + + +async def test_closed_connection_interface_error_is_transient( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A socket closed during LISTEN setup is a transient transport race.""" + connection = make_connection() + connection.is_closed.return_value = True + connection.add_listener.side_effect = _asyncpg.InterfaceError("closed") + monkeypatch.setattr(_asyncpg, "connect", AsyncMock(return_value=connection)) + driver = AsyncpgListenDriver("postgresql://unused", "taskiq") + + with pytest.raises(TransientDatabaseConnectionError): + await driver.on_startup() + + connection.close.assert_awaited_once() + + +async def test_invalid_listener_configuration_is_terminal( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Configuration failures must not enter transient reconnect policy.""" + monkeypatch.setattr( + _asyncpg, + "connect", + AsyncMock(side_effect=ValueError("invalid connection option")), + ) + driver = AsyncpgListenDriver("postgresql://unused", "taskiq") + + with pytest.raises(DatabaseConnectionError) as error_info: + await driver.on_startup() + + assert not isinstance(error_info.value, TransientDatabaseConnectionError) + + +async def test_stale_termination_callback_cannot_poison_new_generation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A callback queued by the old connection must not stop the new listener.""" + first_connection = make_connection() + second_connection = make_connection() + callbacks: list[Callable[[object], None]] = [] + first_connection.add_termination_listener.side_effect = callbacks.append + second_connection.add_termination_listener.side_effect = callbacks.append + monkeypatch.setattr( + _asyncpg, + "connect", + AsyncMock(side_effect=[first_connection, second_connection]), + ) + driver = AsyncpgListenDriver("postgresql://unused", "taskiq") + await driver.on_startup() + await driver.on_shutdown() + await driver.on_startup() + + callbacks[0](first_connection) + driver._notification_handler( + second_connection, + 1, + "taskiq", + str(_MESSAGE_ID), + ) + + assert ( + await asyncio.wait_for(driver.__aiter__().__anext__(), timeout=0.1) + == _MESSAGE_ID + ) diff --git a/tests/test_broker.py b/tests/test_broker.py index d532bf1..483fac1 100644 --- a/tests/test_broker.py +++ b/tests/test_broker.py @@ -1,6 +1,8 @@ import asyncio import json +import os import uuid +from collections.abc import AsyncGenerator from typing import Literal, Union import pytest @@ -8,6 +10,7 @@ from taskiq.utils import maybe_awaitable from taskiq_postgresql import PostgresqlBroker +from taskiq_postgresql.abc.query import BROKER_CHANNEL_LABEL from taskiq_postgresql.exceptions import DatabaseConnectionError pytestmark = pytest.mark.anyio @@ -27,6 +30,31 @@ async def get_first_task( return b"" +@pytest.fixture +async def recovery_broker( + postgresql_dsn: str, + postgres_table: str, +) -> AsyncGenerator[PostgresqlBroker, None]: + """Create the asyncpg broker used by listener-recovery regressions.""" + if os.environ.get("TEST_DRIVER") not in (None, "asyncpg"): + pytest.skip("listener recovery is implemented for asyncpg") + pytest.importorskip("asyncpg") + dsn = os.environ.get("TEST_DATABASE_URL") or postgresql_dsn + broker = PostgresqlBroker( + dsn=dsn, + channel_name=f"{postgres_table}_channel", + table_name=postgres_table, + driver="asyncpg", + run_migrations=True, + ) + await broker.startup() + yield broker + + async with broker.driver, broker.driver.connection() as connection: + await connection.execute(f"DROP TABLE {postgres_table}") + await broker.shutdown() + + @pytest.mark.parametrize("driver", ["asyncpg", "psqlpy", "psycopg"]) async def test_failure_connection_database( driver: Literal["asyncpg", "psqlpy", "psycopg"], @@ -139,6 +167,198 @@ async def test_listen( await maybe_awaitable(message.ack()) +async def test_reconciliation_is_scoped_to_message_channel( + recovery_broker: PostgresqlBroker, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A missed notification is recovered only by its logical channel.""" + broker = recovery_broker + + original_labels = {"source": "channel-scope-test"} + expected = BrokerMessage( + task_id=uuid.uuid4().hex, + task_name="test_task", + message=b"missed-notification", + labels=original_labels, + ) + + async def skip_notification(_message_id: int) -> None: + return None + + monkeypatch.setattr(broker, "_send_notification", skip_notification) + await broker.kick(expected) + + rows = await broker.driver.select([broker.columns.labels]) + stored_labels = rows[0][broker.columns.labels.name] + if isinstance(stored_labels, str): + stored_labels = json.loads(stored_labels) + assert isinstance(stored_labels, dict) + assert stored_labels[BROKER_CHANNEL_LABEL] == broker.channel_name + assert expected.labels == original_labels + + other_broker = PostgresqlBroker( + dsn=broker.dsn, + channel_name=f"{broker.channel_name}_other", + table_name=broker.table_name, + driver="asyncpg", + ) + await other_broker.startup() + try: + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(get_first_task(other_broker), timeout=0.1) + + message = await asyncio.wait_for(get_first_task(broker), timeout=1.0) + finally: + await other_broker.shutdown() + + assert message.data == expected.message + + +async def test_reconciliation_and_queued_notification_claim_once( + recovery_broker: PostgresqlBroker, +) -> None: + """A scanned row and its queued NOTIFY must not yield two messages.""" + broker = recovery_broker + + message_id = await broker.driver.insert( + columns=[ + broker.columns.task_id, + broker.columns.task_name, + broker.columns.message, + broker.columns.labels, + ], + values=[ + uuid.uuid4().hex, + "test_task", + b"single-delivery", + json.dumps({BROKER_CHANNEL_LABEL: broker.channel_name}), + ], + returning=[broker.columns.primary_key], + ) + await broker.driver.execute( + f"NOTIFY {broker.channel_name}, '{message_id}'", + ) + await asyncio.sleep(0.05) + + message = await asyncio.wait_for(get_first_task(broker), timeout=1.0) + + assert message.data == b"single-delivery" + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(get_first_task(broker), timeout=0.1) + + +async def test_reconciliation_delivers_only_due_rows( + recovery_broker: PostgresqlBroker, +) -> None: + """Due rows are recovered without releasing future or malformed delays.""" + broker = recovery_broker + + rows = [ + (b"due-numeric", {"delay": 1.9}), + (b"future", {"delay": 60}), + (b"malformed", {"delay": "not-an-integer"}), + (b"due-boolean", {"delay": False}), + ] + for message, labels in rows: + await broker.driver.insert( + columns=[ + broker.columns.task_id, + broker.columns.task_name, + broker.columns.message, + broker.columns.labels, + ], + values=[ + uuid.uuid4().hex, + "test_task", + message, + json.dumps( + { + **labels, + BROKER_CHANNEL_LABEL: broker.channel_name, + }, + ), + ], + returning=[broker.columns.primary_key], + ) + legacy_message_id = await broker.driver.insert( + columns=[ + broker.columns.task_id, + broker.columns.task_name, + broker.columns.message, + broker.columns.labels, + ], + values=[ + uuid.uuid4().hex, + "test_task", + b"legacy-unmarked", + json.dumps({}), + ], + returning=[broker.columns.primary_key], + ) + await broker.driver.execute( + f"INSERT INTO {broker.table_name} " # noqa: S608 + "(task_id, task_name, message, labels) " + "VALUES ($1::text::uuid, $2, $3, " + "to_jsonb('invalid-label-shape'::text))", + uuid.uuid4().hex, + "test_task", + b"scalar-labels", + ) + + await asyncio.sleep(1.1) + iterator = broker.listen() + recovered = [(await iterator.__anext__()).data for _ in range(2)] + await iterator.aclose() + + assert recovered == [b"due-numeric", b"due-boolean"] + remaining = await broker.driver.select([broker.columns.message]) + assert {row[broker.columns.message.name] for row in remaining} == { + b"future", + b"malformed", + b"legacy-unmarked", + b"scalar-labels", + } + + await broker.driver.execute( + f"NOTIFY {broker.channel_name}, '{legacy_message_id}'", + ) + legacy_message = await asyncio.wait_for(get_first_task(broker), timeout=1.0) + assert legacy_message.data == b"legacy-unmarked" + + +async def test_asyncpg_listener_recovers_without_restarting_consumer( + recovery_broker: PostgresqlBroker, +) -> None: + """A killed asyncpg LISTEN backend must reconnect and reconcile its gap.""" + broker = recovery_broker + + old_listener = broker.listen_driver + old_connection = old_listener.connection + listener_pid = await old_connection.fetchval("SELECT pg_backend_pid()") + broker._listener_retry_delay = lambda _attempt: 0.2 + consumer = asyncio.create_task(get_first_task(broker)) + await asyncio.sleep(0.05) + + await broker.driver.execute("SELECT pg_terminate_backend($1)", listener_pid) + for _ in range(20): + if old_connection.is_closed(): + break + await asyncio.sleep(0.01) + assert old_connection.is_closed() + + expected = BrokerMessage( + task_id=uuid.uuid4().hex, + task_name="test_task", + message=b"published-during-listener-gap", + labels={}, + ) + await broker.kick(expected) + + received = await asyncio.wait_for(consumer, timeout=3.0) + assert received.data == expected.message + assert broker.listen_driver is not old_listener + + async def test_wrong_format( broker: PostgresqlBroker, ) -> None: diff --git a/tests/test_listener_recovery.py b/tests/test_listener_recovery.py new file mode 100644 index 0000000..ccad662 --- /dev/null +++ b/tests/test_listener_recovery.py @@ -0,0 +1,422 @@ +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator, Sequence +from typing import Any + +import pytest + +from taskiq_postgresql.broker import PostgresqlBroker +from taskiq_postgresql.exceptions import ( + ListenerDisconnectedError, + TransientDatabaseConnectionError, +) + +pytestmark = pytest.mark.anyio + +_DISCONNECT = object() +_EXPECTED_QUERY_ATTEMPTS = 2 +_MAX_RETRY_DELAY = 30.0 +_RECONCILIATION_BATCH_SIZE = 100 +_RECONCILIATION_WATERMARK = 205 + + +class FakeQueryDriver: + """Provide only the query operations exercised by broker.listen().""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + self.shutdown_calls = 0 + + async def on_startup(self) -> None: + """Start the fake query driver.""" + + async def on_shutdown(self) -> None: + """Stop the fake query driver.""" + self.shutdown_calls += 1 + + async def delete_returning( + self, + *args: Any, + **kwargs: Any, + ) -> dict[str, bytes]: + """Return a message for every claimed notification ID.""" + return {"message": b"payload"} + + async def max_value(self, *args: Any, **kwargs: Any) -> int: + """Return an empty reconciliation high-watermark.""" + return 0 + + async def select_available_ids( + self, + *args: Any, + **kwargs: Any, + ) -> list[int]: + """Return no rows for reconciliation.""" + return [] + + +class ReconciliationQueryDriver(FakeQueryDriver): + """Expose a backlog large enough to require multiple bounded pages.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.page_requests: list[tuple[int, int, str, int]] = [] + + async def max_value(self, *args: Any, **kwargs: Any) -> int: + """Return a stable high-watermark for this reconciliation pass.""" + return _RECONCILIATION_WATERMARK + + async def select_available_ids( + self, + *args: Any, + **kwargs: Any, + ) -> list[int]: + """Return one keyset page and record its explicit bounds.""" + after_id = int(kwargs["after_id"]) + through_id = int(kwargs["through_id"]) + channel_name = str(kwargs["channel_name"]) + limit = int(kwargs["limit"]) + self.page_requests.append((after_id, through_id, channel_name, limit)) + page_end = min(after_id + limit, through_id) + return list(range(after_id + 1, page_end + 1)) + + +class FlakyQueryDriver(FakeQueryDriver): + """Fail the first reconciliation query with a transient connection error.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.max_value_calls = 0 + + async def max_value(self, *args: Any, **kwargs: Any) -> int: + """Fail once, then allow the listener path to continue.""" + self.max_value_calls += 1 + if self.max_value_calls == 1: + raise TransientDatabaseConnectionError("temporarily unavailable") + return 0 + + +class ScriptedListenDriver: + """Run a deterministic listener startup and iteration script.""" + + def __init__( + self, + events: Sequence[object] = (), + startup_error: Exception | None = None, + ) -> None: + self.events = events + self.startup_error = startup_error + self.started = asyncio.Event() + self.iterated = asyncio.Event() + self.startup_calls = 0 + self.shutdown_calls = 0 + + async def on_startup(self) -> None: + """Start this listener generation or raise its scripted error.""" + self.startup_calls += 1 + self.started.set() + if self.startup_error is not None: + raise self.startup_error + + async def on_shutdown(self) -> None: + """Record cleanup of this listener generation.""" + self.shutdown_calls += 1 + + async def __aiter__(self) -> AsyncIterator[int]: + """Yield IDs, disconnect, or complete according to the script.""" + self.iterated.set() + for event in self.events: + if event is _DISCONNECT: + raise ListenerDisconnectedError("listener disconnected") + yield int(event) + + +class BlockingStartupListenDriver(ScriptedListenDriver): + """Keep listener startup pending until the broker cancels it.""" + + async def on_startup(self) -> None: + """Publish startup progress and wait indefinitely.""" + self.startup_calls += 1 + self.started.set() + await asyncio.Event().wait() + + +class ListenerFactory: + """Return one prebuilt listener for each requested generation.""" + + def __init__(self, listeners: Sequence[ScriptedListenDriver]) -> None: + self.listeners = listeners + self.calls: list[ScriptedListenDriver] = [] + + def __call__(self, *args: Any, **kwargs: Any) -> ScriptedListenDriver: + """Return the next scripted listener.""" + listener = self.listeners[len(self.calls)] + self.calls.append(listener) + return listener + + +async def make_broker( + monkeypatch: pytest.MonkeyPatch, + listeners: Sequence[ScriptedListenDriver], + query_driver_class: type[FakeQueryDriver] = FakeQueryDriver, +) -> tuple[PostgresqlBroker, ListenerFactory]: + """Construct and start a broker backed entirely by deterministic fakes.""" + factory = ListenerFactory(listeners) + monkeypatch.setattr( + "taskiq_postgresql.broker.get_db_driver", + lambda _driver: query_driver_class, + ) + monkeypatch.setattr( + "taskiq_postgresql.broker.get_db_listen_driver", + lambda _driver: factory, + ) + + broker = PostgresqlBroker(dsn="postgresql://unused") + broker._listener_retry_delay = lambda _attempt: 0.0 + await broker.startup() + return broker, factory + + +async def next_message(broker: PostgresqlBroker) -> bytes: + """Return the data from the next broker message.""" + iterator = broker.listen() + try: + return (await iterator.__anext__()).data + finally: + await iterator.aclose() + + +async def test_disconnect_replaces_listener_with_fresh_generation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + initial = ScriptedListenDriver([_DISCONNECT]) + recovered = ScriptedListenDriver([1]) + broker, factory = await make_broker(monkeypatch, [initial, recovered]) + + assert await next_message(broker) == b"payload" + assert factory.calls == [initial, recovered] + assert broker.listen_driver is recovered + assert initial.shutdown_calls == 1 + assert recovered.startup_calls == 1 + + await broker.shutdown() + + +async def test_startup_is_idempotent( + monkeypatch: pytest.MonkeyPatch, +) -> None: + listener = ScriptedListenDriver([1]) + broker, factory = await make_broker(monkeypatch, [listener]) + + await broker.startup() + + assert factory.calls == [listener] + assert listener.startup_calls == 1 + + await broker.shutdown() + + +async def test_transient_startup_failure_is_retried_with_new_listener( + monkeypatch: pytest.MonkeyPatch, +) -> None: + initial = ScriptedListenDriver([_DISCONNECT]) + failed = ScriptedListenDriver( + startup_error=TransientDatabaseConnectionError("temporarily unavailable"), + ) + recovered = ScriptedListenDriver([1]) + broker, factory = await make_broker(monkeypatch, [initial, failed, recovered]) + + assert await next_message(broker) == b"payload" + assert factory.calls == [initial, failed, recovered] + assert failed.shutdown_calls == 1 + assert broker.listen_driver is recovered + + await broker.shutdown() + + +async def test_terminal_reconnect_error_propagates( + monkeypatch: pytest.MonkeyPatch, +) -> None: + initial = ScriptedListenDriver([_DISCONNECT]) + terminal_error = RuntimeError("invalid listener contract") + failed = ScriptedListenDriver(startup_error=terminal_error) + broker, factory = await make_broker(monkeypatch, [initial, failed]) + + iterator = broker.listen() + with pytest.raises(RuntimeError, match="invalid listener contract"): + await iterator.__anext__() + await iterator.aclose() + + assert factory.calls == [initial, failed] + assert failed.shutdown_calls == 1 + + await broker.shutdown() + + +async def test_shutdown_interrupts_reconnect_backoff( + monkeypatch: pytest.MonkeyPatch, +) -> None: + initial = ScriptedListenDriver([_DISCONNECT]) + unused = ScriptedListenDriver([1]) + broker, factory = await make_broker(monkeypatch, [initial, unused]) + broker._listener_retry_delay = lambda _attempt: 3600.0 + + iterator = broker.listen() + pending_message = asyncio.create_task(iterator.__anext__()) + await asyncio.wait_for(initial.iterated.wait(), timeout=1) + await asyncio.sleep(0) + + await broker.shutdown() + + with pytest.raises(StopAsyncIteration): + await asyncio.wait_for(pending_message, timeout=1) + assert factory.calls == [initial] + assert broker._shutdown_event.is_set() + await iterator.aclose() + + +async def test_clean_iterator_completion_triggers_reconnect( + monkeypatch: pytest.MonkeyPatch, +) -> None: + completed = ScriptedListenDriver() + recovered = ScriptedListenDriver([1]) + broker, factory = await make_broker(monkeypatch, [completed, recovered]) + + assert await next_message(broker) == b"payload" + assert factory.calls == [completed, recovered] + assert completed.shutdown_calls == 1 + + await broker.shutdown() + + +async def test_concurrent_consumers_share_one_recovery_generation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + initial = ScriptedListenDriver([_DISCONNECT]) + recovered = ScriptedListenDriver([1]) + broker, factory = await make_broker(monkeypatch, [initial, recovered]) + + messages = await asyncio.gather(next_message(broker), next_message(broker)) + + assert messages == [b"payload", b"payload"] + assert factory.calls == [initial, recovered] + assert initial.shutdown_calls == 1 + assert recovered.startup_calls == 1 + + await broker.shutdown() + + +async def test_shutdown_cancels_initial_listener_startup( + monkeypatch: pytest.MonkeyPatch, +) -> None: + listener = BlockingStartupListenDriver() + factory = ListenerFactory([listener]) + monkeypatch.setattr( + "taskiq_postgresql.broker.get_db_driver", + lambda _driver: FakeQueryDriver, + ) + monkeypatch.setattr( + "taskiq_postgresql.broker.get_db_listen_driver", + lambda _driver: factory, + ) + broker = PostgresqlBroker(dsn="postgresql://unused") + startup_task = asyncio.create_task(broker.startup()) + await asyncio.wait_for(listener.started.wait(), timeout=1) + + await broker.shutdown() + + with pytest.raises(asyncio.CancelledError): + await startup_task + assert listener.shutdown_calls == 1 + assert broker.driver.shutdown_calls == 1 + assert broker._shutdown_event.is_set() + + +async def test_transient_query_failure_retries_with_backoff( + monkeypatch: pytest.MonkeyPatch, +) -> None: + listener = ScriptedListenDriver([1]) + broker, _ = await make_broker( + monkeypatch, + [listener], + query_driver_class=FlakyQueryDriver, + ) + + assert await next_message(broker) == b"payload" + assert broker.driver.max_value_calls == _EXPECTED_QUERY_ATTEMPTS + + await broker.shutdown() + + +async def test_unexpected_iterator_error_propagates( + monkeypatch: pytest.MonkeyPatch, +) -> None: + broken = ScriptedListenDriver([object()]) + broker, factory = await make_broker(monkeypatch, [broken]) + + with pytest.raises(TypeError): + await next_message(broker) + assert factory.calls == [broken] + + await broker.shutdown() + + +async def test_reconciliation_uses_bounded_keyset_pages( + monkeypatch: pytest.MonkeyPatch, +) -> None: + listener = ScriptedListenDriver() + broker, _ = await make_broker( + monkeypatch, + [listener], + query_driver_class=ReconciliationQueryDriver, + ) + iterator = broker.listen() + + messages = [await iterator.__anext__() for _ in range(_RECONCILIATION_WATERMARK)] + await iterator.aclose() + + assert len(messages) == _RECONCILIATION_WATERMARK + assert broker.driver.page_requests == [ + (0, _RECONCILIATION_WATERMARK, "taskiq", _RECONCILIATION_BATCH_SIZE), + ( + _RECONCILIATION_BATCH_SIZE, + _RECONCILIATION_WATERMARK, + "taskiq", + _RECONCILIATION_BATCH_SIZE, + ), + ( + 2 * _RECONCILIATION_BATCH_SIZE, + _RECONCILIATION_WATERMARK, + "taskiq", + _RECONCILIATION_BATCH_SIZE, + ), + ] + + await broker.shutdown() + + +async def test_listener_backoff_is_capped_without_overflow( + monkeypatch: pytest.MonkeyPatch, +) -> None: + listener = ScriptedListenDriver() + broker, _ = await make_broker(monkeypatch, [listener]) + monkeypatch.setattr("taskiq_postgresql.broker.uniform", lambda _low, _high: 1.0) + + delays = [ + PostgresqlBroker._listener_retry_delay(broker, attempt) + for attempt in range(1, 9) + ] + + assert delays == [ + 0.5, + 1.0, + 2.0, + 4.0, + 8.0, + 16.0, + _MAX_RETRY_DELAY, + _MAX_RETRY_DELAY, + ] + assert PostgresqlBroker._listener_retry_delay(broker, 1_000_000) == _MAX_RETRY_DELAY + + await broker.shutdown() diff --git a/tests/test_reconciliation_queries.py b/tests/test_reconciliation_queries.py new file mode 100644 index 0000000..e0a22c2 --- /dev/null +++ b/tests/test_reconciliation_queries.py @@ -0,0 +1,59 @@ +from taskiq_postgresql.abc.query import ( + BROKER_CHANNEL_LABEL, + Column, + CreatedAtColumn, + MaxValueQuery, + PrimaryKeyColumn, + SelectAvailableIdsQuery, +) + + +def _normalize(query: str) -> str: + """Collapse SQL whitespace so tests focus on query semantics.""" + return " ".join(query.split()) + + +def test_max_value_query_uses_the_requested_column() -> None: + query = MaxValueQuery("taskiq_messages") + + assert query.make_query(PrimaryKeyColumn("message_id")) == ( + "SELECT COALESCE(MAX(message_id), 0) FROM taskiq_messages" + ) + + +def test_select_available_ids_query_uses_bounded_keyset_pagination() -> None: + query = SelectAvailableIdsQuery("taskiq_messages").make_query( + primary_key=PrimaryKeyColumn("message_id"), + labels=Column("task_labels", "JSONB"), + created_at=Column("inserted_at", "TIMESTAMP WITH TIME ZONE"), + ) + normalized = _normalize(query) + + assert normalized.startswith("SELECT message_id FROM taskiq_messages") + assert "WHERE message_id > $1 AND message_id <= $2" in normalized + assert f"task_labels->>'{BROKER_CHANNEL_LABEL}' = $3" in normalized + assert normalized.endswith("ORDER BY message_id LIMIT $4") + assert [normalized.index(f"${index}") for index in range(1, 5)] == sorted( + normalized.index(f"${index}") for index in range(1, 5) + ) + assert normalized.count("$1") == 1 + assert normalized.count("$2") == 1 + assert normalized.count("$3") == 1 + assert normalized.count("$4") == 1 + + +def test_select_available_ids_query_filters_only_due_delayed_rows() -> None: + query = SelectAvailableIdsQuery("taskiq_messages").make_query( + primary_key=PrimaryKeyColumn(), + labels=Column("labels", "JSONB"), + created_at=CreatedAtColumn(), + ) + normalized = _normalize(query) + + assert "labels->>'delay' IS NULL" in normalized + assert "EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - created_at)) >= CASE" in normalized + assert "jsonb_typeof(labels->'delay') = 'number'" in normalized + assert "trunc((labels->>'delay')::numeric)" in normalized + assert "jsonb_typeof(labels->'delay') = 'string'" in normalized + assert "btrim(labels->>'delay') ~ '^[+-]?[0-9]+$'" in normalized + assert "btrim(labels->>'delay')::numeric" in normalized diff --git a/uv.lock b/uv.lock index 45db20a..efa06cd 100644 --- a/uv.lock +++ b/uv.lock @@ -148,10 +148,10 @@ name = "bandit" version = "1.8.6" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "colorama", marker = "python_full_version < '3.10' and sys_platform == 'win32'" }, - { name = "pyyaml", marker = "python_full_version < '3.10'" }, - { name = "rich", marker = "python_full_version < '3.10'" }, - { name = "stevedore", marker = "python_full_version < '3.10'" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "pyyaml" }, + { name = "rich" }, + { name = "stevedore" }, ] sdist = { url = "https://files.pythonhosted.org/packages/fb/b5/7eb834e213d6f73aace21938e5e90425c92e5f42abafaf8a6d5d21beed51/bandit-1.8.6.tar.gz", hash = "sha256:dbfe9c25fc6961c2078593de55fd19f2559f9e45b99f1272341f5b95dea4e56b", size = 4240271, upload-time = "2025-07-06T03:10:50.9Z" } wheels = [ @@ -217,38 +217,28 @@ sdist = { url = "https://files.pythonhosted.org/packages/fc/97/c783634659c2920c3 wheels = [ { url = "https://files.pythonhosted.org/packages/de/cc/4635c320081c78d6ffc2cab0a76025b691a91204f4aa317d568ff9280a2d/cffi-1.17.1-cp310-cp310-manylinux_2_12_i686.manylinux2010_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:edae79245293e15384b51f88b00613ba9f7198016a5948b5dddf4917d4d26382", size = 426024, upload-time = "2024-09-04T20:43:34.186Z" }, { url = "https://files.pythonhosted.org/packages/b6/7b/3b2b250f3aab91abe5f8a51ada1b717935fdaec53f790ad4100fe2ec64d1/cffi-1.17.1-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:45398b671ac6d70e67da8e4224a065cec6a93541bb7aebe1b198a61b58c7b702", size = 448188, upload-time = "2024-09-04T20:43:36.286Z" }, - { url = "https://files.pythonhosted.org/packages/d3/48/1b9283ebbf0ec065148d8de05d647a986c5f22586b18120020452fff8f5d/cffi-1.17.1-cp310-cp310-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:ad9413ccdeda48c5afdae7e4fa2192157e991ff761e7ab8fdd8926f40b160cc3", size = 455571, upload-time = "2024-09-04T20:43:38.586Z" }, - { url = "https://files.pythonhosted.org/packages/40/87/3b8452525437b40f39ca7ff70276679772ee7e8b394934ff60e63b7b090c/cffi-1.17.1-cp310-cp310-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:5da5719280082ac6bd9aa7becb3938dc9f9cbd57fac7d2871717b1feb0902ab6", size = 436687, upload-time = "2024-09-04T20:43:40.084Z" }, { url = "https://files.pythonhosted.org/packages/8d/fb/4da72871d177d63649ac449aec2e8a29efe0274035880c7af59101ca2232/cffi-1.17.1-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2bb1a08b8008b281856e5971307cc386a8e9c5b625ac297e853d36da6efe9c17", size = 446211, upload-time = "2024-09-04T20:43:41.526Z" }, { url = "https://files.pythonhosted.org/packages/ab/a0/62f00bcb411332106c02b663b26f3545a9ef136f80d5df746c05878f8c4b/cffi-1.17.1-cp310-cp310-musllinux_1_1_aarch64.whl", hash = "sha256:045d61c734659cc045141be4bae381a41d89b741f795af1dd018bfb532fd0df8", size = 461325, upload-time = "2024-09-04T20:43:43.117Z" }, { url = "https://files.pythonhosted.org/packages/36/83/76127035ed2e7e27b0787604d99da630ac3123bfb02d8e80c633f218a11d/cffi-1.17.1-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:6883e737d7d9e4899a8a695e00ec36bd4e5e4f18fabe0aca0efe0a4b44cdb13e", size = 438784, upload-time = "2024-09-04T20:43:45.256Z" }, { url = "https://files.pythonhosted.org/packages/21/81/a6cd025db2f08ac88b901b745c163d884641909641f9b826e8cb87645942/cffi-1.17.1-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:6b8b4a92e1c65048ff98cfe1f735ef8f1ceb72e3d5f0c25fdb12087a23da22be", size = 461564, upload-time = "2024-09-04T20:43:46.779Z" }, { url = "https://files.pythonhosted.org/packages/94/dd/a3f0118e688d1b1a57553da23b16bdade96d2f9bcda4d32e7d2838047ff7/cffi-1.17.1-cp311-cp311-manylinux_2_12_i686.manylinux2010_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:f75c7ab1f9e4aca5414ed4d8e5c0e303a34f4421f8a0d47a4d019ceff0ab6af4", size = 445259, upload-time = "2024-09-04T20:43:56.123Z" }, { url = "https://files.pythonhosted.org/packages/2e/ea/70ce63780f096e16ce8588efe039d3c4f91deb1dc01e9c73a287939c79a6/cffi-1.17.1-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a1ed2dd2972641495a3ec98445e09766f077aee98a1c896dcb4ad0d303628e41", size = 469200, upload-time = "2024-09-04T20:43:57.891Z" }, - { url = "https://files.pythonhosted.org/packages/1c/a0/a4fa9f4f781bda074c3ddd57a572b060fa0df7655d2a4247bbe277200146/cffi-1.17.1-cp311-cp311-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:46bf43160c1a35f7ec506d254e5c890f3c03648a4dbac12d624e4490a7046cd1", size = 477235, upload-time = "2024-09-04T20:44:00.18Z" }, - { url = "https://files.pythonhosted.org/packages/62/12/ce8710b5b8affbcdd5c6e367217c242524ad17a02fe5beec3ee339f69f85/cffi-1.17.1-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:a24ed04c8ffd54b0729c07cee15a81d964e6fee0e3d4d342a27b020d22959dc6", size = 459721, upload-time = "2024-09-04T20:44:01.585Z" }, { url = "https://files.pythonhosted.org/packages/ff/6b/d45873c5e0242196f042d555526f92aa9e0c32355a1be1ff8c27f077fd37/cffi-1.17.1-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:610faea79c43e44c71e1ec53a554553fa22321b65fae24889706c0a84d4ad86d", size = 467242, upload-time = "2024-09-04T20:44:03.467Z" }, { url = "https://files.pythonhosted.org/packages/1a/52/d9a0e523a572fbccf2955f5abe883cfa8bcc570d7faeee06336fbd50c9fc/cffi-1.17.1-cp311-cp311-musllinux_1_1_aarch64.whl", hash = "sha256:a9b15d491f3ad5d692e11f6b71f7857e7835eb677955c00cc0aefcd0669adaf6", size = 477999, upload-time = "2024-09-04T20:44:05.023Z" }, { url = "https://files.pythonhosted.org/packages/44/74/f2a2460684a1a2d00ca799ad880d54652841a780c4c97b87754f660c7603/cffi-1.17.1-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:de2ea4b5833625383e464549fec1bc395c1bdeeb5f25c4a3a82b5a8c756ec22f", size = 454242, upload-time = "2024-09-04T20:44:06.444Z" }, { url = "https://files.pythonhosted.org/packages/f8/4a/34599cac7dfcd888ff54e801afe06a19c17787dfd94495ab0c8d35fe99fb/cffi-1.17.1-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:fc48c783f9c87e60831201f2cce7f3b2e4846bf4d8728eabe54d60700b318a0b", size = 478604, upload-time = "2024-09-04T20:44:08.206Z" }, { url = "https://files.pythonhosted.org/packages/cc/b6/db007700f67d151abadf508cbfd6a1884f57eab90b1bb985c4c8c02b0f28/cffi-1.17.1-cp312-cp312-manylinux_2_12_i686.manylinux2010_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:1257bdabf294dceb59f5e70c64a3e2f462c30c7ad68092d01bbbfb1c16b1ba36", size = 454803, upload-time = "2024-09-04T20:44:15.231Z" }, { url = "https://files.pythonhosted.org/packages/1a/df/f8d151540d8c200eb1c6fba8cd0dfd40904f1b0682ea705c36e6c2e97ab3/cffi-1.17.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:da95af8214998d77a98cc14e3a3bd00aa191526343078b530ceb0bd710fb48a5", size = 478850, upload-time = "2024-09-04T20:44:17.188Z" }, - { url = "https://files.pythonhosted.org/packages/28/c0/b31116332a547fd2677ae5b78a2ef662dfc8023d67f41b2a83f7c2aa78b1/cffi-1.17.1-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:d63afe322132c194cf832bfec0dc69a99fb9bb6bbd550f161a49e9e855cc78ff", size = 485729, upload-time = "2024-09-04T20:44:18.688Z" }, - { url = "https://files.pythonhosted.org/packages/91/2b/9a1ddfa5c7f13cab007a2c9cc295b70fbbda7cb10a286aa6810338e60ea1/cffi-1.17.1-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:f79fc4fc25f1c8698ff97788206bb3c2598949bfe0fef03d299eb1b5356ada99", size = 471256, upload-time = "2024-09-04T20:44:20.248Z" }, { url = "https://files.pythonhosted.org/packages/b2/d5/da47df7004cb17e4955df6a43d14b3b4ae77737dff8bf7f8f333196717bf/cffi-1.17.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:b62ce867176a75d03a665bad002af8e6d54644fad99a3c70905c543130e39d93", size = 479424, upload-time = "2024-09-04T20:44:21.673Z" }, { url = "https://files.pythonhosted.org/packages/0b/ac/2a28bcf513e93a219c8a4e8e125534f4f6db03e3179ba1c45e949b76212c/cffi-1.17.1-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:386c8bf53c502fff58903061338ce4f4950cbdcb23e2902d86c0f722b786bbe3", size = 484568, upload-time = "2024-09-04T20:44:23.245Z" }, { url = "https://files.pythonhosted.org/packages/d4/38/ca8a4f639065f14ae0f1d9751e70447a261f1a30fa7547a828ae08142465/cffi-1.17.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:4ceb10419a9adf4460ea14cfd6bc43d08701f0835e979bf821052f1805850fe8", size = 488736, upload-time = "2024-09-04T20:44:24.757Z" }, { url = "https://files.pythonhosted.org/packages/0e/2d/eab2e858a91fdff70533cab61dcff4a1f55ec60425832ddfdc9cd36bc8af/cffi-1.17.1-cp313-cp313-manylinux_2_12_i686.manylinux2010_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:d01b12eeeb4427d3110de311e1774046ad344f5b1a7403101878976ecd7a10f3", size = 454792, upload-time = "2024-09-04T20:44:32.01Z" }, { url = "https://files.pythonhosted.org/packages/75/b2/fbaec7c4455c604e29388d55599b99ebcc250a60050610fadde58932b7ee/cffi-1.17.1-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:706510fe141c86a69c8ddc029c7910003a17353970cff3b904ff0686a5927683", size = 478893, upload-time = "2024-09-04T20:44:33.606Z" }, - { url = "https://files.pythonhosted.org/packages/4f/b7/6e4a2162178bf1935c336d4da8a9352cccab4d3a5d7914065490f08c0690/cffi-1.17.1-cp313-cp313-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:de55b766c7aa2e2a3092c51e0483d700341182f08e67c63630d5b6f200bb28e5", size = 485810, upload-time = "2024-09-04T20:44:35.191Z" }, - { url = "https://files.pythonhosted.org/packages/c7/8a/1d0e4a9c26e54746dc08c2c6c037889124d4f59dffd853a659fa545f1b40/cffi-1.17.1-cp313-cp313-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:c59d6e989d07460165cc5ad3c61f9fd8f1b4796eacbd81cee78957842b834af4", size = 471200, upload-time = "2024-09-04T20:44:36.743Z" }, { url = "https://files.pythonhosted.org/packages/26/9f/1aab65a6c0db35f43c4d1b4f580e8df53914310afc10ae0397d29d697af4/cffi-1.17.1-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:dd398dbc6773384a17fe0d3e7eeb8d1a21c2200473ee6806bb5e6a8e62bb73dd", size = 479447, upload-time = "2024-09-04T20:44:38.492Z" }, { url = "https://files.pythonhosted.org/packages/5f/e4/fb8b3dd8dc0e98edf1135ff067ae070bb32ef9d509d6cb0f538cd6f7483f/cffi-1.17.1-cp313-cp313-musllinux_1_1_aarch64.whl", hash = "sha256:3edc8d958eb099c634dace3c7e16560ae474aa3803a5df240542b305d14e14ed", size = 484358, upload-time = "2024-09-04T20:44:40.046Z" }, { url = "https://files.pythonhosted.org/packages/f1/47/d7145bf2dc04684935d57d67dff9d6d795b2ba2796806bb109864be3a151/cffi-1.17.1-cp313-cp313-musllinux_1_1_x86_64.whl", hash = "sha256:72e72408cad3d5419375fc87d289076ee319835bdfa2caad331e377589aebba9", size = 488469, upload-time = "2024-09-04T20:44:41.616Z" }, { url = "https://files.pythonhosted.org/packages/ed/65/25a8dc32c53bf5b7b6c2686b42ae2ad58743f7ff644844af7cdb29b49361/cffi-1.17.1-cp39-cp39-manylinux_2_12_i686.manylinux2010_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:1d599671f396c4723d016dbddb72fe8e0397082b0a77a4fab8028923bec050e8", size = 424910, upload-time = "2024-09-04T20:45:05.315Z" }, { url = "https://files.pythonhosted.org/packages/42/7a/9d086fab7c66bd7c4d0f27c57a1b6b068ced810afc498cc8c49e0088661c/cffi-1.17.1-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:ca74b8dbe6e8e8263c0ffd60277de77dcee6c837a3d0881d8c1ead7268c9e576", size = 447200, upload-time = "2024-09-04T20:45:06.903Z" }, - { url = "https://files.pythonhosted.org/packages/da/63/1785ced118ce92a993b0ec9e0d0ac8dc3e5dbfbcaa81135be56c69cabbb6/cffi-1.17.1-cp39-cp39-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:f7f5baafcc48261359e14bcd6d9bff6d4b28d9103847c9e136694cb0501aef87", size = 454565, upload-time = "2024-09-04T20:45:08.975Z" }, - { url = "https://files.pythonhosted.org/packages/74/06/90b8a44abf3556599cdec107f7290277ae8901a58f75e6fe8f970cd72418/cffi-1.17.1-cp39-cp39-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:98e3969bcff97cae1b2def8ba499ea3d6f31ddfdb7635374834cf89a1a08ecf0", size = 435635, upload-time = "2024-09-04T20:45:10.64Z" }, { url = "https://files.pythonhosted.org/packages/bd/62/a1f468e5708a70b1d86ead5bab5520861d9c7eacce4a885ded9faa7729c3/cffi-1.17.1-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:cdf5ce3acdfd1661132f2a9c19cac174758dc2352bfe37d98aa7512c6b7178b3", size = 445218, upload-time = "2024-09-04T20:45:12.366Z" }, { url = "https://files.pythonhosted.org/packages/5b/95/b34462f3ccb09c2594aa782d90a90b045de4ff1f70148ee79c69d37a0a5a/cffi-1.17.1-cp39-cp39-musllinux_1_1_aarch64.whl", hash = "sha256:9755e4345d1ec879e3849e62222a18c7174d65a6a92d5b346b1863912168b595", size = 460486, upload-time = "2024-09-04T20:45:13.935Z" }, { url = "https://files.pythonhosted.org/packages/fc/fc/a1e4bebd8d680febd29cf6c8a40067182b64f00c7d105f8f26b5bc54317b/cffi-1.17.1-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:f1e22e8c4419538cb197e4dd60acc919d7696e5ef98ee4da4e01d3f8cfa4cc5a", size = 437911, upload-time = "2024-09-04T20:45:15.696Z" }, @@ -347,7 +337,7 @@ resolution-markers = [ "python_full_version < '3.10'", ] dependencies = [ - { name = "colorama", marker = "python_full_version < '3.10' and sys_platform == 'win32'" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/b9/2e/0090cbf739cee7d23781ad4b89a9894a41538e4fcf4c31dcdd705b78eb8b/click-8.1.8.tar.gz", hash = "sha256:ed53c9d8990d83c2a27deae68e4ee337473f6330c040a31d4225c9574d16096a", size = 226593, upload-time = "2024-12-21T18:38:44.339Z" } wheels = [ @@ -362,7 +352,7 @@ resolution-markers = [ "python_full_version >= '3.10'", ] dependencies = [ - { name = "colorama", marker = "python_full_version >= '3.10' and sys_platform == 'win32'" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/60/6c/8ca2efa64cf75a977a0d7fac081354553ebe483345c734fb6b6515d96bbc/click-8.2.1.tar.gz", hash = "sha256:27c491cc05d968d271d5a1db13e3b5a184636d9d930f148c50b038f0d0646202", size = 286342, upload-time = "2025-05-20T23:19:49.832Z" } wheels = [ @@ -556,7 +546,7 @@ name = "exceptiongroup" version = "1.3.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/0b/9f/a65090624ecf468cdca03533906e7c69ed7588582240cfe7cc9e770b50eb/exceptiongroup-1.3.0.tar.gz", hash = "sha256:b241f5885f560bc56a59ee63ca4c6a8bfa46ae4ad651af316d4e81817bb9fd88", size = 29749, upload-time = "2025-05-10T17:42:51.123Z" } wheels = [ @@ -600,8 +590,8 @@ name = "flake8-bandit" version = "4.1.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "bandit", marker = "python_full_version < '3.10'" }, - { name = "flake8", marker = "python_full_version < '3.10'" }, + { name = "bandit" }, + { name = "flake8" }, ] sdist = { url = "https://files.pythonhosted.org/packages/77/1c/4f66a7a52a246d6c64312b5c40da3af3630cd60b27af81b137796af3c0bc/flake8_bandit-4.1.1.tar.gz", hash = "sha256:068e09287189cbfd7f986e92605adea2067630b75380c6b5733dab7d87f9a84e", size = 5403, upload-time = "2022-08-29T13:48:41.225Z" } wheels = [ @@ -613,7 +603,7 @@ name = "flake8-broken-line" version = "1.0.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "flake8", marker = "python_full_version < '3.10'" }, + { name = "flake8" }, ] sdist = { url = "https://files.pythonhosted.org/packages/30/5e/eca08446205afb79e74b6af8e227f06f0b1a26ae892708adbc4e65ccaa86/flake8_broken_line-1.0.0.tar.gz", hash = "sha256:e2c6a17f8d9a129e99c1320fce89b33843e2963871025c4c2bb7b8b8d8732a85", size = 3458, upload-time = "2023-05-31T10:09:11.716Z" } wheels = [ @@ -625,8 +615,8 @@ name = "flake8-bugbear" version = "24.12.12" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "attrs", marker = "python_full_version < '3.10'" }, - { name = "flake8", marker = "python_full_version < '3.10'" }, + { name = "attrs" }, + { name = "flake8" }, ] sdist = { url = "https://files.pythonhosted.org/packages/c7/25/48ba712ff589b0149f21135234f9bb45c14d6689acc6151b5e2ff8ac2ae9/flake8_bugbear-24.12.12.tar.gz", hash = "sha256:46273cef0a6b6ff48ca2d69e472f41420a42a46e24b2a8972e4f0d6733d12a64", size = 82907, upload-time = "2024-12-12T16:49:26.307Z" } wheels = [ @@ -638,7 +628,7 @@ name = "flake8-commas" version = "2.1.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "flake8", marker = "python_full_version < '3.10'" }, + { name = "flake8" }, ] sdist = { url = "https://files.pythonhosted.org/packages/0e/83/814bc8eb02b8883bc004384a1fb8b1f45b4a0b892e579fec7c80a9368526/flake8-commas-2.1.0.tar.gz", hash = "sha256:940441ab8ee544df564ae3b3f49f20462d75d5c7cac2463e0b27436e2050f263", size = 8484, upload-time = "2021-10-13T19:25:41.6Z" } wheels = [ @@ -650,7 +640,7 @@ name = "flake8-comprehensions" version = "3.16.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "flake8", marker = "python_full_version < '3.10'" }, + { name = "flake8" }, ] sdist = { url = "https://files.pythonhosted.org/packages/6d/7d/7ffaa876ca5b330fc244287208dce1d12515b88a69488ea90ab58c94501d/flake8_comprehensions-3.16.0.tar.gz", hash = "sha256:9cbf789905a8f03f9d350fb82b17b264d9a16c7ce3542b2a7b871ef568cafabe", size = 12991, upload-time = "2024-10-27T21:51:18.029Z" } wheels = [ @@ -662,8 +652,8 @@ name = "flake8-debugger" version = "4.1.2" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "flake8", marker = "python_full_version < '3.10'" }, - { name = "pycodestyle", marker = "python_full_version < '3.10'" }, + { name = "flake8" }, + { name = "pycodestyle" }, ] sdist = { url = "https://files.pythonhosted.org/packages/1f/1e/f9bdb98f3df5dceaa2287a8fb5801a22681dbd677a8759704083357e27c4/flake8-debugger-4.1.2.tar.gz", hash = "sha256:52b002560941e36d9bf806fca2523dc7fb8560a295d5f1a6e15ac2ded7a73840", size = 7801, upload-time = "2022-04-30T16:50:55.71Z" } wheels = [ @@ -675,8 +665,8 @@ name = "flake8-docstrings" version = "1.7.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "flake8", marker = "python_full_version < '3.10'" }, - { name = "pydocstyle", marker = "python_full_version < '3.10'" }, + { name = "flake8" }, + { name = "pydocstyle" }, ] sdist = { url = "https://files.pythonhosted.org/packages/93/24/f839e3a06e18f4643ccb81370909a497297909f15106e6af2fecdef46894/flake8_docstrings-1.7.0.tar.gz", hash = "sha256:4c8cc748dc16e6869728699e5d0d685da9a10b0ea718e090b1ba088e67a941af", size = 5995, upload-time = "2023-01-25T14:27:13.903Z" } wheels = [ @@ -688,9 +678,9 @@ name = "flake8-eradicate" version = "1.5.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "attrs", marker = "python_full_version < '3.10'" }, - { name = "eradicate", marker = "python_full_version < '3.10'" }, - { name = "flake8", marker = "python_full_version < '3.10'" }, + { name = "attrs" }, + { name = "eradicate" }, + { name = "flake8" }, ] sdist = { url = "https://files.pythonhosted.org/packages/9e/72/a3975dfa4287396e9fb8fc2b4ee94a80d0809babbf92abed5af9c8e29c95/flake8_eradicate-1.5.0.tar.gz", hash = "sha256:aee636cb9ecb5594a7cd92d67ad73eb69909e5cc7bd81710cf9d00970f3983a6", size = 4508, upload-time = "2023-05-31T09:57:15.484Z" } wheels = [ @@ -702,8 +692,8 @@ name = "flake8-isort" version = "6.1.2" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "flake8", marker = "python_full_version < '3.10'" }, - { name = "isort", marker = "python_full_version < '3.10'" }, + { name = "flake8" }, + { name = "isort" }, ] sdist = { url = "https://files.pythonhosted.org/packages/7c/ea/2f2662d4fefa6ab335c7119cb28e5bc57c935a86a69a7f72df3ea5fe7b2c/flake8_isort-6.1.2.tar.gz", hash = "sha256:9d0452acdf0e1cd6f2d6848e3605e66b54d920e73471fb4744eef0f93df62d5d", size = 17756, upload-time = "2025-01-29T12:29:25.753Z" } wheels = [ @@ -715,8 +705,8 @@ name = "flake8-quotes" version = "3.4.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "flake8", marker = "python_full_version < '3.10'" }, - { name = "setuptools", marker = "python_full_version < '3.10'" }, + { name = "flake8" }, + { name = "setuptools" }, ] sdist = { url = "https://files.pythonhosted.org/packages/dd/57/a173e3eb86072b7ee77650aca496b15d6886367d257f58ea9de5276e330a/flake8-quotes-3.4.0.tar.gz", hash = "sha256:aad8492fb710a2d3eabe68c5f86a1428de650c8484127e14c43d0504ba30276c", size = 14107, upload-time = "2024-02-10T21:58:22.357Z" } @@ -725,9 +715,9 @@ name = "flake8-rst-docstrings" version = "0.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "flake8", marker = "python_full_version < '3.10'" }, - { name = "pygments", marker = "python_full_version < '3.10'" }, - { name = "restructuredtext-lint", marker = "python_full_version < '3.10'" }, + { name = "flake8" }, + { name = "pygments" }, + { name = "restructuredtext-lint" }, ] sdist = { url = "https://files.pythonhosted.org/packages/18/d6/a3e5f86f984d6d8caa1705deffdae84c710e594ab5c1985e26c5e1bb05db/flake8_rst_docstrings-0.3.1.tar.gz", hash = "sha256:26dcc1338caf985990677696a8a6a274f73a0c6845b85f567befd3b648db78e2", size = 12867, upload-time = "2025-04-29T11:34:56.437Z" } wheels = [ @@ -739,7 +729,7 @@ name = "flake8-string-format" version = "0.3.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "flake8", marker = "python_full_version < '3.10'" }, + { name = "flake8" }, ] sdist = { url = "https://files.pythonhosted.org/packages/68/db/500e114a9ee115b03a21a2581c227fd932a0f50c4ae8fee514ef9a373cf4/flake8-string-format-0.3.0.tar.gz", hash = "sha256:65f3da786a1461ef77fca3780b314edb2853c377f2e35069723348c8917deaa2", size = 6495, upload-time = "2020-02-16T15:27:51.045Z" } wheels = [ @@ -886,7 +876,7 @@ resolution-markers = [ "python_full_version < '3.10'", ] dependencies = [ - { name = "mdurl", marker = "python_full_version < '3.10'" }, + { name = "mdurl" }, ] sdist = { url = "https://files.pythonhosted.org/packages/38/71/3b932df36c1a044d397a1f92d1cf91ee0a503d91e470cbd670aa66b07ed0/markdown-it-py-3.0.0.tar.gz", hash = "sha256:e3f60a94fa066dc52ec76661e37c851cb232d92f9886b15cb560aaada2df8feb", size = 74596, upload-time = "2023-06-03T06:41:14.443Z" } wheels = [ @@ -901,7 +891,7 @@ resolution-markers = [ "python_full_version >= '3.10'", ] dependencies = [ - { name = "mdurl", marker = "python_full_version >= '3.10'" }, + { name = "mdurl" }, ] sdist = { url = "https://files.pythonhosted.org/packages/5b/f5/4ec618ed16cc4f8fb3b701563655a69816155e79e24a17b651541804721d/markdown_it_py-4.0.0.tar.gz", hash = "sha256:cb0a2b4aa34f932c007117b194e945bd74e0ec24133ceb5bac59009cda1cb9f3", size = 73070, upload-time = "2025-08-11T12:57:52.854Z" } wheels = [ @@ -1009,7 +999,7 @@ name = "pbr" version = "7.0.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "setuptools", marker = "python_full_version < '3.10'" }, + { name = "setuptools" }, ] sdist = { url = "https://files.pythonhosted.org/packages/80/88/baf6b45d064271f19fefac7def6a030a893f912f430de0024dd595ced61f/pbr-7.0.0.tar.gz", hash = "sha256:cf4127298723dafbce3afd13775ccf3885be5d3c8435751b867f9a6a10b71a39", size = 129146, upload-time = "2025-08-13T09:16:41.654Z" } wheels = [ @@ -1021,7 +1011,7 @@ name = "pep8-naming" version = "0.13.3" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "flake8", marker = "python_full_version < '3.10'" }, + { name = "flake8" }, ] sdist = { url = "https://files.pythonhosted.org/packages/5b/c0/0db8b2867395a9a137e86af8bdf5a566e41d9c6453e509cd3042419ae29e/pep8-naming-0.13.3.tar.gz", hash = "sha256:1705f046dfcd851378aac3be1cd1551c7c1e5ff363bacad707d43007877fa971", size = 16129, upload-time = "2022-12-19T20:45:27.158Z" } wheels = [ @@ -1411,7 +1401,7 @@ name = "pydocstyle" version = "6.3.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "snowballstemmer", marker = "python_full_version < '3.10'" }, + { name = "snowballstemmer" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e9/5c/d5385ca59fd065e3c6a5fe19f9bc9d5ea7f2509fa8c9c22fb6b2031dd953/pydocstyle-6.3.0.tar.gz", hash = "sha256:7ce43f0c0ac87b07494eb9c0b462c0b73e6ff276807f204d6b53edc72b7e44e1", size = 36796, upload-time = "2023-01-17T20:29:19.838Z" } wheels = [ @@ -1623,7 +1613,7 @@ name = "restructuredtext-lint" version = "1.4.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "docutils", marker = "python_full_version < '3.10'" }, + { name = "docutils" }, ] sdist = { url = "https://files.pythonhosted.org/packages/48/9c/6d8035cafa2d2d314f34e6cd9313a299de095b26e96f1c7312878f988eec/restructuredtext_lint-1.4.0.tar.gz", hash = "sha256:1b235c0c922341ab6c530390892eb9e92f90b9b75046063e047cacfb0f050c45", size = 16723, upload-time = "2022-02-24T05:51:10.907Z" } @@ -1721,7 +1711,7 @@ name = "stevedore" version = "5.4.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "pbr", marker = "python_full_version < '3.10'" }, + { name = "pbr" }, ] sdist = { url = "https://files.pythonhosted.org/packages/28/3f/13cacea96900bbd31bb05c6b74135f85d15564fc583802be56976c940470/stevedore-5.4.1.tar.gz", hash = "sha256:3135b5ae50fe12816ef291baff420acb727fcd356106e3e9cbfa9e5985cd6f4b", size = 513858, upload-time = "2025-02-20T14:03:57.285Z" } wheels = [ @@ -1762,7 +1752,7 @@ wheels = [ [[package]] name = "taskiq-postgresql" -version = "0.3.2" +version = "0.4.0" source = { editable = "." } dependencies = [ { name = "taskiq" }, @@ -1958,26 +1948,26 @@ resolution-markers = [ "python_full_version < '3.10'", ] dependencies = [ - { name = "astor", marker = "python_full_version < '3.10'" }, - { name = "attrs", marker = "python_full_version < '3.10'" }, - { name = "darglint", marker = "python_full_version < '3.10'" }, - { name = "flake8", marker = "python_full_version < '3.10'" }, - { name = "flake8-bandit", marker = "python_full_version < '3.10'" }, - { name = "flake8-broken-line", marker = "python_full_version < '3.10'" }, - { name = "flake8-bugbear", marker = "python_full_version < '3.10'" }, - { name = "flake8-commas", marker = "python_full_version < '3.10'" }, - { name = "flake8-comprehensions", marker = "python_full_version < '3.10'" }, - { name = "flake8-debugger", marker = "python_full_version < '3.10'" }, - { name = "flake8-docstrings", marker = "python_full_version < '3.10'" }, - { name = "flake8-eradicate", marker = "python_full_version < '3.10'" }, - { name = "flake8-isort", marker = "python_full_version < '3.10'" }, - { name = "flake8-quotes", marker = "python_full_version < '3.10'" }, - { name = "flake8-rst-docstrings", marker = "python_full_version < '3.10'" }, - { name = "flake8-string-format", marker = "python_full_version < '3.10'" }, - { name = "pep8-naming", marker = "python_full_version < '3.10'" }, - { name = "pygments", marker = "python_full_version < '3.10'" }, - { name = "setuptools", marker = "python_full_version < '3.10'" }, - { name = "typing-extensions", marker = "python_full_version < '3.10'" }, + { name = "astor" }, + { name = "attrs" }, + { name = "darglint" }, + { name = "flake8" }, + { name = "flake8-bandit" }, + { name = "flake8-broken-line" }, + { name = "flake8-bugbear" }, + { name = "flake8-commas" }, + { name = "flake8-comprehensions" }, + { name = "flake8-debugger" }, + { name = "flake8-docstrings" }, + { name = "flake8-eradicate" }, + { name = "flake8-isort" }, + { name = "flake8-quotes" }, + { name = "flake8-rst-docstrings" }, + { name = "flake8-string-format" }, + { name = "pep8-naming" }, + { name = "pygments" }, + { name = "setuptools" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/c2/f4/2a76c59661fae8534b81e992a37d347de241b242aaf5bc651b10d24b7025/wemake_python_styleguide-0.19.2.tar.gz", hash = "sha256:850fe70e6d525fd37ac51778e552a121a489f1bd057184de96ffd74a09aef414", size = 168472, upload-time = "2024-03-26T15:47:38.412Z" } wheels = [ @@ -1992,9 +1982,9 @@ resolution-markers = [ "python_full_version >= '3.10'", ] dependencies = [ - { name = "attrs", marker = "python_full_version >= '3.10'" }, - { name = "flake8", marker = "python_full_version >= '3.10'" }, - { name = "pygments", marker = "python_full_version >= '3.10'" }, + { name = "attrs" }, + { name = "flake8" }, + { name = "pygments" }, ] sdist = { url = "https://files.pythonhosted.org/packages/de/59/489140f56e1d21c1785066f06ec19b539f5bd8f1d572983b9fdc1071979f/wemake_python_styleguide-1.3.0.tar.gz", hash = "sha256:b8fcbeb1271a0a324c30daca2940c4cf769b14215a57ba55412af543cc153c77", size = 156768, upload-time = "2025-07-13T06:22:44.689Z" } wheels = [