diff --git a/services/task-watcher/conftest.py b/services/task-watcher/conftest.py new file mode 100644 index 0000000..1f1d5de --- /dev/null +++ b/services/task-watcher/conftest.py @@ -0,0 +1,5 @@ +"""Make services/ root importable so `from _shared.heartbeat import ...` works during tests.""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) diff --git a/services/task-watcher/tests/test_watcher.py b/services/task-watcher/tests/test_watcher.py new file mode 100644 index 0000000..64fed08 --- /dev/null +++ b/services/task-watcher/tests/test_watcher.py @@ -0,0 +1,16 @@ +"""task-watcher heartbeat payload — state=mode + mode 필드 검증.""" +import json + +from _shared.heartbeat import build_payload, WorkerStats + + +def test_watcher_heartbeat_payload_carries_mode(): + payload = json.loads( + build_payload( + "task-watcher", "watcher", "trading", + WorkerStats(), extra={"mode": "trading"}, + ) + ) + assert payload["kind"] == "watcher" + assert payload["state"] == "trading" + assert payload["mode"] == "trading" diff --git a/services/task-watcher/watcher.py b/services/task-watcher/watcher.py index 15accc1..f3f9463 100644 --- a/services/task-watcher/watcher.py +++ b/services/task-watcher/watcher.py @@ -15,6 +15,7 @@ from zoneinfo import ZoneInfo import redis.asyncio as aioredis from mode import current_mode, fetch_holidays, KST +from _shared.heartbeat import build_payload, WorkerStats logger = logging.getLogger(__name__) @@ -23,6 +24,10 @@ PAUSED_KEY = "queue:paused" LOOP_INTERVAL = 30 # 초 HOLIDAYS_REFRESH = 3600 # 1시간 PAUSED_TTL = 600 # 10분 (watcher 죽어도 자동 해제) +HEARTBEAT_KEY = "worker:task-watcher:heartbeat" +HEARTBEAT_TTL = 45 # LOOP_INTERVAL 30s < TTL 45s → 만료 전 갱신 + +_HB_STATS = WorkerStats() async def watcher_loop(): @@ -46,6 +51,13 @@ async def watcher_loop(): else: await redis.delete(PAUSED_KEY) + # heartbeat (LOOP_INTERVAL=30s < TTL 45s → 만료 전 갱신) + await redis.set( + HEARTBEAT_KEY, + build_payload("task-watcher", "watcher", mode, _HB_STATS, extra={"mode": mode}), + ex=HEARTBEAT_TTL, + ) + if mode != last_mode: logger.info("mode 전환: %s → %s (paused=%s)", last_mode, mode, mode == "trading") last_mode = mode