diff --git a/bot/handlers/admin/sync_admin.py b/bot/handlers/admin/sync_admin.py index 19eacfd..5de5b62 100644 --- a/bot/handlers/admin/sync_admin.py +++ b/bot/handlers/admin/sync_admin.py @@ -1,3 +1,4 @@ +import asyncio import logging from datetime import datetime, timezone from typing import Optional, Union @@ -16,6 +17,11 @@ from db.models import Subscription router = Router(name="admin_sync_router") +# Single-flight guard: panel sync runs concurrently with the bot, but only one +# sync at a time. Overlapping callers (startup, /sync, admin API) return early +# instead of queueing behind the running sync. +_sync_lock = asyncio.Lock() + def _normalize_panel_email(value: Optional[str]) -> Optional[str]: email = (value or "").strip().lower() @@ -122,6 +128,31 @@ async def perform_sync( session: AsyncSession, settings: Settings, i18n_instance: JsonI18n, +) -> dict: + """Single-flight entry point — skips when another sync is already running.""" + if _sync_lock.locked(): + logging.info("perform_sync: skipped because another sync is already in progress") + return { + "status": "skipped", + "details": "Another sync run is already in progress.", + "errors": [], + "users_processed": 0, + "subs_synced": 0, + } + async with _sync_lock: + return await _perform_sync_impl( + panel_service=panel_service, + session=session, + settings=settings, + i18n_instance=i18n_instance, + ) + + +async def _perform_sync_impl( + panel_service: PanelApiService, + session: AsyncSession, + settings: Settings, + i18n_instance: JsonI18n, ) -> dict: """ Perform panel synchronization and return results diff --git a/bot/main_bot.py b/bot/main_bot.py index 9c802ca..b8b9647 100644 --- a/bot/main_bot.py +++ b/bot/main_bot.py @@ -130,31 +130,54 @@ async def on_startup_configured(dispatcher: Dispatcher): except Exception: logging.exception("STARTUP: Failed to initialize message queue manager.") - # Automatic sync on startup + # Automatic sync on startup — runs in background so the dispatcher can + # start serving Telegram webhooks immediately even if the panel is slow. + # perform_sync is single-flight, so concurrent admin-triggered runs will + # be skipped while this one is in progress. try: - logging.info("STARTUP: Running automatic panel sync...") + logging.info("STARTUP: Scheduling automatic panel sync in background...") + asyncio.create_task( + _background_startup_sync( + panel_service=panel_service, + session_factory=async_session_factory, + settings=settings, + i18n_instance=i18n_instance, + ), + name="StartupPanelSync", + ) + except Exception: + logging.exception("STARTUP: Failed to schedule automatic sync.") - async with async_session_factory() as session: + logging.info("STARTUP: Bot on_startup_configured completed.") + + +async def _background_startup_sync( + *, + panel_service: PanelApiService, + session_factory: sessionmaker, + settings: Settings, + i18n_instance: JsonI18n, +) -> None: + try: + async with session_factory() as session: sync_result = await perform_sync( panel_service=panel_service, session=session, settings=settings, i18n_instance=i18n_instance, ) - - if sync_result.get("status") == "completed": - logging.info( - f"STARTUP: Automatic sync completed successfully. Details: {sync_result.get('details', 'N/A')}" # noqa: E501 - ) + status = sync_result.get("status") + details = sync_result.get("details", "N/A") + if status == "completed": + logging.info(f"STARTUP: Background sync completed successfully. Details: {details}") + elif status == "skipped": + logging.info(f"STARTUP: Background sync skipped: {details}") else: logging.warning( - f"STARTUP: Automatic sync completed with issues. Status: {sync_result.get('status', 'unknown')}" # noqa: E501 + f"STARTUP: Background sync finished with status '{status}'. Details: {details}" ) - except Exception: - logging.exception("STARTUP: Failed to run automatic sync.") - - logging.info("STARTUP: Bot on_startup_configured completed.") + logging.exception("STARTUP: Background sync failed.") async def on_shutdown_configured(dispatcher: Dispatcher): diff --git a/bot/services/tariff_worker.py b/bot/services/tariff_worker.py index 7132252..c0e2351 100644 --- a/bot/services/tariff_worker.py +++ b/bot/services/tariff_worker.py @@ -23,6 +23,11 @@ PREMIUM_WARNING_LEVEL_OFFSET = 1000 # Single warning per premium billing period when usage reached or exceeded the quota. PREMIUM_WARNING_DEPLETED_LEVEL = PREMIUM_WARNING_LEVEL_OFFSET + 100 +# Process active subscriptions in chunks and prefetch panel data concurrently +# to avoid an N+1 serial chain to the Remnawave panel each tick. +TARIFF_WORKER_BATCH_SIZE = 50 +TARIFF_WORKER_PANEL_CONCURRENCY = 10 + class TariffTrafficWorker: def __init__( @@ -117,48 +122,71 @@ class TariffTrafficWorker: Subscription.tariff_key.is_not(None), ) ) - for sub in result.scalars().all(): - try: - tariff = self.settings.tariffs_config.require(sub.tariff_key) - except Exception: - continue - panel_data = ( - await self.panel_service.get_user_by_uuid(sub.panel_user_uuid, log_response=False) - or {} - ) - used, limit, panel_strategy = self.subscription_service._extract_panel_traffic_details( - panel_data - ) - panel_status = str(panel_data.get("status") or "").upper() - panel_username = panel_data.get("username") if isinstance(panel_data, dict) else None - if used is not None and used != sub.traffic_used_bytes: - sub.traffic_used_bytes = used - if limit is not None and limit != sub.traffic_limit_bytes: - sub.traffic_limit_bytes = limit - if panel_status and panel_status != (sub.status_from_panel or "").upper(): - sub.status_from_panel = panel_status + subs = list(result.scalars().all()) + if not subs: + return - if tariff.billing_model == "period": - await self._ensure_period_reset_strategy(sub, tariff, limit, panel_strategy) - await self._maybe_warn_or_throttle( - session, - sub, - tariff, - used, - limit, - warning_period_start=warning_period_start - if tariff.billing_model == "period" - else None, - ) + semaphore = asyncio.Semaphore(TARIFF_WORKER_PANEL_CONCURRENCY) - await self._sync_premium_squad_limit( - session, - sub, - tariff, - now, - panel_username=panel_username, - panel_user_dict=panel_data, - ) + async def _fetch_panel(sub: Subscription) -> dict: + async with semaphore: + try: + data = await self.panel_service.get_user_by_uuid( + sub.panel_user_uuid, log_response=False + ) + except Exception: + logging.exception( + "TariffTrafficWorker: failed to fetch panel user %s", + sub.panel_user_uuid, + ) + return {} + return data or {} + + for chunk_start in range(0, len(subs), TARIFF_WORKER_BATCH_SIZE): + chunk = subs[chunk_start : chunk_start + TARIFF_WORKER_BATCH_SIZE] + panel_payloads = await asyncio.gather(*(_fetch_panel(s) for s in chunk)) + for sub, panel_data in zip(chunk, panel_payloads): + try: + tariff = self.settings.tariffs_config.require(sub.tariff_key) + except Exception: + continue + ( + used, + limit, + panel_strategy, + ) = self.subscription_service._extract_panel_traffic_details(panel_data) + panel_status = str(panel_data.get("status") or "").upper() + panel_username = ( + panel_data.get("username") if isinstance(panel_data, dict) else None + ) + if used is not None and used != sub.traffic_used_bytes: + sub.traffic_used_bytes = used + if limit is not None and limit != sub.traffic_limit_bytes: + sub.traffic_limit_bytes = limit + if panel_status and panel_status != (sub.status_from_panel or "").upper(): + sub.status_from_panel = panel_status + + if tariff.billing_model == "period": + await self._ensure_period_reset_strategy(sub, tariff, limit, panel_strategy) + await self._maybe_warn_or_throttle( + session, + sub, + tariff, + used, + limit, + warning_period_start=warning_period_start + if tariff.billing_model == "period" + else None, + ) + + await self._sync_premium_squad_limit( + session, + sub, + tariff, + now, + panel_username=panel_username, + panel_user_dict=panel_data, + ) async def _ensure_period_reset_strategy( self,