Repository navigation
fix(state-manager): stop redis lock handoffs from stalling events until the lock expires - #7534
benedikt-bartscher wants to merge 22 commits into
Conversation
…ffs, keep a lease's lock while its state is in use and wake waiters a release missed
[Medium impact] The PR appears safe to merge; no new actionable issues were found. SummaryFixes Redis lock handoffs so cancelled events and missed notifications do not stall later events.
Reviews (11) · Last reviewed commit: "Condense the redis lock handoff fragment" · Reviewed by Greptile |
Merging this PR will degrade performance by 4.36%1 warning
Warning Please fix the performance issues or acknowledge them on CodSpeed. Performance Changes
Tip Investigate this regression by commenting Comparing |
…thin each use's lock expiration, and report a failed renewal
…tion interrupts it
…ed state changes hands
…of asyncio.wait_for, which can swallow its cancellation on Python 3.11
There was a problem hiding this comment.
All reported issues were addressed across 4 files
Reply with feedback, questions, or to request a fix.
View guided diff | Re-trigger cubic
…ts lock until the write lands
…stead of a try in the loop
| # When a contention notice broke the lease before its hold time. | ||
| broken_at: float | None = None | ||
|
|
||
| async def keep_lock_alive() -> None: |
There was a problem hiding this comment.
Please move the lease out of _create_lease_break_task into its own unit: a small class (or module-level coroutines) that holds token, lock_id, lease_ctx, started and broken_at and exposes keep_lock_alive, flush, run and release_unstarted. _create_lease_break_task then only waits for the previous lease, checks contention, fetches the state, extends the lock and starts the task.
Evidence: ruff check --select C901 --config 'lint.mccabe.max-complexity=8' reports this function at 32 on this branch, up from 21 on main (the five nested closures count toward it). The closures share four nonlocals (lease_started, broken_at, task, lease_ctx), and lease_ctx is only bound at the end of the enclosing function, so a reader has to hold all ~240 lines in mind to follow one handoff.
Generated by Claude Code
| """ | ||
| # A lease cancelled before its first step never ran lease_breaker, | ||
| # so nothing else flushes its state and releases its lock. | ||
| if lease_started: |
There was a problem hiding this comment.
Please retrieve the lease's exception here, so a failed flush is logged once:
if lease_started:
if not lease.cancelled():
lease.exception()
returnEvidence: with a use of the cached state longer than lock_expiration (one background async with self holding 2600 ms against a 2000 ms lock), the backend logs the LockExpiredError twice: once from flush() via call_exception_handler ("Exception in Redis State Manager lease breaker"), and again as "Task exception was never retrieved" for reflex_lease_breaker|... when the task is collected. The flush pops the lease from _local_leases before it raises, so no later modify_state awaits the task. The second log predates this PR, but this callback is the one place that sees every finished lease, and await lease_task in the tests still raises after exception().
Generated by Claude Code
| ttl_ms = int((deadline - now) * 1000) | ||
| if ttl_ms <= 0 or not await self._extend_lock( | ||
| redis_lock_key, lock_id, ttl_ms | ||
| ): |
There was a problem hiding this comment.
A lease can now hold its lock for longer than lock_expiration (each queued use gets a full expiration from its handoff). A waiter on another instance registers in {lock_key}_waiters once, with PEXPIRE lock_expiration (_request_lock_release), and never refreshes it. So while this renewal runs past one expiration, that waiter drops out of the set. After the flush, the next local modify_state sees _n_lock_contenders() == 0 and leases again, and the remote waiter, which only re-checks every lock_expiration / 10, stays behind. Before this PR the lock expired at lock_expiration, so the remote waiter got in (and the local flush lost its writes).
Please refresh the waiter's registration on each re-check in _wait_lock (an SADD + PEXPIRE on the timeout branch), or say in the comment here why the longer lease is acceptable for remote contenders. Evidence is code reading only; I did not reproduce this with two instances.
Generated by Claude Code
| @@ -0,0 +1 @@ | |||
| Fixed the redis state manager keeping a token's lock until it expired after an event was cancelled while taking it (e.g. by `supersedes`), a late release deleting the next holder's lock, a lease's updates being lost when the events queued ahead of its flush outlasted the lock expiration together, and waiters sitting out the whole lock expiration when a release did not wake them. | |||
There was a problem hiding this comment.
Please split this into short sentences, one per fixed case, for example:
Fixed the redis state manager stalling a token's events until its lock expired after an event was cancelled while taking the lock (for example by
supersedes). A late release no longer deletes the next holder's lock. A lease now keeps its lock while the events queued ahead of its flush run, so their updates are no longer lost. Waiters are woken when a lease ends or a lock expires instead of waiting out the whole lock expiration.
Evidence: CLAUDE.md asks for "a sentence or two"; this is one 70-word sentence with four clauses, which the changelog reader has to parse in one go.
Generated by Claude Code
…extract the lease A remote waiter now re-registers as a lock contender on each re-check, so a lease that outlasts lock_expiration no longer leases again ahead of it; the mock redis no longer reports an sadd for an existing member. A failed lease is reported once: the done callback only reports what flush() did not already report. The lease closures move into a module-level _Lease dataclass, with no change in behavior. The news fragment is split into short sentences.
There was a problem hiding this comment.
1 issue found across 4 files (changes from recent commits).
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="tests/units/istate/manager/test_redis.py">
<violation number="1" location="tests/units/istate/manager/test_redis.py:1571">
P2: `remote.close()` closes the connection pool shared with `state_manager_redis` while the original manager is still running and is used again below. Give the remote manager an independently owned Redis client, or stop only its subscription task without closing the shared client; also ensure that cleanup runs in a `finally` block.</violation>
</file>
Reply with feedback, questions, or to request a fix.
View guided diff | Re-trigger cubic
|
|
||
| may_take_lock.set() | ||
| await asyncio.wait_for(waiting, timeout=short_lock_expiration / 1000) | ||
| await remote.close() |
There was a problem hiding this comment.
P2: remote.close() closes the connection pool shared with state_manager_redis while the original manager is still running and is used again below. Give the remote manager an independently owned Redis client, or stop only its subscription task without closing the shared client; also ensure that cleanup runs in a finally block.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At tests/units/istate/manager/test_redis.py, line 1571:
<comment>`remote.close()` closes the connection pool shared with `state_manager_redis` while the original manager is still running and is used again below. Give the remote manager an independently owned Redis client, or stop only its subscription task without closing the shared client; also ensure that cleanup runs in a `finally` block.</comment>
<file context>
@@ -1439,6 +1495,85 @@ async def held_set_state(*args: Any, **kwargs: Any) -> None:
+
+ may_take_lock.set()
+ await asyncio.wait_for(waiting, timeout=short_lock_expiration / 1000)
+ await remote.close()
+ final_state = await state_manager_redis.get_state(token)
+ assert isinstance(final_state, root_state)
</file context>
Closing a second manager on the shared client disconnected the pool the main manager still used.
Under CPU load, our app's events stalled for 5 or 10 seconds (the oplock hold time and the lock expiration), and some updates made under a lease were lost.
StateManagerRedishad several ways to lose a lock or a wakeup:supersedesand a newon_load_internalcancel events at any await. This covers aSET NXwhose reply never arrived, the awaits in_try_modify_statebefore a lease or the single update owns the lock, and a lease task cancelled before its first step, which never enters the context that releases the lock.GETDEL, so a holder releasing after its lock expired deleted the next holder's lock.LockExpiredErrorand dropped every change made under the lease.This PR:
AsyncExitStack.pop_all()once the state is fetchedLockExpiredErrorEach case has a test in
tests/units/istate/manager/test_redis.pythat fails without the fix.test_oplock_immediate_cancelnow checks the outcome, because its lease can be cancelled before it starts.With a script that cancels a waiter around the handoff (2 s lock expiration, 50 ms redis latency), 19 of 30 waiters waited out the expiration before and none of 60 after. A fuzzer with up to 2 s holders lost up to 23 of 26 increments per round before and none after.
The unit tests pass (12512 without Redis, 12525 against Redis with and without oplock).