Make oplock_hold_time_ms configurable (#5975)

Make the oplock hold time configurable and drop the default down to 50% of the
configured lock timeout to avoid lock expiration between the lease breaker
sleep ending and the coroutine actually getting scheduled for execution.

Author: masenf

reflex/environment.py

@@ -733,6 +733,9 @@ class EnvironmentVariables:
     # Whether to opportunistically hold the redis lock to allow fast in-memory access while uncontended.
     REFLEX_OPLOCK_ENABLED: EnvVar[bool] = env_var(False)
 
+    # How long to opportunistically hold the redis lock in milliseconds (must be less than the token expiration).
+    REFLEX_OPLOCK_HOLD_TIME_MS: EnvVar[int] = env_var(0)
+
 
 environment = EnvironmentVariables()
 

reflex/istate/manager/redis.py

@@ -50,6 +50,17 @@ def _default_lock_warning_threshold() -> int:
     return get_config().redis_lock_warning_threshold
 
 
+def _default_oplock_hold_time_ms() -> int:
+    """Get the default opportunistic lock hold time.
+
+    Returns:
+        The default opportunistic lock hold time.
+    """
+    return environment.REFLEX_OPLOCK_HOLD_TIME_MS.get() or (
+        _default_lock_expiration() // 2
+    )
+
+
 SMR = f"[SMR:{os.getpid()}]"
 start = time.monotonic()
 
@@ -85,6 +96,11 @@ class StateManagerRedis(StateManager):
         default_factory=_default_lock_warning_threshold
     )
 
+    # How long to opportunistically hold the redis lock in milliseconds (must be less than the token expiration).
+    oplock_hold_time_ms: int = dataclasses.field(
+        default_factory=_default_oplock_hold_time_ms
+    )
+
     # The keyspace subscription string when redis is waiting for lock to be released.
     _redis_notify_keyspace_events: str = dataclasses.field(
         default="K"  # Enable keyspace notifications (target a particular key)
@@ -154,6 +170,9 @@ def __post_init__(self):
         if self.lock_warning_threshold >= (lock_expiration := self.lock_expiration):
             msg = f"The lock warning threshold({self.lock_warning_threshold}) must be less than the lock expiration time({lock_expiration})."
             raise InvalidLockWarningThresholdError(msg)
+        if self._oplock_enabled and self.oplock_hold_time_ms >= lock_expiration:
+            msg = f"The opportunistic lock hold time({self.oplock_hold_time_ms}) must be less than the lock expiration time({lock_expiration})."
+            raise InvalidLockWarningThresholdError(msg)
         with contextlib.suppress(RuntimeError):
             asyncio.get_running_loop()  # Check if we're in an event loop.
             self._ensure_lock_task()
@@ -620,7 +639,7 @@ async def do_flush() -> None:
         async def lease_breaker():
             cancelled_error: asyncio.CancelledError | None = None
             async with cleanup_ctx:
-                lease_break_time = (self.lock_expiration * 0.8) / 1000
+                lease_break_time = self.oplock_hold_time_ms / 1000
                 if self._debug_enabled:
                     console.debug(
                         f"{SMR} [{time.monotonic() - start:.3f}] {client_token} lease breaker {lock_id.decode()} started, sleeping for {lease_break_time}s"