fix(webhook): устранить гонки lifecycle

Запрос во время shutdown:
Начальная проверка могла пройти до route.match(), security, resolve bot или JSON,
после чего TaskTracker уже закрывался, а update запускался и получал ответ 200.
Состояние теперь проверяется повторно прямо перед spawn без последующих await.
Общий guard выбран, чтобы одинаково защитить все webhook engines.

Конкурентные startup и shutdown:
Startup мог снять флаг остановки до завершения параллельного shutdown.
Lifecycle callbacks сериализованы одним asyncio.Lock.
Это сохраняет повторный startup, но исключает наложение переходов состояния.

Подписка SingleBotEngine до startup:
Route с BotIdParam обращался к bot.info до загрузки данных и падал StateError.
Subscribe теперь вызывает get_my_info только для еще не запущенного бота.
Проверка стоит до webhook kwargs и URL, потому что оба могут зависеть от info.

Частичный startup нескольких ботов:
Ошибка второго бота оставляла первый запущенным, а workflow_data содержал stale bot.
Аварийный путь теперь вызывает shutdown-сигналы, закрывает новые экземпляры,
восстанавливает реестр и workflow_data.
Rollback повторяет lifecycle-контракт и не оставляет частично активное состояние.

Порядок multi-bot shutdown:
BeforeShutdown вызывался до завершения фоновых update handlers.
TaskTracker теперь сначала дожидается задач, затем идут shutdown-сигнал и close.
Порядок выбран по уже используемому SingleBotEngine и исключает гонку cleanup.

Конкурентный реестр TokenEngine:
Add, remove, resolve и shutdown могли одновременно менять bots и token_ids,
терять Bot, создавать tracker для удаленного бота или менять dict при обходе.
Один asyncio.Lock сериализует редкие операции с реестром, а identity-check
запрещает spawn для уже удаленного или замененного экземпляра.
Один lock проще per-bot locks и достаточен без измеренной проблемы throughput.

Замена и удаление Bot:
Старый mapping удалялся до успешного unsubscribe, а новый кандидат мог утечь
при ошибке cleanup.
Mapping теперь меняется только после успешного unsubscribe, tracker завершается
до close, а кандидаты закрываются на обоих путях add и request resolve.
Так ошибка сохраняет рабочий старый Bot и не оставляет локальные ресурсы.

Ошибки Bot API при resolve:
Любой MaxBotApiError превращался в 404, включая временную недоступность MAX.
Ошибки токена по-прежнему скрываются за 404, а серверные ошибки оборачиваются
в BotStartError и возвращают контролируемый 502.
Это отличает неизвестного бота от сбоя внешнего API без раскрытия токена.

Проверки:
Добавлены регрессионные тесты для гонок, rollback, BotIdParam, malformed JSON,
ошибок API и отказов cleanup.
Полный прогон: 1705 тестов, ruff, mypy, codespell, slotscheck и bandit.

Author: K1rL3s

src/maxo/transport/webhook/engines/base.py

@@ -1,3 +1,4 @@
+import asyncio
 from abc import ABC, abstractmethod
 from typing import Any, Generic, TypeVar
 
