fix(admin): avoid stale Telegram webhook alerts
Only surface Telegram delivery errors while updates are still pending, and register the webhook after the aiohttp webhook site starts listening.
This commit is contained in:
@@ -2,6 +2,7 @@ import asyncio
|
|||||||
import functools
|
import functools
|
||||||
import hmac
|
import hmac
|
||||||
import logging
|
import logging
|
||||||
|
from typing import Awaitable, Callable, Optional
|
||||||
|
|
||||||
from aiogram import Bot, Dispatcher
|
from aiogram import Bot, Dispatcher
|
||||||
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
|
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
|
||||||
@@ -84,6 +85,8 @@ async def build_and_start_web_app(
|
|||||||
bot: Bot,
|
bot: Bot,
|
||||||
settings: Settings,
|
settings: Settings,
|
||||||
async_session_factory: sessionmaker,
|
async_session_factory: sessionmaker,
|
||||||
|
*,
|
||||||
|
after_webhooks_started: Optional[Callable[[], Awaitable[None]]] = None,
|
||||||
):
|
):
|
||||||
app = web.Application()
|
app = web.Application()
|
||||||
_inject_shared_instances(app, dp, bot, settings, async_session_factory)
|
_inject_shared_instances(app, dp, bot, settings, async_session_factory)
|
||||||
@@ -159,6 +162,8 @@ async def build_and_start_web_app(
|
|||||||
logging.info(
|
logging.info(
|
||||||
f"AIOHTTP server started on http://{settings.WEB_SERVER_HOST}:{settings.WEB_SERVER_PORT}"
|
f"AIOHTTP server started on http://{settings.WEB_SERVER_HOST}:{settings.WEB_SERVER_PORT}"
|
||||||
)
|
)
|
||||||
|
if after_webhooks_started is not None:
|
||||||
|
await after_webhooks_started()
|
||||||
|
|
||||||
if settings.WEBAPP_ENABLED:
|
if settings.WEBAPP_ENABLED:
|
||||||
from bot.app.web.subscription_webapp import create_subscription_webapp_application
|
from bot.app.web.subscription_webapp import create_subscription_webapp_application
|
||||||
|
|||||||
+19
-5
@@ -91,12 +91,9 @@ async def register_all_routers(dp: Dispatcher, settings: Settings):
|
|||||||
logging.info("All application routers registered.")
|
logging.info("All application routers registered.")
|
||||||
|
|
||||||
|
|
||||||
async def on_startup_configured(dispatcher: Dispatcher):
|
async def configure_telegram_webhook(dispatcher: Dispatcher) -> None:
|
||||||
bot: Bot = dispatcher["bot_instance"]
|
bot: Bot = dispatcher["bot_instance"]
|
||||||
settings: Settings = dispatcher["settings"]
|
settings: Settings = dispatcher["settings"]
|
||||||
i18n_instance: JsonI18n = dispatcher["i18n_instance"]
|
|
||||||
|
|
||||||
logging.info("STARTUP: on_startup_configured executing...")
|
|
||||||
|
|
||||||
telegram_webhook_url_to_set = settings.WEBHOOK_BASE_URL
|
telegram_webhook_url_to_set = settings.WEBHOOK_BASE_URL
|
||||||
if telegram_webhook_url_to_set:
|
if telegram_webhook_url_to_set:
|
||||||
@@ -152,6 +149,14 @@ async def on_startup_configured(dispatcher: Dispatcher):
|
|||||||
)
|
)
|
||||||
raise SystemExit("WEBHOOK_BASE_URL is required. Polling mode is disabled.")
|
raise SystemExit("WEBHOOK_BASE_URL is required. Polling mode is disabled.")
|
||||||
|
|
||||||
|
|
||||||
|
async def on_startup_configured(dispatcher: Dispatcher):
|
||||||
|
bot: Bot = dispatcher["bot_instance"]
|
||||||
|
settings: Settings = dispatcher["settings"]
|
||||||
|
i18n_instance: JsonI18n = dispatcher["i18n_instance"]
|
||||||
|
|
||||||
|
logging.info("STARTUP: on_startup_configured executing...")
|
||||||
|
|
||||||
if settings.SUBSCRIPTION_MINI_APP_URL:
|
if settings.SUBSCRIPTION_MINI_APP_URL:
|
||||||
|
|
||||||
async def _configure_mini_app_menu() -> None:
|
async def _configure_mini_app_menu() -> None:
|
||||||
@@ -331,8 +336,17 @@ async def run_bot(settings_param: Settings):
|
|||||||
_yk_path,
|
_yk_path,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def _after_webhooks_started() -> None:
|
||||||
|
await configure_telegram_webhook(dp)
|
||||||
|
|
||||||
async def web_server_task():
|
async def web_server_task():
|
||||||
await build_and_start_web_app(dp, bot, settings_param, local_async_session_factory)
|
await build_and_start_web_app(
|
||||||
|
dp,
|
||||||
|
bot,
|
||||||
|
settings_param,
|
||||||
|
local_async_session_factory,
|
||||||
|
after_webhooks_started=_after_webhooks_started,
|
||||||
|
)
|
||||||
|
|
||||||
main_tasks = [asyncio.create_task(web_server_task(), name="AIOHTTPServerTask")]
|
main_tasks = [asyncio.create_task(web_server_task(), name="AIOHTTPServerTask")]
|
||||||
|
|
||||||
|
|||||||
@@ -385,6 +385,7 @@ async def telegram_alerts(bot: Any, settings: Any) -> List[ConfigAlert]:
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
pending = int(getattr(info, "pending_update_count", 0) or 0)
|
||||||
last_error_date = getattr(info, "last_error_date", None)
|
last_error_date = getattr(info, "last_error_date", None)
|
||||||
last_error_ts: Optional[float] = None
|
last_error_ts: Optional[float] = None
|
||||||
if last_error_date is not None:
|
if last_error_date is not None:
|
||||||
@@ -393,7 +394,11 @@ async def telegram_alerts(bot: Any, settings: Any) -> List[ConfigAlert]:
|
|||||||
if hasattr(last_error_date, "timestamp")
|
if hasattr(last_error_date, "timestamp")
|
||||||
else float(last_error_date)
|
else float(last_error_date)
|
||||||
)
|
)
|
||||||
if last_error_ts and (time.time() - last_error_ts) < _WEBHOOK_ERROR_RECENT_SECONDS:
|
if (
|
||||||
|
pending > 0
|
||||||
|
and last_error_ts
|
||||||
|
and (time.time() - last_error_ts) < _WEBHOOK_ERROR_RECENT_SECONDS
|
||||||
|
):
|
||||||
alerts.append(
|
alerts.append(
|
||||||
ConfigAlert(
|
ConfigAlert(
|
||||||
id="telegram_webhook_error",
|
id="telegram_webhook_error",
|
||||||
@@ -403,7 +408,6 @@ async def telegram_alerts(bot: Any, settings: Any) -> List[ConfigAlert]:
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
pending = int(getattr(info, "pending_update_count", 0) or 0)
|
|
||||||
if pending > _WEBHOOK_PENDING_THRESHOLD:
|
if pending > _WEBHOOK_PENDING_THRESHOLD:
|
||||||
alerts.append(
|
alerts.append(
|
||||||
ConfigAlert(
|
ConfigAlert(
|
||||||
|
|||||||
@@ -239,16 +239,26 @@ class TelegramAlertsTests(unittest.IsolatedAsyncioTestCase):
|
|||||||
alerts = await health.telegram_alerts(bot, _settings())
|
alerts = await health.telegram_alerts(bot, _settings())
|
||||||
self.assertEqual(_alert_ids(alerts), ["telegram_webhook_mismatch"])
|
self.assertEqual(_alert_ids(alerts), ["telegram_webhook_mismatch"])
|
||||||
|
|
||||||
async def test_recent_delivery_error_reported(self):
|
async def test_recent_delivery_error_with_pending_update_reported(self):
|
||||||
info = self._webhook_info(
|
info = self._webhook_info(
|
||||||
last_error_date=datetime.now(timezone.utc),
|
last_error_date=datetime.now(timezone.utc),
|
||||||
last_error_message="SSL error",
|
last_error_message="SSL error",
|
||||||
|
pending_update_count=1,
|
||||||
)
|
)
|
||||||
bot = SimpleNamespace(get_webhook_info=AsyncMock(return_value=info))
|
bot = SimpleNamespace(get_webhook_info=AsyncMock(return_value=info))
|
||||||
alerts = await health.telegram_alerts(bot, _settings())
|
alerts = await health.telegram_alerts(bot, _settings())
|
||||||
self.assertEqual(_alert_ids(alerts), ["telegram_webhook_error"])
|
self.assertEqual(_alert_ids(alerts), ["telegram_webhook_error"])
|
||||||
self.assertEqual(alerts[0].params["error"], "SSL error")
|
self.assertEqual(alerts[0].params["error"], "SSL error")
|
||||||
|
|
||||||
|
async def test_recent_delivery_error_without_pending_update_not_reported(self):
|
||||||
|
info = self._webhook_info(
|
||||||
|
last_error_date=datetime.now(timezone.utc),
|
||||||
|
last_error_message="Connection refused",
|
||||||
|
pending_update_count=0,
|
||||||
|
)
|
||||||
|
bot = SimpleNamespace(get_webhook_info=AsyncMock(return_value=info))
|
||||||
|
self.assertEqual(await health.telegram_alerts(bot, _settings()), [])
|
||||||
|
|
||||||
async def test_stale_delivery_error_not_reported(self):
|
async def test_stale_delivery_error_not_reported(self):
|
||||||
info = self._webhook_info(last_error_date=time.time() - 7200)
|
info = self._webhook_info(last_error_date=time.time() - 7200)
|
||||||
bot = SimpleNamespace(get_webhook_info=AsyncMock(return_value=info))
|
bot = SimpleNamespace(get_webhook_info=AsyncMock(return_value=info))
|
||||||
|
|||||||
@@ -45,6 +45,29 @@ def test_worker_starts_backup_task_without_enabled_guard():
|
|||||||
assert guarded_backup_tasks == []
|
assert guarded_backup_tasks == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_telegram_webhook_configuration_is_deferred_until_site_start():
|
||||||
|
main_source = Path("backend/bot/main_bot.py").read_text(encoding="utf-8")
|
||||||
|
web_source = Path("backend/bot/app/web/web_server.py").read_text(encoding="utf-8")
|
||||||
|
tree = ast.parse(main_source)
|
||||||
|
startup_node = next(
|
||||||
|
node
|
||||||
|
for node in ast.walk(tree)
|
||||||
|
if isinstance(node, ast.AsyncFunctionDef) and node.name == "on_startup_configured"
|
||||||
|
)
|
||||||
|
|
||||||
|
startup_set_webhook_calls = [
|
||||||
|
node
|
||||||
|
for node in ast.walk(startup_node)
|
||||||
|
if isinstance(node, ast.Attribute) and node.attr == "set_webhook"
|
||||||
|
]
|
||||||
|
|
||||||
|
assert startup_set_webhook_calls == []
|
||||||
|
assert "after_webhooks_started=_after_webhooks_started" in main_source
|
||||||
|
assert web_source.index("await site.start()") < web_source.index(
|
||||||
|
"await after_webhooks_started()"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_telegram_startup_network_error_retries_until_success_without_traceback(caplog):
|
def test_telegram_startup_network_error_retries_until_success_without_traceback(caplog):
|
||||||
calls = []
|
calls = []
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user