diff --git a/app/modules/telegram/botapi.py b/app/modules/telegram/botapi.py index e3458c0ce..a1847f181 100644 --- a/app/modules/telegram/botapi.py +++ b/app/modules/telegram/botapi.py @@ -8,6 +8,7 @@ Telegram Bot API 精简客户端。 import json import threading +import time from typing import Any, Callable, Dict, List, Optional, Union, cast from urllib.parse import urljoin @@ -19,6 +20,8 @@ DEFAULT_API_URL = "https://api.telegram.org" # 普通请求的连接/读取超时，与原 SDK 默认值保持一致。 _CONNECT_TIMEOUT_SECONDS = 15 _READ_TIMEOUT_SECONDS = 30 +# 持续失败仍须在默认日志级别可见，同时避免网络抖动时每次重试都刷屏。 +_POLLING_WARNING_INTERVAL_SECONDS = 60 # 轮询只接收宿主处理的两类更新，其余更新类型不进入 getUpdates 结果。 _ALLOWED_UPDATES = ["message", "callback_query"] # 群聊/私聊中宿主会转发的消息内容类型，与原 SDK message_handler 注册的类型一致。 @@ -41,6 +44,7 @@ class TelegramApiError(Exception): """ def __init__(self, method: str, description: str, error_code: Optional[int] = None): + """保留 API 错误码和描述，供调用方区分业务拒绝与网络失败。""" self.method = method self.description = description self.error_code = error_code @@ -58,6 +62,7 @@ class TelegramNetworkError(Exception): """ def __init__(self, method: str, description: str): + """封装请求层已经脱敏的网络错误描述。""" self.method = method self.description = description super().__init__(f"Telegram API {method} 请求失败：{description}") @@ -404,8 +409,8 @@ class TelegramBotApi: 在当前线程循环拉取更新并逐条交给 handler，直到 ``stop_polling`` 被调用。 更新按 update_id 顺序串行处理；handler 抛错只记录日志，仍确认该更新， - 避免同一条坏消息被反复投递。网络或 API 失败按指数退避重试，失败期间只在 - 首次失败和恢复时各记录一条日志。 + 避免同一条坏消息被反复投递。网络或 API 失败按指数退避重试，首次失败立即 + 警告，持续失败每分钟再警告一次；恢复时记录失败次数和持续时间。 停止后仍在进行的 getUpdates 返回时，结果直接丢弃且不确认 offset，Telegram 会把这些更新重新投递给下一个轮询者；新轮询者的 getUpdates 也会让 Telegram @@ -413,6 +418,8 @@ class TelegramBotApi: """ offset: Optional[int] = None failures = 0 + failure_started_at = 0.0 + next_warning_at = 0.0 while not self._polling_stop.is_set(): try: updates = self.get_updates(offset, long_polling_timeout) @@ -420,14 +427,27 @@ class TelegramBotApi: if self._polling_stop.is_set(): return failures += 1 + now = time.monotonic() + retry_seconds = min(3 * 2 ** (failures - 1), retry_max_seconds) if failures == 1: - logger.warning(f"Telegram消息拉取失败，将自动重试：{err}") + failure_started_at = now + next_warning_at = now + if now >= next_warning_at: + logger.warning( + f"Telegram消息拉取失败，将自动重试（连续{failures}次，" + f"已持续{int(now - failure_started_at)}秒，" + f"{retry_seconds}秒后重试）：{err}" + ) + next_warning_at = now + _POLLING_WARNING_INTERVAL_SECONDS else: logger.debug(f"Telegram消息拉取第{failures}次失败：{err}") - self._polling_stop.wait(min(3 * 2 ** (failures - 1), retry_max_seconds)) + self._polling_stop.wait(retry_seconds) continue if failures: - logger.info("Telegram消息拉取已恢复") + logger.info( + f"Telegram消息拉取已恢复（此前连续{failures}次失败，" + f"持续{int(time.monotonic() - failure_started_at)}秒）" + ) failures = 0 for update in updates or []: with self._dispatch_lock: diff --git a/docs/doctor.md b/docs/doctor.md index 6964de65a..b07d9691e 100644 --- a/docs/doctor.md +++ b/docs/doctor.md @@ -93,6 +93,19 @@ Alembic head 校验和生命周期启动完成后才返回 200；启动失败或 数据库路径、revision、插件名称或异常栈；详细原因应通过本地 `moviepilot doctor`、受控 Agent 诊断或管理员渠道查看。 +## Telegram 收信中断 + +Telegram 长轮询失败后会按指数退避自动重试，默认等待上限为 60 秒。首次失败立即记录 +`Telegram消息拉取失败` 警告；持续失败时，在失败请求返回后每隔至少 60 秒再记录一次， +包含连续失败次数、从首次失败开始的持续时间、下一次等待秒数和最新错误。恢复后记录 +`Telegram消息拉取已恢复`，同时给出本次连续失败次数和持续时间。 + +排查时保留最新的持续失败警告，区分连接或读取超时与 Telegram 返回的 API 错误。 +通过同一代理访问 Telegram 首页成功不能证明 `getUpdates` 长轮询可用；Doctor 或容器 +健康检查正常也不能证明 Telegram 收信正常。线程栈停在退避的 `Event.wait()` 时会表现为 +futex 等待，仅凭 `/proc` 的等待状态无法认定内部死锁。尚未恢复时，先保留失败日志和 +线程调用栈，再重载通知配置；不要采集或公开 Bot token 等局部变量。 + ## Issue 反馈集成 `feedback-issue` skill 的诊断收集脚本会自动调用 `moviepilot doctor --json`，并把 doctor 摘要写入预览和最终 Issue 正文。完整 doctor JSON 存在运行时 diagnostics 文件中，默认不会直接贴入 Issue，避免泄露本机路径和过长输出。连续重复的同类日志模板会保留首条、末条和重复次数，避免轮询或等待日志挤掉真正的错误上下文。 diff --git a/tests/test_telegram_botapi.py b/tests/test_telegram_botapi.py index 505740025..e6bc97eb2 100644 --- a/tests/test_telegram_botapi.py +++ b/tests/test_telegram_botapi.py @@ -15,6 +15,7 @@ from app.modules.telegram.botapi import TelegramApiError, TelegramBotApi, Telegr def _response(payload=None, status_code=200, text=None): """构造 requests.Response 替身；payload 为 None 时 json() 抛 ValueError。""" def _json(): + """返回预设 JSON，模拟非 JSON 响应时保留解析错误。""" if payload is None: raise ValueError(text) return payload @@ -143,6 +144,7 @@ def test_polling_dispatches_in_order_and_survives_handler_errors(): handled = [] def fake_get_updates(offset, timeout): + """依次交付两批更新，最后停止轮询。""" offsets.append(offset) if batches: return batches.pop(0) @@ -150,6 +152,7 @@ def test_polling_dispatches_in_order_and_survives_handler_errors(): return [] def handler(update): + """记录分发顺序并模拟首条消息处理失败。""" handled.append(update["n"]) if update["n"] == 1: raise RuntimeError("boom") @@ -167,6 +170,7 @@ def test_polling_backs_off_on_failure_and_stops_promptly(): calls = threading.Event() def failing_get_updates(offset, timeout): + """通知测试线程已开始拉取，再模拟网络失败。""" calls.set() raise ConnectionError("network down") @@ -181,6 +185,101 @@ def test_polling_backs_off_on_failure_and_stops_promptly(): assert not thread.is_alive() +def test_polling_reports_persistent_failures_and_resets_after_recovery(): + """失败每分钟警告，恢复后再次中断立即警告，offset 和退避计数均保持正确。""" + client = TelegramBotApi("1:abc") + failure_times = [0, 1, 59, 60, 119, 120] + replies = [ + [{"update_id": 4}], + *[TelegramNetworkError("getUpdates", f"failure-{index}") for index in range(6)], + [{"update_id": 5}], + TelegramApiError("getUpdates", "Conflict", 409), + [], + ] + + def fake_get_updates(offset, timeout): + """交付更新后持续失败，再模拟恢复和一次新的 API 错误。""" + if replies: + reply = replies.pop(0) + if isinstance(reply, Exception): + raise reply + return reply + client.stop_polling(0) + return [] + + with patch.object(client, "get_updates", side_effect=fake_get_updates) as get_updates, \ + patch.object(client._polling_stop, "wait") as wait, \ + patch("app.modules.telegram.botapi.time.monotonic", + side_effect=[*failure_times, 121, 122, 123]), \ + patch("app.modules.telegram.botapi.logger") as log: + client.polling(lambda update: None, long_polling_timeout=30) + + warnings = [call.args[0] for call in log.warning.call_args_list] + assert len(warnings) == 4 + assert "连续1次，已持续0秒，3秒后重试" in warnings[0] + assert "连续4次，已持续60秒，24秒后重试" in warnings[1] + assert "failure-3" in warnings[1] + assert "连续6次，已持续120秒，60秒后重试" in warnings[2] + assert "连续1次，已持续0秒，3秒后重试" in warnings[3] + assert "Conflict" in warnings[3] + assert log.debug.call_count == 3 + assert [call.args[0] for call in log.info.call_args_list] == [ + "Telegram消息拉取已恢复（此前连续6次失败，持续121秒）", + "Telegram消息拉取已恢复（此前连续1次失败，持续1秒）", + ] + assert [call.args[0] for call in wait.call_args_list] == [3, 6, 12, 24, 48, 60, 3] + assert [call.args[0] for call in get_updates.call_args_list] == [ + None, 5, 5, 5, 5, 5, 5, 5, 6, 6, 6, + ] + + +def test_polling_persistent_warning_hides_token_and_keeps_retry_limit(post_res): + """实际请求边界的超时日志保持脱敏，持续警告不改变调用方指定的退避上限。""" + token = "123456:SECRET-token" + client = TelegramBotApi(token) + post_res.side_effect = requests.ReadTimeout(f"/bot{token}/getUpdates timed out") + waits = [] + + def wait_or_stop(timeout): + """记录等待预算，第二次失败后结束循环。""" + waits.append(timeout) + if len(waits) == 2: + client.stop_polling(0) + + with patch.object(client._polling_stop, "wait", side_effect=wait_or_stop), \ + patch("app.modules.telegram.botapi.time.monotonic", side_effect=[0, 60]), \ + patch("app.modules.telegram.botapi.logger") as log: + client.polling(lambda update: None, long_polling_timeout=30, retry_max_seconds=4) + + assert post_res.call_count == 2 + assert waits == [3, 4] + assert log.warning.call_count == 2 + for call in log.warning.call_args_list: + assert token not in call.args[0] + assert "" in call.args[0] + assert "ReadTimeout" in call.args[0] + + +def test_polling_stop_during_failed_request_does_not_warn_or_wait(): + """停止期间返回的请求错误直接退出，不报持续失败也不再退避重试。""" + client = TelegramBotApi("1:abc") + + def stop_then_fail(offset, timeout): + """模拟请求完成前模块已经停止。""" + client.stop_polling(0) + raise TelegramNetworkError("getUpdates", "stopped") + + with patch.object(client, "get_updates", side_effect=stop_then_fail) as get_updates, \ + patch.object(client._polling_stop, "wait") as wait, \ + patch("app.modules.telegram.botapi.logger") as log: + client.polling(lambda update: None, long_polling_timeout=30) + + get_updates.assert_called_once() + wait.assert_not_called() + log.warning.assert_not_called() + log.info.assert_not_called() + + def test_get_updates_requests_only_handled_update_types(post_res): """getUpdates 只订阅消息和按钮回调，读取超时大于长轮询等待。""" post_res.return_value = _response({"ok": True, "result": []})