From 5001185bf83863d6fd57a58e2d02ef604e8a99dd Mon Sep 17 00:00:00 2001 From: 3252a8 <3252a8@proton.me> Date: Wed, 20 May 2026 23:18:03 +0300 Subject: [PATCH] refactor: improve panel sync performance --- .env.example | 3 + backend/bot/handlers/admin/sync_admin.py | 96 ++++++++++++++++++----- backend/bot/services/panel_api_service.py | 46 +++++++++-- backend/config/settings.py | 3 + backend/db/dal/user_dal.py | 4 + backend/db/migrator.py | 14 ++++ backend/db/models.py | 1 + scripts/perf_benchmarks.py | 9 ++- tests/test_admin_sync_performance.py | 53 +++++++++++++ tests/test_panel_api_service_logging.py | 19 +++++ tests/test_support_migration.py | 10 ++- 11 files changed, 230 insertions(+), 28 deletions(-) diff --git a/.env.example b/.env.example index 82465e3..87075f1 100644 --- a/.env.example +++ b/.env.example @@ -20,8 +20,11 @@ WEBAPP_DEVICES_CACHE_TTL_SECONDS=5 # PANEL_USER_CACHE_TTL_SECONDS=5 # Short TTL for Remnawave /users/{uuid} cache PANEL_DEVICES_CACHE_TTL_SECONDS=5 # Short TTL for Remnawave user devices cache PANEL_ALL_USERS_CACHE_TTL_SECONDS=5 # Short TTL for concurrent Remnawave full user scans +PANEL_ALL_USERS_PAGE_SIZE=1000 # Remnawave /users page size with fallback to 100 ADMIN_PANEL_STATS_CACHE_TTL_SECONDS=15 # Short TTL for admin panel stats fetched from Remnawave PROFILE_SYNC_CACHE_TTL_SECONDS=900 # Minimum seconds between Telegram profile sync checks per user +PANEL_SYNC_LIFETIME_TRAFFIC_MIN_INTERVAL_SECONDS=3600 # Min seconds between local lifetime traffic writes per user +PANEL_SYNC_LIFETIME_TRAFFIC_MIN_DELTA_BYTES=104857600 # Write lifetime traffic sooner when delta is at least this many bytes WEBAPP_RATE_LIMIT_TTL_SECONDS=60 # Redis rate-limit window WEBAPP_RATE_LIMIT_MAX_REQUESTS=30 # Requests per window/action/user/IP WEBHOOK_QUEUE_NAME=webhook-events # Redis queue for heavy webhook processing diff --git a/backend/bot/handlers/admin/sync_admin.py b/backend/bot/handlers/admin/sync_admin.py index 25f6f27..fa74835 100644 --- a/backend/bot/handlers/admin/sync_admin.py +++ b/backend/bot/handlers/admin/sync_admin.py @@ -55,6 +55,56 @@ def _datetime_matches(current: Optional[datetime], desired: datetime) -> bool: return abs(delta.total_seconds()) < 1 +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _should_update_lifetime_used_traffic( + existing_user, + lifetime_used: int, + *, + now: datetime, + settings: Settings, + is_duplicate_panel_identity: bool = False, +) -> bool: + if is_duplicate_panel_identity: + return False + + current_value = existing_user.lifetime_used_traffic_bytes + if current_value == lifetime_used: + return False + if current_value is None: + return True + + try: + min_delta_bytes = max( + 0, + int(getattr(settings, "PANEL_SYNC_LIFETIME_TRAFFIC_MIN_DELTA_BYTES", 0) or 0), + ) + except (TypeError, ValueError): + min_delta_bytes = 0 + if min_delta_bytes and abs(int(lifetime_used) - int(current_value or 0)) >= min_delta_bytes: + return True + + try: + min_interval_seconds = max( + 0, + int(getattr(settings, "PANEL_SYNC_LIFETIME_TRAFFIC_MIN_INTERVAL_SECONDS", 0) or 0), + ) + except (TypeError, ValueError): + min_interval_seconds = 0 + if min_interval_seconds <= 0: + return True + + last_synced_at = getattr(existing_user, "lifetime_used_traffic_synced_at", None) + if not last_synced_at: + return True + + return (_as_utc(now) - _as_utc(last_synced_at)).total_seconds() >= min_interval_seconds + + def _subscription_update_delta( subscription: Subscription, desired: dict[str, Any] ) -> dict[str, Any]: @@ -469,6 +519,7 @@ async def _perform_sync_impl( # Get the actual user_id for subscription operations actual_user_id = existing_user.user_id + is_duplicate_panel_identity = False # Update panel UUID if different if existing_user.panel_user_uuid != panel_uuid: @@ -480,6 +531,7 @@ async def _perform_sync_impl( in panel_uuids_by_telegram_id.get(telegram_id_from_panel, set()) ) if linked_uuid_still_present: + is_duplicate_panel_identity = True logging.warning( "Sync: duplicate panel users share telegramId %s; keeping local panel UUID %s and skipping duplicate panel UUID %s.", # noqa: E501 telegram_id_from_panel, @@ -492,32 +544,40 @@ async def _perform_sync_impl( users_uuid_updated += 1 users_by_panel_uuid[panel_uuid] = existing_user logging.info(f"Updated panel UUID for user {actual_user_id}: {panel_uuid}") - existing_user, email_was_bound = await _bind_panel_email_to_user( - session, - existing_user=existing_user, - email_from_panel=email_from_panel, - panel_uuid=panel_uuid, - ) - if email_was_bound: - user_was_updated = True - if email_from_panel: - users_by_email[email_from_panel] = existing_user - if telegram_id_from_panel and existing_user.telegram_id != telegram_id_from_panel: - existing_user.telegram_id = telegram_id_from_panel - user_was_updated = True - users_by_telegram_id[telegram_id_from_panel] = existing_user + if not is_duplicate_panel_identity: + existing_user, email_was_bound = await _bind_panel_email_to_user( + session, + existing_user=existing_user, + email_from_panel=email_from_panel, + panel_uuid=panel_uuid, + ) + if email_was_bound: + user_was_updated = True + if email_from_panel: + users_by_email[email_from_panel] = existing_user + if ( + telegram_id_from_panel + and existing_user.telegram_id != telegram_id_from_panel + ): + existing_user.telegram_id = telegram_id_from_panel + user_was_updated = True + users_by_telegram_id[telegram_id_from_panel] = existing_user lifetime_used = _extract_lifetime_used_traffic_bytes(panel_user_dict) - if ( - lifetime_used is not None - and existing_user.lifetime_used_traffic_bytes != lifetime_used + if lifetime_used is not None and _should_update_lifetime_used_traffic( + existing_user, + lifetime_used, + now=datetime.now(timezone.utc), + settings=settings, + is_duplicate_panel_identity=is_duplicate_panel_identity, ): existing_user.lifetime_used_traffic_bytes = lifetime_used + existing_user.lifetime_used_traffic_synced_at = datetime.now(timezone.utc) user_was_updated = True # Ensure panel description contains Telegram fields try: - if panel_uuid and existing_user: + if panel_uuid and existing_user and not is_duplicate_panel_identity: description_text = "\n".join( line for line in [ diff --git a/backend/bot/services/panel_api_service.py b/backend/bot/services/panel_api_service.py index 0d6a8cf..0f12cc9 100644 --- a/backend/bot/services/panel_api_service.py +++ b/backend/bot/services/panel_api_service.py @@ -242,20 +242,54 @@ class PanelApiService: ) return {"error": True, "status_code": -4, "message": f"Unexpected error: {str(e)}"} + def _resolve_all_users_page_size(self, page_size: Optional[int] = None) -> int: + raw_value = ( + page_size + if page_size is not None + else getattr(self.settings, "PANEL_ALL_USERS_PAGE_SIZE", 1000) + ) + try: + value = int(raw_value or 1000) + except (TypeError, ValueError): + value = 1000 + return min(1000, max(1, value)) + async def get_all_panel_users( - self, page_size: int = 100, log_responses: bool = False + self, page_size: Optional[int] = None, log_responses: bool = False ) -> Optional[List[Dict[str, Any]]]: - if log_responses or page_size != 100 or self._all_users_cache.ttl_seconds <= 0: + resolved_page_size = self._resolve_all_users_page_size(page_size) + if log_responses or self._all_users_cache.ttl_seconds <= 0: return await self._get_all_panel_users_uncached( - page_size=page_size, log_responses=log_responses + page_size=resolved_page_size, log_responses=log_responses ) return await self._all_users_cache.get_or_load( - f"page_size:{page_size}", - lambda: self._get_all_panel_users_uncached(page_size=page_size, log_responses=False), + f"page_size:{resolved_page_size}", + lambda: self._get_all_panel_users_uncached( + page_size=resolved_page_size, log_responses=False + ), ) async def _get_all_panel_users_uncached( - self, page_size: int = 100, log_responses: bool = False + self, page_size: Optional[int] = None, log_responses: bool = False + ) -> Optional[List[Dict[str, Any]]]: + resolved_page_size = self._resolve_all_users_page_size(page_size) + users = await self._fetch_all_panel_users_pages( + page_size=resolved_page_size, + log_responses=log_responses, + ) + if users is None and resolved_page_size != 100: + logging.warning( + "Panel API users fetch failed with page size %s; retrying with page size 100.", + resolved_page_size, + ) + users = await self._fetch_all_panel_users_pages( + page_size=100, + log_responses=log_responses, + ) + return users + + async def _fetch_all_panel_users_pages( + self, page_size: int, log_responses: bool = False ) -> Optional[List[Dict[str, Any]]]: all_users = [] start_offset = 0 diff --git a/backend/config/settings.py b/backend/config/settings.py index b09596a..230b76d 100644 --- a/backend/config/settings.py +++ b/backend/config/settings.py @@ -99,8 +99,11 @@ class Settings(BaseSettings): PANEL_USER_CACHE_TTL_SECONDS: int = Field(default=5) PANEL_DEVICES_CACHE_TTL_SECONDS: int = Field(default=5) PANEL_ALL_USERS_CACHE_TTL_SECONDS: int = Field(default=5) + PANEL_ALL_USERS_PAGE_SIZE: int = Field(default=1000) ADMIN_PANEL_STATS_CACHE_TTL_SECONDS: int = Field(default=15) PROFILE_SYNC_CACHE_TTL_SECONDS: int = Field(default=900) + PANEL_SYNC_LIFETIME_TRAFFIC_MIN_INTERVAL_SECONDS: int = Field(default=3600) + PANEL_SYNC_LIFETIME_TRAFFIC_MIN_DELTA_BYTES: int = Field(default=104857600) WEBAPP_RATE_LIMIT_TTL_SECONDS: int = Field(default=60) WEBAPP_RATE_LIMIT_MAX_REQUESTS: int = Field(default=30) WEBHOOK_QUEUE_NAME: str = Field(default="webhook-events") diff --git a/backend/db/dal/user_dal.py b/backend/db/dal/user_dal.py index 1c1d76a..fab6a07 100644 --- a/backend/db/dal/user_dal.py +++ b/backend/db/dal/user_dal.py @@ -404,6 +404,10 @@ async def merge_users( target.lifetime_used_traffic_bytes = ( target.lifetime_used_traffic_bytes or 0 ) + source.lifetime_used_traffic_bytes + source_synced_at = getattr(source, "lifetime_used_traffic_synced_at", None) + target_synced_at = getattr(target, "lifetime_used_traffic_synced_at", None) + if source_synced_at and (not target_synced_at or source_synced_at > target_synced_at): + target.lifetime_used_traffic_synced_at = source_synced_at if not target.referred_by_id and source.referred_by_id != target_user_id: target.referred_by_id = source.referred_by_id if target.referred_by_id == source_user_id: diff --git a/backend/db/migrator.py b/backend/db/migrator.py index 9e27890..3ef6a0c 100644 --- a/backend/db/migrator.py +++ b/backend/db/migrator.py @@ -873,6 +873,15 @@ def _migration_0025_add_support_notification_timestamps(connection: Connection) ) +def _migration_0026_add_lifetime_traffic_synced_at(connection: Connection) -> None: + inspector = inspect(connection) + columns: Set[str] = {col["name"] for col in inspector.get_columns("users")} + if "lifetime_used_traffic_synced_at" not in columns: + connection.execute( + text("ALTER TABLE users ADD COLUMN lifetime_used_traffic_synced_at TIMESTAMPTZ") + ) + + MIGRATIONS: List[Migration] = [ Migration( id="0001_add_channel_subscription_fields", @@ -1010,6 +1019,11 @@ MIGRATIONS: List[Migration] = [ description="Track support ticket admin notification cooldown timestamps", upgrade=_migration_0025_add_support_notification_timestamps, ), + Migration( + id="0026_add_lifetime_traffic_synced_at", + description="Track when lifetime traffic usage was last synced from panel", + upgrade=_migration_0026_add_lifetime_traffic_synced_at, + ), ] diff --git a/backend/db/models.py b/backend/db/models.py index 51476e3..bb8b78b 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -42,6 +42,7 @@ class User(Base): referral_code = Column(String(16), nullable=True, unique=True, index=True) referred_by_id = Column(BigInteger, ForeignKey("users.user_id"), nullable=True) lifetime_used_traffic_bytes = Column(BigInteger, nullable=True) + lifetime_used_traffic_synced_at = Column(DateTime(timezone=True), nullable=True) channel_subscription_verified = Column(Boolean, nullable=True) channel_subscription_checked_at = Column(DateTime(timezone=True), nullable=True) channel_subscription_verified_for = Column(BigInteger, nullable=True) diff --git a/scripts/perf_benchmarks.py b/scripts/perf_benchmarks.py index cee0a9c..f8947ce 100644 --- a/scripts/perf_benchmarks.py +++ b/scripts/perf_benchmarks.py @@ -32,7 +32,7 @@ from bot.utils.ttl_cache import AsyncTTLCache # noqa: E402 DEFAULT_USER_SIZES = (200, 500, 1000, 5000, 10000) -def estimated_panel_user_pages(users: int, page_size: int = 100) -> int: +def estimated_panel_user_pages(users: int, page_size: int = 1000) -> int: if users <= 0: return 1 # get_all_panel_users stops on a short/empty page, so exact page multiples @@ -113,7 +113,7 @@ async def bench_panel_user_prefetch(users: int) -> dict: "service_calls": panel.calls, "matched": len(by_uuid or {}), "legacy_user_get_calls": users, - "estimated_bulk_http_pages_at_100": estimated_panel_user_pages(users), + "estimated_bulk_http_pages": estimated_panel_user_pages(users), } @@ -171,6 +171,7 @@ async def bench_panel_user_cache(users: int) -> dict: PANEL_USER_CACHE_TTL_SECONDS=60, PANEL_DEVICES_CACHE_TTL_SECONDS=60, PANEL_ALL_USERS_CACHE_TTL_SECONDS=60, + PANEL_ALL_USERS_PAGE_SIZE=1000, REDIS_URL=None, REDIS_KEY_PREFIX="bench", ) @@ -202,6 +203,7 @@ async def bench_panel_all_users_cache(users: int) -> dict: PANEL_USER_CACHE_TTL_SECONDS=60, PANEL_DEVICES_CACHE_TTL_SECONDS=60, PANEL_ALL_USERS_CACHE_TTL_SECONDS=60, + PANEL_ALL_USERS_PAGE_SIZE=1000, REDIS_URL=None, REDIS_KEY_PREFIX="bench", ) @@ -243,6 +245,7 @@ async def bench_panel_devices_cache(users: int) -> dict: PANEL_USER_CACHE_TTL_SECONDS=60, PANEL_DEVICES_CACHE_TTL_SECONDS=60, PANEL_ALL_USERS_CACHE_TTL_SECONDS=60, + PANEL_ALL_USERS_PAGE_SIZE=1000, REDIS_URL=None, REDIS_KEY_PREFIX="bench", ) @@ -444,7 +447,7 @@ def _print_table(results: dict[str, dict]) -> None: ) print( f"{users:>5} | " - f"{data['panel_user_bulk_prefetch']['estimated_bulk_http_pages_at_100']:>14} | " + f"{data['panel_user_bulk_prefetch']['estimated_bulk_http_pages']:>14} | " f"{data['premium_usage_1_node']['seconds']:>15.6f} | " f"{data['premium_usage_1_node']['panel_calls']:>19} | " f"{sync_optimized_reads:>17} | " diff --git a/tests/test_admin_sync_performance.py b/tests/test_admin_sync_performance.py index 65f5823..68a7ecd 100644 --- a/tests/test_admin_sync_performance.py +++ b/tests/test_admin_sync_performance.py @@ -1,8 +1,10 @@ from datetime import datetime, timedelta, timezone +from types import SimpleNamespace from bot.handlers.admin.sync_admin import ( _coerce_panel_telegram_id, _description_matches, + _should_update_lifetime_used_traffic, _subscription_update_delta, ) from db.models import Subscription @@ -69,3 +71,54 @@ def test_subscription_update_delta_returns_only_changed_fields(): "is_active": False, "status_from_panel": "EXPIRED", } + + +def test_lifetime_traffic_update_waits_for_time_window_for_small_delta(): + now = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc) + settings = SimpleNamespace( + PANEL_SYNC_LIFETIME_TRAFFIC_MIN_INTERVAL_SECONDS=3600, + PANEL_SYNC_LIFETIME_TRAFFIC_MIN_DELTA_BYTES=100 * 1024 * 1024, + ) + user = SimpleNamespace( + lifetime_used_traffic_bytes=10 * 1024 * 1024, + lifetime_used_traffic_synced_at=now - timedelta(minutes=15), + ) + + assert not _should_update_lifetime_used_traffic( + user, + 11 * 1024 * 1024, + now=now, + settings=settings, + ) + assert _should_update_lifetime_used_traffic( + user, + 11 * 1024 * 1024, + now=now + timedelta(hours=1), + settings=settings, + ) + + +def test_lifetime_traffic_update_allows_large_delta_and_skips_duplicate_panel_identity(): + now = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc) + settings = SimpleNamespace( + PANEL_SYNC_LIFETIME_TRAFFIC_MIN_INTERVAL_SECONDS=3600, + PANEL_SYNC_LIFETIME_TRAFFIC_MIN_DELTA_BYTES=100 * 1024 * 1024, + ) + user = SimpleNamespace( + lifetime_used_traffic_bytes=10 * 1024 * 1024, + lifetime_used_traffic_synced_at=now, + ) + + assert _should_update_lifetime_used_traffic( + user, + 200 * 1024 * 1024, + now=now, + settings=settings, + ) + assert not _should_update_lifetime_used_traffic( + user, + 0, + now=now + timedelta(hours=2), + settings=settings, + is_duplicate_panel_identity=True, + ) diff --git a/tests/test_panel_api_service_logging.py b/tests/test_panel_api_service_logging.py index fb00de7..e3efab2 100644 --- a/tests/test_panel_api_service_logging.py +++ b/tests/test_panel_api_service_logging.py @@ -106,6 +106,25 @@ class PanelApiServiceLoggingTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(get_calls, 2) + async def test_get_all_panel_users_falls_back_to_100_when_large_page_fails(self): + service = self._make_service() + requested_sizes = [] + + async def fake_request(method, endpoint, **kwargs): + params = kwargs.get("params") or {} + size = params.get("size") + requested_sizes.append(size) + if size == 1000: + return {"error": True, "status_code": 400} + return {"response": {"users": [{"uuid": "user-uuid"}]}} + + service._request = AsyncMock(side_effect=fake_request) + + users = await service.get_all_panel_users() + + self.assertEqual(users, [{"uuid": "user-uuid"}]) + self.assertEqual(requested_sizes, [1000, 100]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_support_migration.py b/tests/test_support_migration.py index 9a8bdaa..ec113a2 100644 --- a/tests/test_support_migration.py +++ b/tests/test_support_migration.py @@ -1,5 +1,5 @@ from db.migrator import MIGRATIONS -from db.models import SupportTicket, SupportTicketMessage +from db.models import SupportTicket, SupportTicketMessage, User def test_support_migration_is_registered_after_existing_revisions(): @@ -11,6 +11,10 @@ def test_support_migration_is_registered_after_existing_revisions(): assert ids.index("0025_add_support_notification_timestamps") > ids.index( "0024_add_support_tickets" ) + assert "0026_add_lifetime_traffic_synced_at" in ids + assert ids.index("0026_add_lifetime_traffic_synced_at") > ids.index( + "0025_add_support_notification_timestamps" + ) def test_support_models_expose_expected_tables(): @@ -21,3 +25,7 @@ def test_support_models_expose_expected_tables(): assert "ix_support_tickets_status_last_msg" in { index.name for index in SupportTicket.__table__.indexes } + + +def test_user_model_tracks_lifetime_traffic_sync_timestamp(): + assert "lifetime_used_traffic_synced_at" in User.__table__.columns