fix(broker): recover asyncpg listener after disconnect

[GefMar]

Description

Recreate terminated asyncpg 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.

Type of Change

• 🐛 Bug fix (non-breaking change which fixes an issue)
• ✨ New feature (non-breaking change which adds functionality)
• 💥 Breaking change (fix or feature that would cause existing functionality to not work as expected)
• 📚 Documentation update
• 🔧 Maintenance (dependency updates, CI improvements, etc.)

Changes Made

• Detect unexpected asyncpg listener termination and wake pending consumers with an explicit terminal error.
• Replace failed listener generations using serialized, shutdown-aware reconnect attempts with capped exponential backoff and jitter.
• Reconcile channel-scoped durable messages after LISTEN registration using bounded keyset pagination and atomic DELETE ... RETURNING claims.
• Retry transient query-connection failures without advancing past unclaimed messages.
• Harden startup, reconnect, cancellation, and shutdown cleanup paths.
• Document the asyncpg recovery guarantee and compatibility boundaries in the README.

Testing

• I have added tests that prove the fix is effective.
• New and affected existing unit and integration tests pass locally.
• I have tested recovery with multiple PostgreSQL drivers. Not applicable: transparent recovery is intentionally scoped to asyncpg; psycopg and psqlpy behavior remains unchanged.

Coverage includes:

• asyncpg termination waking a pending listener;
• exact listener backend termination with pg_terminate_backend();
• publication during a forced reconnect gap;
• baseline-versus-fix publish-before-worker behavior;
• fresh listener generation and single-flight recovery;
• transient and terminal reconnect failures;
• shutdown during backoff and listener startup;
• missed-notification reconciliation;
• channel isolation and legacy-row compatibility;
• duplicate notification and reconciliation claims;
• due delayed-message filtering;
• bounded reconciliation pagination.

The baseline commit left the publish-before-worker row queued and undelivered. The fixed working tree delivered the same scenario and removed the claimed row.

Test Configuration

• Python version(s): 3.9.6, 3.13.14, and Python 3.14 Docker profile
• PostgreSQL version: PostgreSQL 17, including 17.10 integration testing
• Driver(s) tested: asyncpg 0.30.0; psycopg 3.2.9 and psqlpy 0.11.6 characterized separately
• Operating System: macOS host with Linux Docker containers

Performance Impact

• No material steady-state performance impact
• Positive performance impact
• Negative performance impact

Performance details:

Normal notification processing retains the existing atomic claim path. Additional database work is limited to a bounded, paginated reconciliation scan during listener startup and after reconnect. Reconnect attempts use capped backoff with jitter to avoid tight retry loops.

Documentation

• I have updated the documentation accordingly.
• I have updated the README.
• I have added docstrings to new functions and lifecycle helpers.

Checklist

• My code follows the style guidelines of this project.
• I have performed a self-review of my own code.
• I have documented the non-obvious lifecycle and recovery paths.
• I have made corresponding changes to the documentation.
• My changes generate no new warnings.
• I have added tests that prove the fix is effective.
• New and affected existing tests pass locally.
• No dependent downstream changes are required.

Related Issues

Refs taskiq-python/taskiq#633.

Screenshots (if applicable)

Not applicable.

Additional Notes

• Transparent listener recovery is intentionally limited to asyncpg.
• Initial authentication, configuration, and invalid-DSN failures remain fail-fast.
• New messages include an internal channel marker used only for reconciliation; user-provided labels and wire payloads are not mutated.
• Legacy rows without the marker remain available through their normal explicit NOTIFY path but are not automatically reconciled because their original channel cannot be determined safely.
• Psycopg and psqlpy listener-recovery parity is tracked separately and is outside this bug-fix scope.