diff --git a/apps/api/src/core/redis_client.py b/apps/api/src/core/redis_client.py index a23008f06..11c29c4fd 100644 --- a/apps/api/src/core/redis_client.py +++ b/apps/api/src/core/redis_client.py @@ -27,11 +27,17 @@ logger = structlog.get_logger(__name__) # ============================================================================= -# Connection Pool +# Connection Pool (API 用 - 短超時) # ============================================================================= _redis_pool: redis.Redis | None = None +# ============================================================================= +# Worker 專屬連線池 (長超時,用於 XREADGROUP 阻塞操作) +# ============================================================================= + +_worker_redis_pool: redis.Redis | None = None + async def init_redis_pool() -> redis.Redis: """ @@ -83,7 +89,7 @@ async def close_redis_pool() -> None: def get_redis() -> redis.Redis: """ - 取得 Redis 連線 + 取得 Redis 連線 (API 用,短超時) Raises: RuntimeError: 若連線池未初始化 @@ -93,6 +99,73 @@ def get_redis() -> redis.Redis: return _redis_pool +# ============================================================================= +# Worker 專屬連線池 (長超時) +# ============================================================================= + + +async def init_worker_redis_pool() -> redis.Redis: + """ + 初始化 Worker 專屬 Redis 連線池 + + 統帥鐵律 2026-03-23: + - Worker 使用 XREADGROUP 阻塞操作,需要長超時 + - 絕對禁止與 API 共用短超時連線池 + - socket_timeout=None 表示無限等待 + """ + global _worker_redis_pool + + if _worker_redis_pool is not None: + return _worker_redis_pool + + _worker_redis_pool = redis.from_url( + settings.REDIS_URL, + encoding="utf-8", + decode_responses=True, + max_connections=5, # Worker 不需要太多連線 + socket_timeout=None, # 無限等待 (XREADGROUP 阻塞操作) + socket_connect_timeout=10.0, # 連線超時仍需設定 + ) + + # 測試連線 + try: + await _worker_redis_pool.ping() + logger.info( + "worker_redis_pool_initialized", + url=settings.REDIS_URL.split("@")[-1], + socket_timeout="None (unlimited)", + ) + except redis.ConnectionError as e: + logger.error("worker_redis_connection_failed", error=str(e)) + raise + + return _worker_redis_pool + + +async def close_worker_redis_pool() -> None: + """ + 關閉 Worker 專屬 Redis 連線池 + """ + global _worker_redis_pool + + if _worker_redis_pool is not None: + await _worker_redis_pool.close() + _worker_redis_pool = None + logger.info("worker_redis_pool_closed") + + +def get_worker_redis() -> redis.Redis: + """ + 取得 Worker 專屬 Redis 連線 (長超時,用於 XREADGROUP) + + Raises: + RuntimeError: 若連線池未初始化 + """ + if _worker_redis_pool is None: + raise RuntimeError("Worker Redis pool not initialized. Call init_worker_redis_pool() first.") + return _worker_redis_pool + + # ============================================================================= # Distributed Lock (分散式鎖) # ============================================================================= diff --git a/apps/api/src/workers/signal_worker.py b/apps/api/src/workers/signal_worker.py index 8f438d7f1..22d6156b1 100644 --- a/apps/api/src/workers/signal_worker.py +++ b/apps/api/src/workers/signal_worker.py @@ -26,7 +26,7 @@ from typing import Any import structlog -from src.core.redis_client import get_redis +from src.core.redis_client import get_redis, get_worker_redis from src.services.incident_engine import get_incident_engine logger = structlog.get_logger(__name__) @@ -149,8 +149,12 @@ class SignalWorker: 主消費循環 XREADGROUP 阻塞等待新訊息,處理後 XACK。 + + 統帥鐵律 2026-03-23: + - 使用 Worker 專屬 Redis 連線 (無超時限制) + - 絕對禁止使用 API 共用的短超時連線 """ - redis_client = get_redis() + redis_client = get_worker_redis() # Worker 專屬長連線 while self._running: try: @@ -321,7 +325,7 @@ async def _main() -> None: # Initialize settings first (loads env vars) from src.core.config import settings # noqa: F401 - from src.core.redis_client import init_redis_pool, close_redis_pool + from src.core.redis_client import init_redis_pool, close_redis_pool, init_worker_redis_pool, close_worker_redis_pool logger.info( "signal_worker_standalone_starting", @@ -329,8 +333,9 @@ async def _main() -> None: redis_url=settings.REDIS_URL.split("@")[-1] if settings.REDIS_URL else "N/A", ) - # Initialize Redis + # Initialize Redis (API pool + Worker 專屬長連線池) await init_redis_pool() + await init_worker_redis_pool() # Worker 專屬,無超時限制 # Write health files for K8s probes await _write_health_files() @@ -354,6 +359,7 @@ async def _main() -> None: # Graceful shutdown logger.info("signal_worker_shutting_down") await worker.stop() + await close_worker_redis_pool() # 關閉 Worker 專屬連線 await close_redis_pool() # Remove health files