fix: refresh an expiring OAuth token once across workers and replicas (#30450)

#30426 stopped concurrent requests from refreshing the same OAuth session twice, but its lock only lives inside one process. With several uvicorn workers or replicas, two requests on different workers still send the same refresh token, a rotating provider rejects the second with invalid_grant, and the session gets deleted, so the user's OAuth session is logged out again.

When Redis is configured, which multi-worker and multi-replica deployments require, the refresh now takes a Redis lock per session instead of the in-process one. Single-process deployments without Redis keep the in-process lock. The waiter re-reads the session inside the lock as before and uses the token that was just stored.

It uses redis-py's own async lock because the existing RedisLock is synchronous and never waits. The Sentinel proxy now passes `lock` through unwrapped like `pipeline` and `pubsub`; otherwise it returned a coroutine and every refresh behind Sentinel would fail.

Tested with separate OS processes on one sqlite DB, a real Redis and a rotating mock provider: 2 and 5 processes (and 5 processes x 3 requests) now cause 1 refresh, every caller gets the new token and the session is kept (before: one refresh per process, session deleted every run). Single refresh, failed refresh, valid token and the single-process path without Redis are unchanged.

Follow-up to #30426, refs #30416

backend/open_webui/utils/oauth.py

@@ -72,6 +72,7 @@ from open_webui.env import (
     ENABLE_OAUTH_ID_TOKEN_COOKIE,
     OAUTH_CLIENT_INFO_ENCRYPTION_KEY,
     OAUTH_MAX_SESSIONS_PER_USER,
+    REDIS_KEY_PREFIX,
     WEBUI_AUTH_COOKIE_SAME_SITE,
     WEBUI_AUTH_COOKIE_SECURE,
 )
@@ -1429,7 +1430,14 @@ class OAuthManager:
         Returns:
             dict: Refreshed token data, or None if refresh failed
         """
-        async with self._refresh_locks.setdefault(session.id, asyncio.Lock()):
+        redis = self.app.state.redis
+        if redis:
+            # Shared across workers and replicas
+            refresh_lock = redis.lock(f'{REDIS_KEY_PREFIX}:oauth:refresh_lock:{session.id}', timeout=60)
+        else:
+            refresh_lock = self._refresh_locks.setdefault(session.id, asyncio.Lock())
+
+        async with refresh_lock:
             # Another request may have refreshed while we waited; its refresh token is now spent
             current_session = await OAuthSessions.get_session_by_id(session.id)
             if current_session and current_session.token != session.token:

backend/open_webui/utils/redis.py

@@ -35,7 +35,7 @@ _SENTINEL_RETRYABLE = (
     _redis_sync.exceptions.ReadOnlyError,
     _redis_sync.exceptions.TimeoutError,
 )
-_FACTORY_METHODS = frozenset({'pipeline', 'pubsub', 'monitor', 'client', 'transaction'})
+_FACTORY_METHODS = frozenset({'pipeline', 'pubsub', 'monitor', 'client', 'transaction', 'lock'})
 _CONNECTION_POOL: dict[tuple, Any] = {}