Merging this PR will not alter performance

✅ 8 untouched benchmarks

_{Comparing masenf/statemanager-redis-race (797c8a1) with main (0a5e76e)}

Greptile Summary

This PR fixes race conditions in StateManagerRedis lock detection by introducing a synchronization barrier (_lock_updates_subscribed asyncio.Event) that ensures the Redis pubsub subscription is fully established before any lock-waiting logic can proceed, preventing missed lock-release notifications. The test infrastructure is also hardened with a complementary event_log_on_update event that gives tests a deterministic way to wait for Redis side-effects rather than relying on timing.

Key changes:

• redis.py: Adds _lock_updates_subscribed event; _subscribe_lock_updates sets it after psubscribe completes and clears it in a finally block on exit; _wait_for_lock awaits the event before registering a lock waiter.
• mock_redis.py: Extracts _event_log_append_notify helper; adds event_log_on_update to _internals; adds pubsub_ready barrier to real_redis() so the fixture yields only after the monitoring subscription is active.
• test_redis.py: Adopts event_log_on_update.clear()/wait() pattern for deterministic assertions; promotes SubState1/SubState2 to module level to avoid duplicate state class registrations across test runs.

Confidence Score: 3/5

• Safe to merge once the unbounded wait risk in _wait_for_lock is addressed or consciously accepted.
• The core race-condition fix is correct and well-motivated. The one notable concern is that await self._lock_updates_subscribed.wait() has no timeout: if _subscribe_lock_updates exhausts its retry budget (5 failures by default), the event is never set and modify_state hangs indefinitely with no timeout or error propagation. The rest of the changes — mock infrastructure improvements and test determinism — are clean and well-structured.
• Pay close attention to reflex/istate/manager/redis.py line 983 — the unbounded wait() with no timeout guard.

Important Files Changed

Sequence Diagram

sequenceDiagram
    participant C as Caller (modify_state)
    participant SMR as StateManagerRedis
    participant ET as _lock_task (ensure_task)
    participant PS as Redis pubsub

    C->>SMR: _wait_for_lock(lock_key, lock_id)
    SMR->>SMR: _try_get_lock() → fails (lock held)
    SMR->>ET: _ensure_lock_task()
    Note over ET: Creates task if not running
    ET->>PS: _enable_keyspace_notifications()
    ET->>PS: pubsub.psubscribe(**handlers)
    PS-->>ET: subscription confirmed
    ET->>SMR: _lock_updates_subscribed.set()
    SMR->>SMR: await _lock_updates_subscribed.wait() ✓
    SMR->>SMR: _lock_waiter(lock_key) registered
    SMR->>SMR: _request_lock_release(lock_key, lock_id)
    PS-->>ET: keyspace event: lock del
    ET->>SMR: _handle_lock_release(event)
    SMR->>SMR: lock_released_event.set()
    SMR->>SMR: _try_get_lock() → success
    SMR-->>C: lock acquired

_{Last reviewed commit: "Fix race conditions ..."}