Handle Redis Connection Failure (#856)

* Better handle redis failing and working without redis

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

* Remove REDIS_REQUIRED and instead default to crashing on error connecting to redis

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

* Migrate from os._exit(1) to sys.exit(1)

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

* formatting

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

* Extra testing

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

* Implement proper degradation for redis crashing

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

* Refactor Redis error handling in socketio.py

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

- Removed custom exception classes for Redis connection, configuration, and operation errors to simplify error handling.
- Updated error handling in `create_socket_manager()` to log failures without specific Redis error classification.
- Cleaned up imports in test files to reflect the removal of the custom exceptions.
- Enhanced comments for clarity on exception handling strategy.

* Rename Redis connection validation functions

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

- Renamed functions for clarity: `_validate_redis_before_manager_creation` to `can_we_reach_redis` and `validate_redis_connection` to `should_we_use_redis`.

* removed single use function

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

* Corrected tests to use safe_emit

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

---------

Signed-off-by: Gavin Jaeger-Freeborn <[email protected]>

Author: Gavinok

docker/docker-compose.yaml

@@ -78,6 +78,11 @@ services:
     depends_on:
       redis:
         condition: service_healthy
+    healthcheck:
+      test: ["CMD", "curl", "-f", "localhost:5000/health"]
+      interval: 10s
+      timeout: 3s
+      retries: 3
 
   redis:
     image: redis:8-alpine

oidc-controller/api/core/config.py

@@ -259,6 +259,14 @@ class GlobalConfig(BaseSettings):
     REDIS_DB: int = int(os.environ.get("REDIS_DB", 0))
     USE_REDIS_ADAPTER: bool = strtobool(os.environ.get("USE_REDIS_ADAPTER", False))
 
+    # Redis error handling and retry configuration
+    REDIS_THREAD_MAX_RETRIES: int = int(os.environ.get("REDIS_THREAD_MAX_RETRIES", 5))
+    REDIS_PUBSUB_MAX_FAILURES: int = int(
+        os.environ.get("REDIS_PUBSUB_MAX_FAILURES", 10)
+    )
+    REDIS_RETRY_BASE_DELAY: int = int(os.environ.get("REDIS_RETRY_BASE_DELAY", 1))
+    REDIS_RETRY_MAX_DELAY: int = int(os.environ.get("REDIS_RETRY_MAX_DELAY", 60))
+
     model_config = ConfigDict(case_sensitive=True)
 
 

oidc-controller/api/main.py

@@ -7,6 +7,7 @@
 from pathlib import Path
 
 import uvicorn
+import redis.asyncio as async_redis
 from api.core.config import settings
 from fastapi import FastAPI
 from starlette.requests import Request
@@ -26,7 +27,7 @@
 )
 from .verificationConfigs.router import router as ver_configs_router
 from .clientConfigurations.router import router as client_config_router
-from .routers.socketio import sio_app
+from .routers.socketio import sio_app, _build_redis_url, _handle_redis_failure
 from api.core.oidc.provider import init_provider
 
 logger: structlog.typing.FilteringBoundLogger = structlog.getLogger(__name__)
@@ -105,34 +106,32 @@ async def logging_middleware(request: Request, call_next) -> Response:
     try:
         response: Response = await call_next(request)
         return response
-    finally:
+    except Exception as e:
         process_time = time.time() - start_time
-        # If we have a response object, log the details
-        if "response" in locals():
-            logger.info(
-                "processed a request",
-                status_code=response.status_code,
-                process_time=process_time,
-            )
-        # Otherwise, extract the exception from traceback, log and return a 500 response
-        else:
-            logger.info(
-                "failed to process a request",
-                status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
-                process_time=process_time,
-            )
-
-            # Need to explicitly log the traceback
-            logger.error(traceback.format_exc())
-
-            return JSONResponse(
-                status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
-                content={
-                    "status": "error",
-                    "message": "Internal Server Error",
-                    "process_time": process_time,
-                },
-            )
+        logger.info(
+            "failed to process a request",
+            status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
+            process_time=process_time,
+        )
+
+        # Need to explicitly log the traceback
+        logger.error(traceback.format_exc())
+
+        return JSONResponse(
+            status_code=http_status.HTTP_500_INTERNAL_SERVER_ERROR,
+            content={
+                "status": "error",
+                "message": "Internal Server Error",
+                "process_time": process_time,
+            },
+        )
+    else:
+        process_time = time.time() - start_time
+        logger.info(
+            "processed a request",
+            status_code=response.status_code,
+            process_time=process_time,
+        )
 
 
 @app.on_event("startup")
