From 71a0e8c1d6558310281c42cba1cf2d723614cc6c Mon Sep 17 00:00:00 2001 From: kaireewu Date: Sat, 31 Oct 2026 22:01:05 +0800 Subject: [PATCH] fix(telegram): back off on transient polling failures Give getUpdates a wider HTTP timeout than the long-poll window, log transient SSL/network disconnects as warnings, apply exponential backoff with recovery logging, and only escalate to error after five consecutive failures. --- reaction_board/notify/telegram.py | 17 ++++++-- reaction_board/services/telegram_poller.py | 38 +++++++++++++++-- tests/test_telegram.py | 48 ++++++++++++++++++++++ 3 files changed, 96 insertions(+), 7 deletions(-) diff --git a/reaction_board/notify/telegram.py b/reaction_board/notify/telegram.py index 3ff41d7..fbe2b97 100644 --- a/reaction_board/notify/telegram.py +++ b/reaction_board/notify/telegram.py @@ -86,9 +86,17 @@ class TelegramClient: self.poll_timeout = poll_timeout self.logger = logger - def _call(self, method: str, params: Mapping[str, Any] | None = None) -> Any: + def _call( + self, + method: str, + params: Mapping[str, Any] | None = None, + *, + timeout: float | None = None, + ) -> Any: url = f"https://api.telegram.org/bot{self.token}/{method}" - payload = self.transport.post(url, params or {}, timeout=self.timeout) + payload = self.transport.post( + url, params or {}, timeout=self.timeout if timeout is None else timeout + ) if not isinstance(payload, Mapping): raise TelegramError("Telegram 响应不是对象") if not payload.get("ok"): @@ -145,7 +153,10 @@ class TelegramClient: } if offset is not None: params["offset"] = int(offset) - result = self._call("getUpdates", params) + # 长轮询本身会挂起 poll_timeout 秒，HTTP 读超时必须留出更大余量。 + result = self._call( + "getUpdates", params, timeout=self.poll_timeout + 15 + ) return result if isinstance(result, list) else [] def answer_callback(self, callback_id: str, text: str = "") -> Any: diff --git a/reaction_board/services/telegram_poller.py b/reaction_board/services/telegram_poller.py index 917338e..215830c 100644 --- a/reaction_board/services/telegram_poller.py +++ b/reaction_board/services/telegram_poller.py @@ -88,11 +88,41 @@ class TelegramPoller: self._log("warning", "telegram.answer_failed", error=str(exc)) return False - def run_forever(self, stop_event=None) -> None: + def run_forever( + self, + stop_event=None, + *, + initial_delay: float = 3.0, + max_delay: float = 60.0, + max_iterations: int | None = None, + ) -> None: offset: int | None = None + failures = 0 + delay = initial_delay + iterations = 0 while stop_event is None or not stop_event.is_set(): try: offset = self.run_once(offset) - except Exception as exc: # noqa: BLE001 长轮询失败后退避重试 - self._log("error", "telegram.poll_failed", error=str(exc)) - self._sleep(3) + except Exception as exc: # noqa: BLE001 长轮询瞬时断连后退避重试 + failures += 1 + level = "error" if failures >= 5 else "warning" + self._log( + level, "telegram.poll_failed", + error=str(exc), failures=failures, retry_in=delay, + ) + if stop_event is not None: + if stop_event.wait(delay): + break + else: + self._sleep(delay) + delay = min(delay * 2, max_delay) + else: + if failures: + self._log( + "info", "telegram.poll_recovered", failures=failures + ) + failures = 0 + delay = initial_delay + iterations += 1 + if max_iterations is not None and iterations >= max_iterations: + break diff --git a/tests/test_telegram.py b/tests/test_telegram.py index 4fc5b8b..bcd8eac 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -87,6 +87,7 @@ def test_get_updates_and_answer(): data = transport.calls[0]["data"] assert data["offset"] == 99 assert data["timeout"] == 17 + assert transport.calls[0]["timeout"] == 32 assert json.loads(data["allowed_updates"]) == ["callback_query"] transport = FakeTransport(result=True) @@ -203,6 +204,52 @@ def test_answer_failure_falls_back_to_text(): assert bot.edits == [10] +class Recorder: + def __init__(self): + self.events = [] + + def _add(self, level, event, fields): + self.events.append((level, event, fields)) + + def info(self, event, **fields): + self._add("info", event, fields) + + def warning(self, event, **fields): + self._add("warning", event, fields) + + def error(self, event, **fields): + self._add("error", event, fields) + + def debug(self, event, **fields): + self._add("debug", event, fields) + + +class FlakyBot(FakeBot): + def __init__(self): + super().__init__([]) + self.calls = 0 + + def get_updates(self, offset=None): + self.calls += 1 + if self.calls < 3: + raise OSError("SSL: UNEXPECTED_EOF_WHILE_READING") + return [] + + +def test_poller_backoff_and_recovery(): + sleeps = [] + recorder = Recorder() + bot = FlakyBot() + poller = TelegramPoller( + bot, FakeActions(), logger=recorder, sleeper=sleeps.append + ) + poller.run_forever(initial_delay=3, max_delay=60, max_iterations=3) + assert bot.calls == 3 + assert sleeps == [3, 6] + assert [event[0] for event in recorder.events[:2]] == ["warning", "warning"] + assert any(event[1] == "telegram.poll_recovered" for event in recorder.events) + + def main(): test_format_duration() test_send_candidates_and_chunking() @@ -211,6 +258,7 @@ def main(): test_telegram_error_description_preserved() test_poller_dispatch_and_authorization() test_answer_failure_falls_back_to_text() + test_poller_backoff_and_recovery() print("OK: Telegram 消息、分片、长轮询与回调授权测试全部通过")