PR #7900 - reflex-dev/reflex

All Submissions:

Have you followed the guidelines stated in CONTRIBUTING.md file?
Have you checked to ensure there aren't any other open Pull Requests for the desired changed?

Type of change

Bug fix (non-breaking change which fixes an issue)

Changes To Core Features:

Have you added an explanation of what your changes do and why you'd like us to include them?
Have you written new tests for your core changes, as applicable?
Have you successfully ran tests with your changes locally?

Description

RedisTokenManager's two subscriber tasks ran under ensure_task's default limit, so five errors within a minute ended them, and nothing started them again until a new socket reached that worker. This follows the fix direction in the issue:

ensure_task and _run_forever accept exception_limit=None. The task then retries without a limit, the delay doubles after each error in a row up to exception_delay_max (30 s by default), which also caps the first delay, only the first error of a run is logged, and the coroutine function is passed a recovered callable that logs the end of the run and resets the delay. The defaults are unchanged, and CancelledError and RuntimeError still propagate.
Both subscribers pass exception_limit=None and call recovered when listen() yields the psubscribe confirmation, the signal StateManagerRedis._subscribe_lock_updates already waits for. An outage that raises in a subscriber logs two lines for it, one ERROR and one INFO: _subscribe_lost_and_found_updates: task error, retrying until it recovers: <error> and _subscribe_lost_and_found_updates: task recovered.
On every confirmation the socket record subscriber writes this instance's own records back with SET NX and reads each key back in the same pipeline, so a record that expired while it was not listening exists again. Where Redis holds a different record, such as one another worker stored for that token in the meantime, that record is left alone and the local entry is refreshed through _get_token_owner, as _handle_socket_record_del does when its own SET NX is rejected. A failed write is logged once and the subscription stays up.

I left out the restart from the emit path. I tried it on main with two RedisTokenManagers over one fake Redis: re-ensuring the task on the instance that publishes left the task on the instance that owns the socket stopped, and the delta reached no subscriber. emit_lost_and_found runs on the publisher, so it cannot restart the owner's task. It would restart the publisher's own subscribers; I can add that if you want it.

What this does not cover

A delta published while the owning worker is not subscribed is still lost. Pub/sub has no replay.
The delay grows during an outage, so the next attempt can be up to 30 s away when Redis comes back. In a run with three failed attempts, delivery resumed after about 7 s, against about 3 s on main.
If the socket that lost its token to another worker disconnects before the restore has read the newer record, the disconnect deletes the other worker's record, as on main; that worker's subscriber writes it back when it receives the deletion.
If a socket disconnects while the restore's pipeline is in flight, its record can be written back after the delete and then stays until it expires, so that client's next connect is given a new token. I forced this with mock_redis; the same interleaving leaves the same record on main through _handle_socket_record_del.
A Redis that keeps refusing CONFIG SET is logged once and then retried with no further lines.
StateManagerRedis._ensure_lock_task keeps the default limit. It is re-ensured on each lock wait.

Tests

tests/units/utils/test_token_manager.py, on mock_redis with a pubsub that can be made to fail: test_lost_and_found_subscriber_survives_redis_outage (two managers: a delta is delivered, the pubsub fails seven times, the two log lines are compared in full, the next delta is delivered), test_socket_record_subscriber_survives_redis_outage (the record deleted during the outage is stored again, one claimed by another instance is left alone and adopted, and the disconnect of the socket that lost it does not delete it), test_socket_record_subscriber_retries_refused_config and test_socket_record_subscriber_restores_records_on_resubscribe. On main the first three fail on the log comparison, after the task has ended with task exceeded exception limit 5 within 60.0s, and the fourth because the record is not restored.
test_restore_socket_records_adopts_newer_records runs the restore directly: a missing record is written back, an unchanged one is not read again, a newer one is adopted, one deleted between the pipeline and the read leaves the local record as it was, and a failed pipeline is logged once and reads nothing. On main it fails because the method does not exist.
tests/units/utils/test_tasks.py: test_ensure_task_no_limit compares the delays (1, 2, 4, 8, 16, 30, 30 s, then 1 s after a recovery, and the cap throughout when the first delay is above it) and the log records in full; test_ensure_task_no_limit_cancel_during_retry_delay cancels during a 30 s delay. Both use the new argument, so on main they fail with a TypeError and a timeout, not with the issue's message.
test_ensure_task_default_limit and test_close_during_subscriber_retry_delay pin behaviour that does not change and pass on main too.
mock_redis gains publish, with a test in tests/units/test_mock_redis.py.

Run locally on macOS arm64 with Python 3.13.14 and 3.11.15, same result on both:

$ pytest tests/units/utils/test_tasks.py tests/units/utils/test_token_manager.py tests/units/test_mock_redis.py -q
56 passed, 5 skipped

The 5 skipped tests need a Redis URL. The full unit suite also passes on both (pytest tests/units --cov --no-cov-on-fail --cov-report=: 15314 tests pass and 33 are skipped on 3.13, 15306 pass and 34 are skipped on 3.11), and the twelve new test cases passed 30 runs in a row. ruff check . and ruff format --check . pass, pyright --pythonversion 3.14 reflex tests reports 0 errors, and the pre-commit hooks pass with biome-format skipped. Changelog fragment: news/+redis-subscriber-outage.bugfix.md.

Not run: a real Redis server (none is installed here), two worker processes and the integration tests. I ran redis-py 8.1.0's client against a small RESP stub, which is not Redis: after a dropped connection that can be re-established it resubscribes by itself and listen() yields a new psubscribe confirmation without raising, so the restore runs then too.

This change was prepared with an AI coding assistant (Claude Code) working at my direction on my machine.