diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 8712b0020..c852f9166 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -24,6 +24,7 @@ SOUL.md 鐵律 (4.1 Telegram 訊息壓縮原則): import asyncio import html +import os from dataclasses import dataclass from datetime import UTC, datetime @@ -48,6 +49,13 @@ SILENCE_KEY_PREFIX = "telegram_silence:" # {resource_name} -> 靜默 SNOOZE_TTL_SECONDS = 30 * 60 # 30 分鐘 SILENCE_TTL_SECONDS = 60 * 60 # 1 小時 +# 2026-04-01 Claude Code: Long Polling 分散式 Leader Election +# 防止多 Pod 同時 getUpdates → 409 Conflict 互搶問題 +POLLING_LEADER_KEY = "telegram:polling:leader" +POLLING_LEADER_TTL = 45 # seconds - Pod 宕掉後 45s 自動轉移 +POLLING_LEADER_RENEW = 20 # seconds - 每 20s 續約 +POLLING_LEADER_WATCH = 30 # seconds - 非 Leader Pod 每 30s 嘗試接管 + logger = structlog.get_logger(__name__) # ============================================================================= @@ -1018,6 +1026,9 @@ class TelegramGateway: self._polling_active = False self._polling_task: asyncio.Task | None = None self._last_update_id = 0 + # 2026-04-01 Claude Code: 分散式 Leader Election (防 2-Pod 409 互搶) + self._pod_id = os.environ.get("POD_NAME", os.urandom(8).hex()) + self._leader_task: asyncio.Task | None = None # Phase 6.5: 心跳監控 (防止沉默盲點) self._last_message_time: datetime | None = None self._heartbeat_task: asyncio.Task | None = None @@ -2232,15 +2243,17 @@ class TelegramGateway: async def close(self) -> None: """關閉 Gateway""" - # 停止 Long Polling + # 停止 Long Polling 與 Leader 相關 Tasks self._polling_active = False - if self._polling_task and not self._polling_task.done(): - self._polling_task.cancel() - try: - await self._polling_task - except asyncio.CancelledError: - pass - self._polling_task = None + for task in (self._polling_task, self._leader_task): + if task and not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self._polling_task = None + self._leader_task = None if self._http_client: await self._http_client.aclose() @@ -2258,6 +2271,9 @@ class TelegramGateway: 取代 Webhook 模式，適用於內網環境 統帥鐵律: 內網無法接收外部 Webhook，必須主動輪詢 + + 2026-04-01 Claude Code: 加入 Redis Leader Election + 多 Pod 環境下，只有 Leader 執行 getUpdates，其餘 Pod 進入 Watcher 模式 """ if not self._initialized: success = await self.initialize() @@ -2269,15 +2285,33 @@ class TelegramGateway: logger.warning("telegram_long_polling_already_running") return - # 先刪除 Webhook，確保可以使用 getUpdates + # 嘗試取得 Leader Lock (NX = 僅在不存在時設定) + redis = await get_redis() + acquired = await redis.set(POLLING_LEADER_KEY, self._pod_id, nx=True, ex=POLLING_LEADER_TTL) + + if not acquired: + current_leader = await redis.get(POLLING_LEADER_KEY) + logger.info( + "telegram_polling_not_leader", + pod_id=self._pod_id, + current_leader=current_leader, + action="watcher_mode", + ) + # 啟動 Watcher：定期嘗試接管 + self._leader_task = asyncio.create_task(self._leader_watcher()) + return + + # 取得 Leader Lock，開始 Polling await self._delete_webhook() self._polling_active = True self._last_update_id = 0 self._polling_task = asyncio.create_task(self._polling_loop()) + self._leader_task = asyncio.create_task(self._leader_renewer()) logger.info( "telegram_long_polling_started", + pod_id=self._pod_id, timeout=LONG_POLLING_TIMEOUT, chat_id=self.chat_id[:10] + "..." if self.chat_id else "N/A", ) @@ -2354,15 +2388,23 @@ class TelegramGateway: except httpx.HTTPStatusError as e: if e.response.status_code == 409: - # 409 Conflict: 另一個 getUpdates session 殘留 - # 可能是舊 pod/host 的連線未清除，等待後重試 + # 409 Conflict: 另一個 Pod 正在 polling，主動釋放 Leader Lock + # 2026-04-01 Claude Code: 改為釋放 Lock 讓 Watcher 競爭 + # (舊: 侵略性搶佔 2s，已不適用 - 現在是多 Pod 場景而非 .188 搶佔) logger.warning( "telegram_polling_conflict", status=409, - message="409 衝突，等待殘留 session 過期...", + pod_id=self._pod_id, + action="releasing_leader_lock", ) - # 2026-03-31 ogt: 侵略性搶佔模式 - 縮短等待以從 .188 手中奪回主權 - await asyncio.sleep(2) + redis = await get_redis() + current = await redis.get(POLLING_LEADER_KEY) + if current == self._pod_id: + await redis.delete(POLLING_LEADER_KEY) + self._polling_active = False + # Watcher 會在 POLLING_LEADER_WATCH 秒後重新競爭 + self._leader_task = asyncio.create_task(self._leader_watcher()) + break else: logger.error("telegram_polling_http_error", status=e.response.status_code) await asyncio.sleep(LONG_POLLING_RETRY_DELAY) @@ -2374,6 +2416,68 @@ class TelegramGateway: logger.info("telegram_long_polling_stopped") + async def _leader_renewer(self) -> None: + """ + Leader Lock 續約背景任務 + + 每 POLLING_LEADER_RENEW 秒更新 Redis TTL， + 確保 Leader 在 Poll 期間持續持有 Lock。 + 若 Lock 被搶走，停止 Polling。 + + 2026-04-01 Claude Code: 分散式 Leader Election + """ + while self._polling_active: + await asyncio.sleep(POLLING_LEADER_RENEW) + if not self._polling_active: + break + try: + redis = await get_redis() + current = await redis.get(POLLING_LEADER_KEY) + if current != self._pod_id: + logger.warning( + "telegram_leader_lock_lost", + pod_id=self._pod_id, + current_leader=current, + ) + self._polling_active = False + break + await redis.expire(POLLING_LEADER_KEY, POLLING_LEADER_TTL) + except Exception as e: + logger.error("telegram_leader_renew_error", error=str(e)) + + async def _leader_watcher(self) -> None: + """ + 非 Leader Pod 的接管監控任務 + + 每 POLLING_LEADER_WATCH 秒嘗試取得 Leader Lock。 + 若原 Leader 宕掉（TTL 過期），此 Pod 接管 Polling。 + + 2026-04-01 Claude Code: 分散式 Leader Election + """ + while not self._polling_active: + await asyncio.sleep(POLLING_LEADER_WATCH) + try: + redis = await get_redis() + acquired = await redis.set( + POLLING_LEADER_KEY, self._pod_id, nx=True, ex=POLLING_LEADER_TTL + ) + if acquired: + logger.info( + "telegram_leader_acquired", + pod_id=self._pod_id, + action="starting_polling", + ) + await self._delete_webhook() + self._polling_active = True + self._last_update_id = 0 + self._polling_task = asyncio.create_task(self._polling_loop()) + self._leader_task = asyncio.create_task(self._leader_renewer()) + break + except asyncio.CancelledError: + break + except Exception as e: + logger.error("telegram_leader_watch_error", error=str(e)) + async def _get_updates(self) -> list[dict]: """ 呼叫 Telegram getUpdates API