fix(telegram): report persistent polling failures

This commit is contained in:
jxxghp
2026-10-08 13:23:58 +08:00
parent 97a6dc3fa1
commit cf889ba7a0
3 changed files with 137 additions and 5 deletions

View File

@@ -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:

View File

@@ -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,避免泄露本机路径和过长输出。连续重复的同类日志模板会保留首条、末条和重复次数,避免轮询或等待日志挤掉真正的错误上下文。

View File

@@ -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 "<token>" 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": []})