perf: bound NATS user-sync KV work and cut /sub plus usage-job CPU

app/db/crud/user.py

@@ -1,9 +1,10 @@
-from collections.abc import Sequence
+from collections.abc import AsyncIterator, Sequence
+from contextlib import asynccontextmanager
 from copy import deepcopy
 from datetime import UTC, datetime, timedelta
 from typing import Literal
 
-from sqlalchemy import and_, case, delete, desc, func, literal, not_, or_, select, update
+from sqlalchemy import and_, bindparam, case, delete, desc, func, literal, not_, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import joinedload, selectinload, with_expression
 from sqlalchemy.sql import Select
@@ -47,6 +48,7 @@
     UserSortOption,
 )
 from app.models.validators import MAX_ON_HOLD_EXPIRE_DURATION_SECONDS
+from app.subscription.sub_update_buffer import flush_user_sub_updates, queue_user_sub_update
 from config import user_cleanup_settings
 
 from .general import (
@@ -65,11 +67,21 @@
     tags_from_groups,
 )
 
-_USER_AGENT_MAX_LEN = UserSubscriptionUpdate.__table__.columns.user_agent.type.length or 512
-_SUBSCRIPTION_UPDATE_IP_MAX_LEN = UserSubscriptionUpdate.__table__.columns.ip.type.length or 64
 _ONLINE_USERS_WINDOW = timedelta(minutes=2)
 
 
