austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

refactor: parallelize tariff worker panel calls and unblock startup sync

Browse Source
This commit is contained in:
3252a8
2026-05-15 23:18:24 +03:00
parent 4d96a3e646
commit b72a23f1e1
3 changed files with 135 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot/handlers/admin/sync_admin.py
+31
View File
@@ -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
bot/main_bot.py
+36 -13
View File
@@ -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):
bot/services/tariff_worker.py
+68 -40
View File
@@ -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,