@@ -141,6 +140,23 @@ async def on_tenant_startup():
     await init_db()
     await init_provider(await get_db())
 
+    # Check Redis availability if adapter is enabled
+    if settings.USE_REDIS_ADAPTER:
+        try:
+            # Test Redis connectivity during startup
+            redis_url = _build_redis_url()
+            redis_client = async_redis.from_url(redis_url)
+            await redis_client.ping()
+            await redis_client.close()
+            logger.info("Redis adapter is available and ready")
+        except Exception as e:
+            error_type = _handle_redis_failure("startup Redis check", e)
+            logger.warning(
+                f"Redis adapter enabled but unavailable (type: {error_type}) - continuing with degraded Socket.IO functionality"
+            )
+    else:
+        logger.debug("Redis adapter disabled")
+
     logger.info(">>> Starting up app new ...")
 
 

oidc-controller/api/routers/acapy_handler.py

@@ -13,7 +13,7 @@
 from ..verificationConfigs.crud import VerificationConfigCRUD
 
 from ..core.config import settings
-from ..routers.socketio import sio, get_socket_id_for_pid
+from ..routers.socketio import sio, get_socket_id_for_pid, safe_emit
 
 logger: structlog.typing.FilteringBoundLogger = structlog.getLogger(__name__)
 
@@ -106,7 +106,7 @@ async def _emit_status_to_socket(
     pid = str(auth_session.id)
     sid = await get_socket_id_for_pid(pid, db)
     if sid:
-        await sio.emit("status", {"status": status}, to=sid)
+        await safe_emit("status", {"status": status}, to=sid)
 
 
 async def _parse_webhook_body(request: Request) -> dict[Any, Any]:

oidc-controller/api/routers/oidc.py

@@ -31,7 +31,7 @@
 from ..db.session import get_db
 
 # Access to the websocket
-from ..routers.socketio import get_socket_id_for_pid, sio
+from ..routers.socketio import get_socket_id_for_pid, sio, safe_emit
 
 from ..verificationConfigs.crud import VerificationConfigCRUD
 from ..verificationConfigs.helpers import VariableSubstitutionError
@@ -80,7 +80,7 @@ async def poll_pres_exch_complete(pid: str, db: Database = Depends(get_db)):
         )
         # Send message through the websocket.
         if sid:
-            await sio.emit("status", {"status": "expired"}, to=sid)
+            await safe_emit("status", {"status": "expired"}, to=sid)
 
     return {"proof_status": auth_session.proof_status}
 

oidc-controller/api/routers/presentation_request.py

@@ -9,7 +9,7 @@
 from ..authSessions.models import AuthSession, AuthSessionState
 
 from ..core.config import settings
-from ..routers.socketio import sio, get_socket_id_for_pid
+from ..routers.socketio import sio, get_socket_id_for_pid, safe_emit
 from ..routers.oidc import gen_deep_link
 from ..db.session import get_db
 
@@ -24,7 +24,7 @@ async def toggle_pending(db, auth_session: AuthSession):
     await AuthSessionCRUD(db).patch(auth_session.id, auth_session)
     sid = await get_socket_id_for_pid(str(auth_session.id), db)
     if sid:
-        await sio.emit("status", {"status": "pending"}, to=sid)
+        await safe_emit("status", {"status": "pending"}, to=sid)
 
 
 @router.get("/url/pres_exch/{pres_exch_id}")