+def _review_user_select_stmt(*, load_groups: bool = True) -> Select:
+    """Load relations needed for status review jobs without materializing usage logs."""
+    return _build_user_select_stmt(
+        load_admin=True,
+        load_admin_role=True,
+        load_next_plan=True,
+        load_usage_logs=False,
+        load_groups=load_groups,
+        load_lifetime_used_traffic=True,
+    )
+
+
 def _user_reset_traffic_subquery():
     return (
         select(func.coalesce(func.sum(UserUsageResetLogs.used_traffic_at_reset), 0))
@@ -592,19 +604,19 @@ async def remove_expired_users(
 
 
 async def get_active_to_expire_users(db: AsyncSession) -> list[User]:
-    stmt = _build_user_select_stmt().where(User.status == UserStatus.active).where(User.is_expired)
+    stmt = _review_user_select_stmt().where(User.status == UserStatus.active).where(User.is_expired)
 
     return list((await db.execute(stmt)).unique().scalars().all())
 
 
 async def get_active_to_limited_users(db: AsyncSession) -> list[User]:
-    stmt = _build_user_select_stmt().where(User.status == UserStatus.active).where(User.is_limited)
+    stmt = _review_user_select_stmt().where(User.status == UserStatus.active).where(User.is_limited)
 
     return list((await db.execute(stmt)).unique().scalars().all())
 
 
 async def get_on_hold_to_active_users(db: AsyncSession) -> list[User]:
-    stmt = _build_user_select_stmt().where(User.status == UserStatus.on_hold).where(User.become_online)
+    stmt = _review_user_select_stmt().where(User.status == UserStatus.on_hold).where(User.become_online)
 
     return list((await db.execute(stmt)).unique().scalars().all())
 
@@ -637,7 +649,7 @@ async def get_users_to_reset_data_usage(db: AsyncSession) -> list[User]:
     )
 
     stmt = (
-        _build_user_select_stmt()
+        _review_user_select_stmt()
         .outerjoin(last_reset_subq, User.id == last_reset_subq.c.user_id)
         .where(
             User.status.in_([UserStatus.active, UserStatus.limited]),
@@ -666,14 +678,13 @@ async def get_usage_percentage_reached_users(db: AsyncSession, percentage: int)
     )
 
     stmt = (
-        _build_user_select_stmt()
+        _review_user_select_stmt()
         .options(joinedload(User.notification_reminders))
         .where(User.status == UserStatus.active)
         .where(User.usage_percentage >= percentage)
         .where(not_(existing_reminder_subq))  # Only users without existing reminders
     )
 
-    # All relations (admin, next_plan, usage_logs, groups) eagerly loaded via _build_user_select_stmt
     return list((await db.execute(stmt)).unique().scalars().all())
 
 
@@ -694,15 +705,14 @@ async def get_days_left_reached_users(db: AsyncSession, days: int) -> list[User]
     )
 
     stmt = (
-        _build_user_select_stmt()
+        _review_user_select_stmt()
         .options(joinedload(User.notification_reminders))
         .where(User.status == UserStatus.active)
         .where(User.expire.isnot(None))
         .where(User.days_left == days)
         .where(not_(existing_reminder_subq))  # Only users without existing reminders
     )
 
-    # All relations (admin, next_plan, usage_logs, groups) eagerly loaded via _build_user_select_stmt
     return list((await db.execute(stmt)).unique().scalars().all())
 
 
@@ -1339,28 +1349,33 @@ async def bulk_revoke_user_sub(
     return users
 
 
+@asynccontextmanager
+async def _subscription_update_read_session(db: AsyncSession) -> AsyncIterator[AsyncSession]:
+    """Read committed updates without reusing the caller's pre-flush snapshot."""
+    await flush_user_sub_updates()
+    # MySQL/MariaDB REPEATABLE READ may already have a snapshot from the
+    # authorization/user lookup. A separate session sees the flushed rows
+    # without committing or rolling back any work owned by the caller.
+    async with AsyncSession(bind=db.bind, expire_on_commit=False) as read_db:
+        yield read_db
+
+
 async def user_sub_update(
     db: AsyncSession, user_id: int, user_agent: str, ip: str | None = None, hwid: str | None = None
 ) -> None:
     """
-    Updates the user's subscription details.
+    Queue a subscription-update row. The request session is unused; writes are
+    flushed in the background so the public /sub path stays read-mostly.
 
     Args:
-        db (AsyncSession): Database session.
+        db (AsyncSession): Database session (unused; kept for call-site compatibility).
         user_id (int): The user id whose subscription is to be updated.
         user_agent (str): The user agent string.
         ip (str | None): The client IP address.
         hwid (str | None): The hardware ID of the client.
     """
-    # Clamp to column length; some clients send very long strings (e.g. encoded configs) as User-Agent.
-    sanitized_user_agent = (user_agent or "")[:_USER_AGENT_MAX_LEN]
-    sanitized_ip = (ip or "")[:_SUBSCRIPTION_UPDATE_IP_MAX_LEN] or None
-    sanitized_hwid = (hwid or "")[:256] or None
-    agent = UserSubscriptionUpdate(
-        user_id=user_id, user_agent=sanitized_user_agent, ip=sanitized_ip, hwid=sanitized_hwid
-    )
-    db.add(agent)
-    await db.commit()
+    _ = db
+    await queue_user_sub_update(user_id, user_agent, ip=ip, hwid=hwid)
 
 
 async def get_users_sub_update_list(
@@ -1372,15 +1387,16 @@ async def get_users_sub_update_list(
         .order_by(desc(UserSubscriptionUpdate.created_at))
     )
 
-    result = await db.execute(select(func.count()).select_from(stmt.subquery()))
-    count = result.scalar() or 0
+    async with _subscription_update_read_session(db) as read_db:
+        result = await read_db.execute(select(func.count()).select_from(stmt.subquery()))
+        count = result.scalar() or 0
 
-    if offset:
-        stmt = stmt.offset(offset)
-    if limit:
-        stmt = stmt.limit(limit)
+        if offset:
+            stmt = stmt.offset(offset)
+        if limit:
+            stmt = stmt.limit(limit)
 
-    result = (await db.execute(stmt)).unique().scalars().all()
+        result = (await read_db.execute(stmt)).unique().scalars().all()
 
     return result, count
 
@@ -1424,7 +1440,8 @@ async def get_users_subscription_agent_counts(
         stmt = stmt.where(and_(*conditions))
     stmt = stmt.group_by(UserSubscriptionUpdate.user_agent)
 
-    result = await db.execute(stmt)
+    async with _subscription_update_read_session(db) as read_db:
+        result = await read_db.execute(stmt)
     return [(agent, count) for agent, count in result.all()]
 
 
@@ -1460,7 +1477,8 @@ async def get_users_subscription_agent_stats(
         .order_by(trunc_expr)
     )
 
-    result = await db.execute(stmt)
+    async with _subscription_update_read_session(db) as read_db:
+        result = await read_db.execute(stmt)
     dialect = db.bind.dialect.name
     rows = []
     for row in result.mappings():
@@ -1825,24 +1843,32 @@ async def start_users_expire(db: AsyncSession, users: list[User]) -> list[User]:
     Returns:
         list[User]: The updated users list.
     """
+    if not users:
+        return []
+
     now = datetime.now(UTC)
+    params = []
     for user in users:
         duration = _safe_on_hold_expire_duration(user.on_hold_expire_duration)
         expire_time = now + timedelta(seconds=duration) if duration is not None else None
         user.expire = expire_time
         user.on_hold_expire_duration = None
         user.on_hold_timeout = None
         user.status = UserStatus.active
-        stmt = (
-            update(User)
-            .where(User.id == user.id)
-            .values(expire=expire_time, on_hold_expire_duration=None, on_hold_timeout=None, status=UserStatus.active)
-        )
-        await db.execute(stmt)
-
+        params.append({"u_id": user.id, "u_expire": expire_time})
+
+    await db.execute(
+        update(User.__table__)
+        .where(User.__table__.c.id == bindparam("u_id"))
+        .values(
+            expire=bindparam("u_expire"),
+            on_hold_expire_duration=None,
+            on_hold_timeout=None,
+            status=UserStatus.active,
+        ),
+        params,
+    )
     await db.commit()
-    for user in users:
-        await refresh_and_load_user(db, user)
     return users
 
 

app/db/models.py

@@ -283,6 +283,21 @@ def last_traffic_reset_time(self):
 
     async def inbounds(self) -> list[str]:
         """Returns a flat list of all included inbound tags for enabled groups."""
+        loaded_groups = self.__dict__.get("groups")
+        if loaded_groups is not None:
+            inbound_tags: set[str] = set()
+            inbounds_loaded = True
+            for group in loaded_groups:
+                if "inbounds" not in group.__dict__:
+                    inbounds_loaded = False
+                    break
+                if group.is_disabled:
+                    continue
+                for inbound in group.__dict__.get("inbounds") or []:
+                    inbound_tags.add(inbound.tag)
+            if inbounds_loaded:
+                return list(inbound_tags)
+
         session = async_object_session(self)
         if session is not None:
             stmt = (
@@ -299,7 +314,7 @@ async def inbounds(self) -> list[str]:
 
         # Fallback for detached instances: use already-loaded attrs only.
         included_tags = set()
-        for group in self.__dict__.get("groups") or []:
+        for group in loaded_groups or []:
             if group.is_disabled:
                 continue
             for inbound in group.__dict__.get("inbounds") or []:

app/jobs/cleanup_node_sync.py

@@ -0,0 +1,42 @@
+"""Compact completed node-sync work on the scheduler leader only."""
+
+import asyncio
+
+from app import scheduler
+from app.nats import needs_shared_bridge_memory
+from app.nats.client import create_nats_client, get_jetstream_context
+from app.nats.kv_cleanup import compact_deleted_keys
+from app.nats.leader import is_job_leader
+from app.utils.logger import get_logger
+from config import nats_settings, runtime_settings
+
+logger = get_logger("jobs")
+
+
+async def cleanup_node_sync():
+    if not is_job_leader():
+        return
+    nc = await create_nats_client()
+    if nc is None:
+        return
+    try:
+        async with asyncio.timeout(240):
+            count = await compact_deleted_keys(await get_jetstream_context(nc), nats_settings.node_user_sync_kv_bucket)
+        if count:
+            logger.info("Compacted %s completed node-sync keys", count)
+    except TimeoutError:
+        logger.warning("node-sync key compaction timed out; retrying on the next interval")
+    finally:
+        await nc.close()
+
+
+if runtime_settings.role.runs_node and needs_shared_bridge_memory():
+    scheduler.add_job(
+        cleanup_node_sync,
+        "interval",
+        seconds=300,
+        max_instances=1,
+        coalesce=True,
+        id="cleanup_node_sync",
+        replace_existing=True,
+    )

app/jobs/cleanup_subscription_updates.py

@@ -3,6 +3,7 @@
 from app import scheduler
 from app.db import GetDB
 from app.db.models import UserSubscriptionUpdate
+from app.subscription.sub_update_buffer import flush_user_sub_updates
 from app.utils.logger import get_logger
 from config import job_settings, runtime_settings, subscription_env_settings
 
@@ -12,6 +13,8 @@
 async def cleanup_user_subscription_updates():
     """Clean up excess user subscription updates."""
 
+    await flush_user_sub_updates()
+
     async with GetDB() as db:
         # First query: Find users that have more than the limit
         users_with_excess = await db.execute(