fix: avoid async DB races in background bulk sync tasks

Author: x0sina

app/operation/admin.py

@@ -20,7 +20,8 @@
 from app.db.crud.bulk import activate_all_disabled_users, disable_all_active_users
 from app.db.crud.user import get_users, remove_users
 from app.models.admin import AdminCreate, AdminDetails, AdminModify, AdminSimple, AdminsResponse, AdminsSimpleResponse
-from app.node.sync import schedule_sync_task, sync_remove_users, sync_users, remove_user as sync_remove_user
+from app.node.sync import schedule_sync_task, sync_remove_users, sync_users, sync_proto_users, remove_user as sync_remove_user
+from app.node.user import serialize_users_for_node
 from app.models.stats import Period, UserUsageStatsList
 from app.operation import BaseOperation, OperatorType
 from app.operation.user import UserOperation
@@ -171,7 +172,8 @@ async def disable_all_active_users(self, db: AsyncSession, username: str, admin:
 
         users = await get_users(db, admin=db_admin)
         if self._is_non_blocking_sync_operator(self.operator_type):
-            schedule_sync_task(sync_users(users))
+            proto_users = await serialize_users_for_node(users)
+            schedule_sync_task(sync_proto_users(proto_users))
         else:
             await sync_users(users)
 
@@ -188,7 +190,8 @@ async def activate_all_disabled_users(self, db: AsyncSession, username: str, adm
 
         users = await get_users(db, admin=db_admin)
         if self._is_non_blocking_sync_operator(self.operator_type):
-            schedule_sync_task(sync_users(users))
+            proto_users = await serialize_users_for_node(users)
+            schedule_sync_task(sync_proto_users(proto_users))
         else:
             await sync_users(users)
 

app/operation/user.py

@@ -63,8 +63,8 @@
     UserSubscriptionUpdateList,
 )
 from app.node.sync import remove_user as sync_remove_user
-from app.node.sync import schedule_sync_task, sync_proto_user, sync_user, sync_users
-from app.node.user import serialize_user
+from app.node.sync import schedule_sync_task, sync_proto_user, sync_proto_users, sync_user, sync_users
+from app.node.user import serialize_user, serialize_users_for_node
 from app.operation import BaseOperation, OperatorType
 from app.settings import subscription_settings
 from app.utils.jwt import create_subscription_token
@@ -746,7 +746,8 @@ def builder(username: str):
     async def bulk_modify_expire(self, db: AsyncSession, bulk_model: BulkUser):
         users, users_count = await update_users_expire(db, bulk_model)
         if self._is_non_blocking_sync_operator(self.operator_type):
-            schedule_sync_task(sync_users(users))
+            proto_users = await serialize_users_for_node(users)
+            schedule_sync_task(sync_proto_users(proto_users))
         else:
             await sync_users(users)
 
@@ -757,7 +758,8 @@ async def bulk_modify_expire(self, db: AsyncSession, bulk_model: BulkUser):
     async def bulk_modify_datalimit(self, db: AsyncSession, bulk_model: BulkUser):
         users, users_count = await update_users_datalimit(db, bulk_model)
         if self._is_non_blocking_sync_operator(self.operator_type):
-            schedule_sync_task(sync_users(users))
+            proto_users = await serialize_users_for_node(users)
+            schedule_sync_task(sync_proto_users(proto_users))
         else:
             await sync_users(users)
 
@@ -768,7 +770,8 @@ async def bulk_modify_datalimit(self, db: AsyncSession, bulk_model: BulkUser):
     async def bulk_modify_proxy_settings(self, db: AsyncSession, bulk_model: BulkUsersProxy):
         users, users_count = await update_users_proxy_settings(db, bulk_model)
         if self._is_non_blocking_sync_operator(self.operator_type):
-            schedule_sync_task(sync_users(users))
+            proto_users = await serialize_users_for_node(users)
+            schedule_sync_task(sync_proto_users(proto_users))
         else:
             await sync_users(users)