fix: устранение race condition при покупке устройств через re-lock после коммита

subtract_user_balance() делает внутренний коммит, что освобождает
SELECT FOR UPDATE блокировки. Добавлен паттерн re-lock + re-validate +
refund после вызова subtract_user_balance во всех 8 путях мутации
device_limit:

- cabinet: /devices, /devices/purchase, /devices/reduce
- miniapp: /subscription/devices
- bot handlers: execute_change_devices, confirm_add_devices
- auto-purchase: _auto_add_devices
- CRUD: add_subscription_devices

Также добавлен populate_existing=True ко всем SELECT FOR UPDATE запросам
для корректного обновления SQLAlchemy identity map.

app/cabinet/routes/subscription.py

@@ -1000,7 +1000,10 @@ async def purchase_devices_legacy(
 
     # Lock subscription row to prevent concurrent device purchases exceeding the limit
     result = await db.execute(
-        select(Subscription).where(Subscription.user_id == user.id).with_for_update()
+        select(Subscription)
+        .where(Subscription.user_id == user.id)
+        .with_for_update()
+        .execution_options(populate_existing=True)
     )
     subscription = result.scalar_one_or_none()
 
@@ -1092,9 +1095,33 @@ async def purchase_devices_legacy(
             detail='Insufficient funds',
         )
 
-    # Add devices
-    subscription.device_limit = new_devices
+    # Re-lock subscription after subtract_user_balance committed (which released all locks).
+    # Re-validate max device limit to prevent concurrent purchases exceeding the limit.
+    relock_result = await db.execute(
+        select(Subscription)
+        .where(Subscription.id == subscription.id)
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
+    subscription = relock_result.scalar_one()
 
+    actual_current = subscription.device_limit or 1
+    actual_new = actual_current + request.devices
+    if max_devices > 0 and actual_new > max_devices:
+        # Concurrent purchase already exceeded limit — refund balance
+        user_refund = await db.execute(
+            select(User).where(User.id == user.id).with_for_update().execution_options(populate_existing=True)
+        )
+        refund_user = user_refund.scalar_one()
+        refund_user.balance_kopeks += total_price
+        await db.commit()
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=f'Maximum device limit is {max_devices}. Balance refunded.',
+        )
+
+    # Add devices (under lock)
+    subscription.device_limit = actual_new
     await db.commit()
     await db.refresh(user)
 
@@ -1111,10 +1138,10 @@ async def purchase_devices_legacy(
                 await notification_service.send_subscription_update_notification(
                     db=db,
                     user=user,
-                    subscription=user.subscription,
+                    subscription=subscription,
                     update_type='devices',
                     old_value=current_devices,
-                    new_value=new_devices,
+                    new_value=actual_new,
                     price_paid=total_price,
                 )
             finally:
@@ -1125,7 +1152,7 @@ async def purchase_devices_legacy(
     response = {
         'message': 'Devices added successfully',
         'devices_added': request.devices,
-        'new_device_limit': new_devices,
+        'new_device_limit': actual_new,
         'amount_paid_kopeks': total_price,
     }
 
@@ -2298,7 +2325,10 @@ async def purchase_devices(
     try:
         # Lock subscription row to prevent concurrent device purchases exceeding the limit
         result = await db.execute(
-            select(Subscription).where(Subscription.user_id == user.id).with_for_update()
+            select(Subscription)
+            .where(Subscription.user_id == user.id)
+            .with_for_update()
+            .execution_options(populate_existing=True)
         )
         subscription = result.scalar_one_or_none()
 
@@ -2430,8 +2460,33 @@ async def purchase_devices(
                 detail='Insufficient funds',
             )
 
-        # Increase device limit
-        subscription.device_limit += request.devices
+        # Re-lock subscription after subtract_user_balance committed (which released all locks).
+        # Re-validate max device limit to prevent concurrent purchases exceeding the limit.
+        relock_result = await db.execute(
+            select(Subscription)
+            .where(Subscription.id == subscription.id)
+            .with_for_update()
+            .execution_options(populate_existing=True)
+        )
+        subscription = relock_result.scalar_one()
+
+        actual_current = subscription.device_limit or 1
+        actual_new = actual_current + request.devices
+        if max_device_limit and actual_new > max_device_limit:
+            # Concurrent purchase already exceeded limit — refund balance
+            user_refund = await db.execute(
+                select(User).where(User.id == user.id).with_for_update().execution_options(populate_existing=True)
+            )
+            refund_user = user_refund.scalar_one()
+            refund_user.balance_kopeks += price_kopeks
+            await db.commit()
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT,
+                detail=f'Максимальное количество устройств: {max_device_limit}. Баланс возвращён.',
+            )
+
+        # Increase device limit (under lock)
+        subscription.device_limit = actual_new
         await db.commit()
         await db.refresh(subscription)
 
@@ -3720,16 +3775,21 @@ async def reduce_devices(
             detail='Invalid new_device_limit',
         )
 
-    await db.refresh(user, ['subscription'])
+    # Lock subscription to prevent concurrent device modifications
+    result = await db.execute(
+        select(Subscription)
+        .where(Subscription.user_id == user.id)
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
+    subscription = result.scalar_one_or_none()
 
-    if not user.subscription:
+    if not subscription:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail='No subscription found',
         )
 
-    subscription = user.subscription
-
     if subscription.is_trial:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,

app/database/crud/subscription.py

@@ -613,7 +613,29 @@ async def add_subscription_traffic(db: AsyncSession, subscription: Subscription,
 
 
 async def add_subscription_devices(db: AsyncSession, subscription: Subscription, devices: int) -> Subscription:
-    subscription.device_limit += devices
+    # Lock subscription to prevent concurrent modifications
+    locked_result = await db.execute(
+        select(Subscription)
+        .where(Subscription.id == subscription.id)
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
+    subscription = locked_result.scalar_one()
+
+    # Check max device limit
+    max_devices = settings.MAX_DEVICES_LIMIT
+    new_limit = (subscription.device_limit or 1) + devices
+    if max_devices > 0 and new_limit > max_devices:
+        logger.warning(
+            '📱 Попытка превысить лимит устройств',
+            user_id=subscription.user_id,
+            current=subscription.device_limit,
+            requested=devices,
+            max_devices=max_devices,
+        )
+        new_limit = max_devices
+
+    subscription.device_limit = new_limit
     subscription.updated_at = datetime.now(UTC)
 
     await db.commit()

app/database/crud/user.py

@@ -515,7 +515,9 @@ async def subtract_user_balance(
     logger.info('📝 Описание', description=description)
 
     # Lock the user row to prevent concurrent balance race conditions
-    locked_result = await db.execute(select(User).where(User.id == user.id).with_for_update())
+    locked_result = await db.execute(
+        select(User).where(User.id == user.id).with_for_update().execution_options(populate_existing=True)
+    )
     user = locked_result.scalar_one()
 
     log_context: dict[str, object] | None = None

app/handlers/subscription/devices.py

@@ -2,13 +2,13 @@ import html as html_mod
 from datetime import UTC, datetime
 
 from aiogram import types
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
-from app.database.crud.subscription import add_subscription_devices
 from app.database.crud.transaction import create_transaction
 from app.database.crud.user import subtract_user_balance
-from app.database.models import TransactionType, User
+from app.database.models import Subscription, TransactionType, User
 from app.keyboards.inline import (
     get_app_selection_keyboard,
     get_back_keyboard,
@@ -554,6 +554,51 @@ async def execute_change_devices(callback: types.CallbackQuery, db_user: User, d
                 description=f'Изменение устройств с {current_devices} до {new_devices_count} на {charged_months} мес',
             )
 
+        # Re-lock subscription after subtract_user_balance committed (released all locks)
+        relock_result = await db.execute(
+            select(Subscription)
+            .where(Subscription.id == subscription.id)
+            .with_for_update()
+            .execution_options(populate_existing=True)
+        )
+        subscription = relock_result.scalar_one()
+
+        # Re-validate: prevent double-charge and max-limit violation
+        if new_devices_count > current_devices:
+            max_devices = settings.MAX_DEVICES_LIMIT
+            if max_devices > 0 and new_devices_count > max_devices:
+                if price > 0:
+                    user_refund = await db.execute(
+                        select(User)
+                        .where(User.id == db_user.id)
+                        .with_for_update()
+                        .execution_options(populate_existing=True)
+                    )
+                    refund_user = user_refund.scalar_one()
+                    refund_user.balance_kopeks += price
+                    await db.commit()
+                await callback.answer(
+                    f'⚠️ Лимит устройств ({max_devices}) превышен. Баланс возвращён.',
+                    show_alert=True,
+                )
+                return
+            # Check if concurrent request already applied the same change
+            if price > 0 and subscription.device_limit >= new_devices_count:
+                user_refund = await db.execute(
+                    select(User)
+                    .where(User.id == db_user.id)
+                    .with_for_update()
+                    .execution_options(populate_existing=True)
+                )
+                refund_user = user_refund.scalar_one()
+                refund_user.balance_kopeks += price
+                await db.commit()
+                await callback.answer(
+                    '⚠️ Изменение уже применено. Баланс возвращён.',
+                    show_alert=True,
+                )
+                return
+
         subscription.device_limit = new_devices_count
         subscription.updated_at = datetime.now(UTC)
 
@@ -1196,7 +1241,36 @@ async def confirm_add_devices(callback: types.CallbackQuery, db_user: User, db:
             await callback.answer('⚠️ Ошибка списания средств', show_alert=True)
             return
 
-        await add_subscription_devices(db, subscription, devices_count)
+        # Re-lock subscription after subtract_user_balance committed (released all locks)
+        relock_result = await db.execute(
+            select(Subscription)
+            .where(Subscription.id == subscription.id)
+            .with_for_update()
+            .execution_options(populate_existing=True)
+        )
+        subscription = relock_result.scalar_one()
+
+        # Re-validate max device limit after re-lock
+        actual_current = subscription.device_limit or 1
+        actual_new = actual_current + devices_count
+        max_devices = settings.MAX_DEVICES_LIMIT
+        if max_devices > 0 and actual_new > max_devices:
+            # Concurrent purchase exceeded limit — refund
+            user_refund = await db.execute(
+                select(User).where(User.id == db_user.id).with_for_update().execution_options(populate_existing=True)
+            )
+            refund_user = user_refund.scalar_one()
+            refund_user.balance_kopeks += price
+            await db.commit()
+            await callback.answer(
+                f'⚠️ Лимит устройств ({max_devices}) превышен. Баланс возвращён.',
+                show_alert=True,
+            )
+            return
+
+        subscription.device_limit = actual_new
+        subscription.updated_at = datetime.now(UTC)
+        await db.commit()
 
         subscription_service = SubscriptionService()
         await subscription_service.update_remnawave_user(db, subscription)

app/services/subscription_auto_purchase_service.py

@@ -8,6 +8,7 @@ from datetime import UTC, datetime, timedelta
 import structlog
 from aiogram import Bot
 from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
@@ -1181,7 +1182,6 @@ async def _auto_add_devices(
     """Auto-purchase devices from saved cart after balance topup."""
     from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
 
-    from app.database.crud.subscription import get_subscription_by_user_id
     from app.database.crud.user import subtract_user_balance
     from app.database.models import PaymentMethod
 
@@ -1207,8 +1207,14 @@ async def _auto_add_devices(
         )
         return False
 
-    # Проверяем подписку
-    subscription = await get_subscription_by_user_id(db, user.id)
+    # Проверяем подписку (with lock to prevent concurrent device modifications)
+    locked_result = await db.execute(
+        select(Subscription)
+        .where(Subscription.user_id == user.id)
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
+    subscription = locked_result.scalar_one_or_none()
     if not subscription:
         logger.warning('🔁 Автопокупка устройств: у пользователя нет подписки', format_user_id=_format_user_id(user))
         await user_cart_service.delete_user_cart(user.id)
@@ -1223,6 +1229,21 @@ async def _auto_add_devices(
         await user_cart_service.delete_user_cart(user.id)
         return False
 
+    # Check max device limit before charging
+    old_device_limit = subscription.device_limit or 1
+    new_device_limit = old_device_limit + devices_to_add
+    max_devices = settings.MAX_DEVICES_LIMIT
+    if max_devices > 0 and new_device_limit > max_devices:
+        logger.warning(
+            '🔁 Автопокупка устройств: превышен лимит устройств',
+            format_user_id=_format_user_id(user),
+            current=old_device_limit,
+            requested=new_device_limit,
+            max_devices=max_devices,
+        )
+        await user_cart_service.delete_user_cart(user.id)
+        return False
+
     # Списываем баланс
     description = f'Покупка {devices_to_add} доп. устройств'
     try:
@@ -1248,9 +1269,35 @@ async def _auto_add_devices(
         )
         return False
 
-    # Добавляем устройства
+    # Re-lock subscription after subtract_user_balance committed (released locks)
+    relock_result = await db.execute(
+        select(Subscription)
+        .where(Subscription.id == subscription.id)
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
+    subscription = relock_result.scalar_one()
+
     old_device_limit = subscription.device_limit or 1
-    subscription.device_limit = old_device_limit + devices_to_add
+    new_device_limit = old_device_limit + devices_to_add
+
+    if max_devices > 0 and new_device_limit > max_devices:
+        # Concurrent modification exceeded limit — refund
+        user_refund = await db.execute(
+            select(User).where(User.id == user.id).with_for_update().execution_options(populate_existing=True)
+        )
+        refund_user = user_refund.scalar_one()
+        refund_user.balance_kopeks += price_kopeks
+        await db.commit()
+        logger.warning(
+            '🔁 Автопокупка устройств: лимит превышен после оплаты, баланс возвращён',
+            format_user_id=_format_user_id(user),
+        )
+        await user_cart_service.delete_user_cart(user.id)
+        return False
+
+    # Добавляем устройства (under lock)
+    subscription.device_limit = new_device_limit
 
     try:
         await db.commit()

app/webapi/routes/miniapp.py

@@ -6115,7 +6115,10 @@ async def update_subscription_devices_endpoint(
 
     # Re-read subscription under row lock to prevent concurrent device purchases exceeding limit
     locked_result = await db.execute(
-        select(Subscription).where(Subscription.id == subscription.id).with_for_update()
+        select(Subscription)
+        .where(Subscription.id == subscription.id)
+        .with_for_update()
+        .execution_options(populate_existing=True)
     )
     subscription = locked_result.scalar_one()
 
@@ -6192,6 +6195,45 @@ async def update_subscription_devices_endpoint(
             description=f'{description} на {charged_months or get_remaining_months(subscription.end_date)} мес',
         )
 
+    if price_to_charge > 0:
+        # Re-lock subscription after subtract_user_balance committed (which released all locks).
+        # Re-validate to prevent concurrent device purchases from exceeding the limit or double-charging.
+        relock_result = await db.execute(
+            select(Subscription)
+            .where(Subscription.id == subscription.id)
+            .with_for_update()
+            .execution_options(populate_existing=True)
+        )
+        subscription = relock_result.scalar_one()
+
+        actual_current = subscription.device_limit or 1
+        actual_delta = new_devices - actual_current
+        max_devices_limit = settings.MAX_DEVICES_LIMIT
+
+        if actual_delta <= 0 or (max_devices_limit > 0 and new_devices > max_devices_limit):
+            # Concurrent request already applied the change or pushed limit beyond max — refund
+            user_refund = await db.execute(
+                select(User).where(User.id == user.id).with_for_update().execution_options(populate_existing=True)
+            )
+            refund_user = user_refund.scalar_one()
+            refund_user.balance_kopeks += price_to_charge
+            await db.commit()
+            if actual_delta <= 0:
+                raise HTTPException(
+                    status.HTTP_409_CONFLICT,
+                    detail={
+                        'code': 'already_applied',
+                        'message': 'Изменение уже применено параллельным запросом. Баланс возвращён.',
+                    },
+                )
+            raise HTTPException(
+                status.HTTP_409_CONFLICT,
+                detail={
+                    'code': 'devices_limit_exceeded',
+                    'message': f'Превышен максимальный лимит устройств ({max_devices_limit}). Баланс возвращён.',
+                },
+            )
+
     subscription.device_limit = new_devices
     subscription.updated_at = datetime.now(UTC)
     await db.commit()