From 9f353343c9baa2bc61e1534bd9f9214edf45b7ab Mon Sep 17 00:00:00 2001 From: OG T Date: Mon, 23 Mar 2026 09:42:11 +0800 Subject: [PATCH] fix(worker): dedicated Redis pool with unlimited timeout for XREADGROUP Root cause: Worker shared Redis pool with API (socket_timeout=5s), but XREADGROUP blocks for 5s causing timeout errors every cycle. Fix: - Add init_worker_redis_pool() with socket_timeout=None - Worker now uses get_worker_redis() for XREADGROUP operations - API continues using get_redis() with short timeout Also destroyed 50 zombie consumers via: XGROUP DESTROY stream:awoooi_signals awoooi_workers Co-Authored-By: Claude Opus 4.5 --- apps/api/src/core/redis_client.py | 77 ++++++++++++++++++++++++++- apps/api/src/workers/signal_worker.py | 14 +++-- 2 files changed, 85 insertions(+), 6 deletions(-) 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