fix(broker): recover asyncpg listener after disconnect

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 |

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."""

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."""