[lesnik512]

Problem

When the Postgres primary disappears without closing its connections (host power loss, network partition), the subscriber stops consuming for good. The fetch query and the terminal writes have no client-side timeout, so they wait on the dead socket until the kernel gives up on TCP retransmits (about 15 minutes on Linux by default). The LISTEN health probe can't help, because it runs between fetches and the fetch itself is the call that's stuck.

A primary that dies cleanly (process killed) was already fine: the next statement fails, and the existing reconnect-with-backoff takes over.

Fix

• Every fetch, terminal delete, batch delete and retry update runs under asyncio.timeout(query_timeout_seconds). On timeout the connection is invalidated, so it gets closed without another round trip on the dead socket, and the error reaches the reconnect loop.
• New subscriber option query_timeout_seconds, available on @broker.subscriber, OutboxRoute and the FastAPI router. It defaults to None, which means lease_ttl_seconds. Values must be > 0, finite, and at most lease_ttl_seconds. A fetch or write still running past the lease has already lost the row, so a larger value buys nothing.
• Why a separate option rather than the lease alone: a lease-only bound ties failure detection to handler duration. The docs recommend long leases for slow-handler queues, and a 10-minute lease would mean roughly 20 minutes before such a subscriber notices a dead primary. The cost of the option is misconfiguration risk: a value that's too low makes a statement waiting on a lock (for example during a migration) time out, drop its connection and reconnect, and the server-side query isn't cancelled.
• Recovery after a silent loss takes about 2 × query_timeout_seconds. The fetch loop hits the cap first. The worker then hits it again on its own stale connection once it gets a row. Bringing that down to one timeout would mean recycling the worker's connection whenever the fetch loop reconnects. I left that out of this PR.

Failover drills

just test-dr brings up a primary and a streaming replica (docker-compose.dr.yml, replica built with pg_basebackup -R). The app connects through a multi-host URL with target_session_attrs=primary. Each scenario runs on a fresh cluster:

• kill: docker kill the primary, then pg_promote() the replica.
• blackhole: disconnect the primary from the Docker network, which is a silent drop with no RST, then promote the replica.

The test checks three things: every message published after the failover is handled within 90s, messages handled before it stay handled, and a NOTIFY wakeup works afterwards. The drills sit in tests_dr/, outside testpaths, and are excluded from coverage, so CI doesn't run them.

Local results (OrbStack):

Scenario	Before	After
kill	subscriber recovered in 11.6s	4.9s
blackhole, default engine	no post-failover message handled in 90s	33.1s with query_timeout_seconds=10

One limitation: after a network disconnect, Docker's DNS stops resolving the old primary's name, so new connection attempts to it fail quickly. A real partition would make them wait for asyncpg's connect timeout (60s by default) before trying the next host.

Tests

• The two new integration tests below simulate a statement hanging on a dropped socket. Both failed before the fix:
  • test_hung_fetch_times_out_and_fetch_loop_reconnects;
  • test_hung_terminal_write_times_out_and_worker_reconnects.
• test_query_timeout_bounds_hung_fetch_below_lease_ttl uses a 600s lease with a 0.5s query timeout, which shows the option is what actually bounds the hang.
• Unit tests cover validation, the default, and passthrough from OutboxRoute and the FastAPI router.
• Full suite: 823 passed at 100% coverage.

Docs

• A new "Database failover" section in docs/usage/subscriber.md covers the engine URL, recovery time per failure mode, and what asynchronous replication can lose.
• The option table, the production checklist and AGENTS.md (for the new recipe) are updated.

[github-actions]

Benchmark gate

✅ gate passed

scenario	msg/s	delete/msg	WALrec/msg	WALB/msg	fpi	upd	del	dead_tup
consumer/w1/b10	1464	1.000	6.73	923	242	5000	5000	10000
consumer/w1/b100	1650	1.000	7.11	1154	243	5000	5000	10000
consumer/w2/b10	1767	1.000	6.73	924	242	5000	5000	10000
consumer/w2/b100	1956	1.000	7.11	1154	243	5000	5000	10000
consumer/w4/b10	1742	1.000	6.76	953	242	5000	5000	10000
consumer/w4/b100	1997	1.000	7.11	1153	243	5000	5000	10000
consumer/w1/b100/tfbs100	6479	0.010	6.11	1130	243	5000	5000	10000
consumer/w1/b100/interval1-10	1583	1.000	7.17	1196	265	5000	5000	10000
producer/w1/b100	3766	0.000	3.04	597	1	0	0	0
producer/w1/b100/ppt1	1194	0.000	4.04	635	0	0	0	0

Gated (fails the build): delete_calls + tuple counters (upd/del/ins) + the producer's insert_calls, exact; select_calls and other_calls within +2; wal_records within a 10% band; fetch_calls within 2x. msg/s, WAL bytes and total calls are informational (timing/FPI noise).