@@ -44,6 +45,7 @@ def __init__(
 
         self.shutdown_timeout = shutdown_timeout
         self._is_shutting_down = False
+        self._lifecycle_lock = asyncio.Lock()
 
     def register(self, app: AppT) -> None:
         webhook.info(
@@ -71,8 +73,7 @@ async def handle_request(
         request: WebRequest[RawRequestT],
     ) -> FrameworkResponseT:
         try:
-            if self._is_shutting_down:
-                raise RequestHandlingStoppedError
+            self._ensure_accepting_requests()
 
             route_params = await self.route.match(request)
 
@@ -98,6 +99,8 @@ async def handle_request(
 
             webhook.debug("New update: %s", update.update)
 
+            self._ensure_accepting_requests(bot)
+
             self._get_task_tracker(bot).spawn(  # type: ignore[unused-awaitable]
                 self._feed_update(bot=bot, update=update),
             )
@@ -121,12 +124,14 @@ async def handle_request(
             )
 
     async def on_startup(self, app: AppT, *args: Any, **kwargs: Any) -> None:
-        await self._on_startup(app, *args, **kwargs)
-        self._is_shutting_down = False
+        async with self._lifecycle_lock:
+            await self._on_startup(app, *args, **kwargs)
+            self._is_shutting_down = False
 
     async def on_shutdown(self, app: AppT, *args: Any, **kwargs: Any) -> None:
-        self._is_shutting_down = True
-        await self._on_shutdown(app, *args, **kwargs)
+        async with self._lifecycle_lock:
+            self._is_shutting_down = True
+            await self._on_shutdown(app, *args, **kwargs)
 
     @abstractmethod
     async def _on_startup(self, app: AppT, *args: Any, **kwargs: Any) -> None:
@@ -171,3 +176,7 @@ async def _feed_update(self, bot: Bot, update: MaxoUpdate[Any]) -> None:
         result = await self.dispatcher.feed_update(bot=bot, update=update)
         if isinstance(result, MaxoMethod):
             await bot.silent_call_method(method=result)
+
+    def _ensure_accepting_requests(self, bot: Bot | None = None) -> None:
+        if self._is_shutting_down or (bot is not None and bot.closed):
+            raise RequestHandlingStoppedError

src/maxo/transport/webhook/engines/multi.py

@@ -18,6 +18,7 @@
     FrameworkResponseT,
     RawRequestT,
 )
+from maxo.transport.webhook.engines.errors import RequestHandlingStoppedError
 from maxo.transport.webhook.route import Route
 from maxo.transport.webhook.security import Security
 from maxo.transport.webhook.tasks import TaskTracker
@@ -60,7 +61,10 @@ async def _on_startup(
         bots: Iterable[Bot] | None = None,
         **kwargs: Any,
     ) -> None:
+        initial_bots = dict(self._bots)
+        initial_workflow_data = dict(self.dispatcher.workflow_data)
         all_bots = set(self.bots.values())
+        processed_bots: list[Bot] = []
 
         if bots is not None:
             all_bots |= set(bots)
@@ -69,16 +73,47 @@ async def _on_startup(
             "Starting multi-bot webhook engine with %s bot(s)",
             len(all_bots),
         )
-        for bot in all_bots:
-            lifecycle_data = self._build_lifecycle_data(app=app, bot=bot, **kwargs)
-            self.dispatcher.workflow_data.update(lifecycle_data)
+        try:
+            for bot in all_bots:
+                processed_bots.append(bot)
+                lifecycle_data = self._build_lifecycle_data(
+                    app=app,
+                    bot=bot,
+                    **kwargs,
+                )
+                self.dispatcher.workflow_data.update(lifecycle_data)
 
-            await self.dispatcher.feed_signal(BeforeStartup(), bot)
-            await bot.get_my_info()
-            self._bots[bot.info.id] = bot
-            await self.dispatcher.feed_signal(AfterStartup(), bot)
+                await self.dispatcher.feed_signal(BeforeStartup(), bot)
+                await bot.get_my_info()
+                self._bots[bot.info.id] = bot
+                await self.dispatcher.feed_signal(AfterStartup(), bot)
+        except BaseException:
+            try:
+                if processed_bots:
+                    lifecycle_data = self._build_lifecycle_data(
+                        app=app,
+                        bots=set(processed_bots),
+                        **kwargs,
+                    )
+                    self.dispatcher.workflow_data.update(lifecycle_data)
+                    await self.dispatcher.feed_signal(BeforeShutdown())
+
+                for bot in all_bots - set(initial_bots.values()):
+                    await bot.close()
+
+                if processed_bots:
+                    await self.dispatcher.feed_signal(AfterShutdown())
+            finally:
+                self._bots.clear()
+                self._bots.update(initial_bots)
+                self.dispatcher.workflow_data.clear()
+                self.dispatcher.workflow_data.update(initial_workflow_data)
+            raise
 
     def _get_task_tracker(self, bot: Bot) -> TaskTracker:
+        if self._bots.get(bot.info.id) is not bot:
+            raise RequestHandlingStoppedError
+
         tracker = self._task_trackers.get(bot.info.id)
 
         if tracker is None:
@@ -88,24 +123,28 @@ def _get_task_tracker(self, bot: Bot) -> TaskTracker:
         return tracker
 
     async def _on_shutdown(self, app: AppT, *args: Any, **kwargs: Any) -> None:
+        bots = tuple(self._bots.values())
         webhook.info(
             "Stopping %s with %s bot(s)",
             self.__class__.__name__,
-            len(self._bots),
+            len(bots),
         )
 
+        for bot in bots:
+            if (tracker := self._task_trackers.pop(bot.info.id, None)) is not None:
+                await tracker.close(timeout=self.shutdown_timeout)
+
         lifecycle_data = self._build_lifecycle_data(
             app=app,
-            bots=set(self._bots.values()),
+            bots=set(bots),
             **kwargs,
         )
         self.dispatcher.workflow_data.update(lifecycle_data)
         await self.dispatcher.feed_signal(BeforeShutdown())
 
-        for bot in self._bots.values():
-            if (tracker := self._task_trackers.pop(bot.info.id, None)) is not None:
-                await tracker.close(timeout=self.shutdown_timeout)
+        for bot in bots:
             await bot.close()
 
         self._bots.clear()
+        self._task_trackers.clear()
         await self.dispatcher.feed_signal(AfterShutdown())

src/maxo/transport/webhook/engines/single.py

@@ -58,6 +58,9 @@ async def subscribe(
         self,
         webhook_config: WebhookConfig | None = None,
     ) -> SimpleQueryResult:
+        if not self.bot.started:
+            await self.bot.get_my_info()
+
         kwargs = await self._build_webhook_kwargs(
             bot=self.bot,
             base_config=webhook_config or WebhookConfig(),

src/maxo/transport/webhook/engines/token.py

@@ -1,11 +1,19 @@
+import asyncio
 from typing import Any, Generic
 
 from maxo import Bot, Dispatcher
-from maxo.errors import MaxBotApiError
+from maxo.errors import (
+    MaxBotApiError,
+    MaxBotBadRequestError,
+    MaxBotForbiddenError,
+    MaxBotNotFoundError,
+    MaxBotUnauthorizedError,
+)
 from maxo.loggers import webhook
 from maxo.transport.webhook.configs.bot import BotConfig
 from maxo.transport.webhook.configs.webhook import WebhookConfig
 from maxo.transport.webhook.engines.base import AppT, FrameworkResponseT, RawRequestT
+from maxo.transport.webhook.engines.errors import BotStartError
 from maxo.transport.webhook.engines.multi import BaseMultiBotEngine
 from maxo.transport.webhook.route import Route
 from maxo.transport.webhook.route.params import RouteParams
@@ -43,84 +51,119 @@ def __init__(
         )
         self.bot_config = bot_config
         self._token_ids: dict[str, int] = {}
+        # ponytail: один lock на реестр, per-bot locks нужны только при замерах задержек
+        self._bots_lock = asyncio.Lock()
 
     async def add_bot(
         self,
         token: str,
         webhook_config: WebhookConfig | None = None,
     ) -> Bot:
-        bot = self._build_bot(token)
-        await bot.get_my_info()
+        async with self._bots_lock:
+            bot = self._build_bot(token)
+            await bot.get_my_info()
 
-        kwargs = await self._build_webhook_kwargs(
-            bot=bot,
-            base_config=webhook_config or self.webhook_config,
-        )
-        await bot.subscribe(url=await self.route.build_url(bot=bot), **kwargs)
+            kwargs = await self._build_webhook_kwargs(
+                bot=bot,
+                base_config=webhook_config or self.webhook_config,
+            )
+            await bot.subscribe(url=await self.route.build_url(bot=bot), **kwargs)
+
+            if (existing := self._bots.get(bot.info.id)) is not None:
+                # URL contains the token: same token = same URL, already subscribed
+                try:
+                    await self._remove_bot(
+                        bot.info.id,
+                        unsubscribe=existing.token != token,
+                    )
+                except Exception:
+                    await bot.close()
+                    raise
+
+            self._bots[bot.info.id] = bot
+            self._token_ids[token] = bot.info.id
+
+            webhook.info("Added bot %s to token engine and set webhook", bot.info.id)
+            return bot
 
-        if (existing := self._bots.get(bot.info.id)) is not None:
-            # URL contains the token: same token = same URL, already subscribed
-            await self.remove_bot(bot.info.id, unsubscribe=existing.token != token)
+    async def remove_bot(self, bot_id: int, unsubscribe: bool = True) -> bool:
+        async with self._bots_lock:
+            return await self._remove_bot(bot_id, unsubscribe=unsubscribe)
 
-        self._bots[bot.info.id] = bot
-        self._token_ids[token] = bot.info.id
+    async def _resolve_bot(self, route_params: RouteParams) -> Bot | None:
+        async with self._bots_lock:
+            token = route_params.get("bot_token")
+            if not isinstance(token, str) or not token:
+                return None
+
+            if (bot_id := self._token_ids.get(token)) is not None:
+                return self._bots.get(bot_id)
+
+            bot = self._build_bot(token)
+            try:
+                await bot.get_my_info()
+            except (
+                MaxBotBadRequestError,
+                MaxBotForbiddenError,
+                MaxBotNotFoundError,
+                MaxBotUnauthorizedError,
+            ):
+                await bot.close()
+                return None
+            except MaxBotApiError as exc:
+                await bot.close()
+                raise BotStartError(original_error=exc) from exc
+
+            existing = self._bots.get(bot.info.id)
+            if existing is not None and existing.token == token:
+                self._token_ids[token] = existing.info.id
+                await bot.close()
+                return existing
+
+            if existing is not None:
+                try:
+                    await self._remove_bot(existing.info.id)
+                except Exception as exc:
+                    await bot.close()
+                    raise BotStartError(original_error=exc) from exc
+
+            self._bots[bot.info.id] = bot
+            self._token_ids[token] = bot.info.id
+            return bot
 
-        webhook.info("Added bot %s to token engine and set webhook", bot.info.id)
-        return bot
+    async def _on_shutdown(self, app: AppT, *args: Any, **kwargs: Any) -> None:
+        async with self._bots_lock:
+            await super()._on_shutdown(app, *args, **kwargs)
+            self._token_ids.clear()
 
-    async def remove_bot(self, bot_id: int, unsubscribe: bool = True) -> bool:
-        bot = self._bots.pop(bot_id, None)
+    def _build_bot(self, token: str) -> Bot:
+        return Bot(
+            token=token,
+            client=self.bot_config.client,
+            defaults=self.bot_config.defaults,
+            upload_config=self.bot_config.upload_config,
+            warming_up=self.bot_config.warming_up,
+        )
+
+    async def _remove_bot(self, bot_id: int, *, unsubscribe: bool = True) -> bool:
+        bot = self._bots.get(bot_id)
         if bot is None:
             return False
 
-        self._token_ids.pop(bot.token, None)
+        if unsubscribe:
+            await bot.unsubscribe(url=await self.route.build_url(bot=bot))
 
+        self._bots.pop(bot_id, None)
+        self._token_ids.pop(bot.token, None)
         try:
-            if unsubscribe:
-                await bot.unsubscribe(url=await self.route.build_url(bot=bot))
-        finally:
             if (tracker := self._task_trackers.pop(bot_id, None)) is not None:
                 await tracker.close(timeout=self.shutdown_timeout)
+        finally:
             await bot.close()
 
         webhook.info("Removed bot %s from token engine", bot_id)
         return True
 
-    async def _resolve_bot(self, route_params: RouteParams) -> Bot | None:
-        token = route_params.get("bot_token")
-        if not isinstance(token, str) or not token:
-            return None
-
-        if (bot_id := self._token_ids.get(token)) is not None:
-            return self._bots.get(bot_id)
-
-        bot = self._build_bot(token)
-        try:
-            await bot.get_my_info()
-        except MaxBotApiError:
-            return None
-
-        existing = self._bots.get(bot.info.id)
-        if existing is not None and existing.token == token:
-            self._token_ids[token] = existing.info.id
-            return existing
-
-        if existing is not None:
-            self._token_ids.pop(existing.token, None)
-
-        self._bots[bot.info.id] = bot
-        self._token_ids[token] = bot.info.id
-        return bot
-
-    async def _on_shutdown(self, app: AppT, *args: Any, **kwargs: Any) -> None:
-        await super()._on_shutdown(app, *args, **kwargs)
-        self._token_ids.clear()
-
-    def _build_bot(self, token: str) -> Bot:
-        return Bot(
-            token=token,
-            client=self.bot_config.client,
-            defaults=self.bot_config.defaults,
-            upload_config=self.bot_config.upload_config,
-            warming_up=self.bot_config.warming_up,
-        )
+    async def _on_startup(self, app: AppT, *args: Any, **kwargs: Any) -> None:
+        async with self._bots_lock:
+            await super()._on_startup(app, *args, **kwargs)

tests/maxo_webhook/fixtures/shutdown.py

@@ -85,3 +85,22 @@ async def make_request(self, request: HTTPRequest) -> HTTPResponse[Any]:
 
     async def close(self) -> None:
         self.closed = True
+
+
+class SubscribingClient(TrackableClient):
+    def __init__(self, bot_id: int = 42) -> None:
+        super().__init__(bot_id=bot_id)
+        self.subscribed: list[str] = []
+
+    async def make_request(self, request: HTTPRequest) -> HTTPResponse[Any]:
+        if request.url != "subscriptions":
+            return await super().make_request(request)
+
+        self.subscribed.append(request.body["url"])
+        return HTTPResponse(
+            status_code=200,
+            headers={},
+            data={"success": True},
+            cookies={},
+            raw_response=None,
+        )