[lesnik512]

Problem

If the Redis primary stops answering without closing its connections, the subscriber stops delivering for good. Common causes are a frozen VM, a network partition, or a host that loses power before Sentinel fails over. The subscriber's fetch, claim and ack commands have no time limit of their own, so they wait on the dead socket until the kernel gives up on TCP.

That limit comes from redis-py's socket_timeout, which the caller controls. redis-py 8.0 defaults it to 5s. Versions 5.0.1 through 7.4.0 default it to None, and this package allows redis>=5.0.1. A hung fetch stops the poll loop. A hung ack or claim keeps its max_concurrent slot forever, so once enough of them hang, the loop stops fetching too.

The Sentinel drill from #117 reproduces this. With socket_timeout=None and the primary paused, Sentinel promotes the replica and publishes recover, but none of the 100 timers due after the failure is delivered within 90s.

Fix

• New subscriber option command_timeout, available on @broker.subscriber and TimersRoute. It defaults to None, which means lease_ttl. Values must be finite, greater than 0, and at most lease_ttl. Past the lease, another worker may claim the timer anyway, so a larger value gains nothing.
• The fetch (due), the claim and the ack/reject removal each run under anyio.fail_after(command_timeout). When a command is cancelled mid-read, redis-py disconnects that connection (checked back to 5.0.1), so the next command opens a fresh one, and with a Sentinel pool that finds the new primary. The existing error handling takes it from there: the fetch loop logs and backs off, and a timed-out ack or claim is logged and frees its slot. A timer whose ack timed out is redelivered after its lease, which the at-least-once contract already allows.
• I made it a separate option, as outbox did in faststream-outbox#262, because bounding by the lease alone ties failure detection to handler duration. Someone running lease_ttl=600 for slow handlers would otherwise wait 10 minutes per hung command. The risk on the other side: if the value is set too low, slow but healthy commands time out and reconnect.
• publish() and the inspection methods still use the caller's socket timeout. They run in the caller's own code, where the caller can wrap them in a timeout.

Tests

• tests/test_command_timeout.py runs the client through a small TCP proxy that freezes every open connection but lets new ones through, which is what a silent primary followed by failover looks like. Both tests failed before the fix:
  • test_a_fetch_hung_on_a_silent_connection_is_abandoned
  • test_an_ack_hung_on_a_silent_connection_frees_its_handler_slot (max_concurrent=1)
• Unit tests cover the default, the override, validation, and passing the option through TimersRoute. The defaults invariant in test_tuning_defaults.py now includes command_timeout.
• Full suite: 179 passed, 100% coverage. The two hang tests passed 8 runs in a row.
• just test-dr from test: Sentinel failover drills (just test-dr) #117 with socket_timeout=None: the pause drill went from no delivery after the failure to recovery in 10.8s, and kill still passes (11.0s).