From 5235f7395f4baef573e86a2ebe5037c4346631f4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Fri, 25 Sep 2026 15:30:57 +0200 Subject: [PATCH 1/2] fix(indexer): add graceful shutdown and correctly propagate Telegram API exceptions This adds an asyncio.sleep(0.250) on close to allow aiohttp SSL transports to cleanly disconnect, preventing SIGKILL from file descriptor leaks. It also ensures 'user not found' exceptions are correctly raised rather than wiping a user's DB gifts. --- backend/indexer_gifts/indexers/bot_api.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/backend/indexer_gifts/indexers/bot_api.py b/backend/indexer_gifts/indexers/bot_api.py index 75bd7a07..f7478533 100644 --- a/backend/indexer_gifts/indexers/bot_api.py +++ b/backend/indexer_gifts/indexers/bot_api.py @@ -1,4 +1,5 @@ import logging +import asyncio from typing import AsyncGenerator from aiogram import Bot @@ -59,7 +60,9 @@ async def iter_user_gifts( except Exception as e: logger.error(f"Error fetching gifts for user {telegram_user_id}: {e}") - break + raise async def close(self): await self.bot.session.close() + # Allow underlying aiohttp SSL connections to close gracefully + await asyncio.sleep(0.250) From aabd84b613e120a567c8a210227bc02588983f0a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Danii=C5=82=20M=2E?= Date: Fri, 25 Sep 2026 15:31:03 +0200 Subject: [PATCH 2/2] feat(gifts): batch sync optimization and auto-blocking for inactive users - Moves batch syncing into IndexerUserGiftAction to share a single BotApiGiftIndexer instance, vastly reducing overhead and event loop thrashing. - Adds UserService.mark_as_blocked() and excludes blocked users from global fetches. - Automatically marks users as blocked when Telegram returns a 'user not found' exception during gift sync. --- backend/core/src/core/services/user.py | 16 +- backend/indexer_gifts/actions/user.py | 229 +++++++++++++--------- backend/indexer_gifts/indexers/bot_api.py | 6 + backend/indexer_gifts/tasks.py | 16 +- 4 files changed, 156 insertions(+), 111 deletions(-) diff --git a/backend/core/src/core/services/user.py b/backend/core/src/core/services/user.py index 9b05cf23..fc44ad3a 100644 --- a/backend/core/src/core/services/user.py +++ b/backend/core/src/core/services/user.py @@ -1,4 +1,5 @@ -from typing import Iterable, Any +import logging +from typing import Any, Iterable from sqlalchemy.orm import joinedload, load_only, QueryableAttribute @@ -6,6 +7,8 @@ from core.models.user import User from core.services.base import BaseService +logger = logging.getLogger(__name__) + class UserService(BaseService): def get_all( @@ -27,9 +30,20 @@ def get_all_telegram_ids(self) -> list[int]: u.telegram_id for u in self.db_session.query(User.telegram_id) .filter(User.telegram_id.is_not(None)) + .filter(User.is_blocked.is_(False)) .all() ] + def mark_as_blocked(self, telegram_ids: list[int]) -> None: + if not telegram_ids: + return + + logger.info(f"Marking {len(telegram_ids)} users as blocked.") + self.db_session.query(User).filter(User.telegram_id.in_(telegram_ids)).update( + {"is_blocked": True}, synchronize_session=False + ) + self.db_session.flush() + def get_by_telegram_id(self, telegram_id: int) -> User: return ( self.db_session.query(User) diff --git a/backend/indexer_gifts/actions/user.py b/backend/indexer_gifts/actions/user.py index 651a8de9..db98344c 100644 --- a/backend/indexer_gifts/actions/user.py +++ b/backend/indexer_gifts/actions/user.py @@ -2,10 +2,11 @@ from sqlalchemy.orm import Session from core.actions.base import BaseAction -from core.services.gift.item import GiftUniqueService +from core.constants import UPDATED_GIFT_USER_IDS from core.services.gift.collection import GiftCollectionService +from core.services.gift.item import GiftUniqueService from core.services.superredis import RedisService -from core.constants import UPDATED_GIFT_USER_IDS +from core.services.user import UserService from indexer_gifts.indexers.bot_api import BotApiGiftIndexer logger = logging.getLogger(__name__) @@ -17,16 +18,51 @@ def __init__(self, db_session: Session): self.gift_service = GiftUniqueService(db_session) self.gift_collection_service = GiftCollectionService(db_session) self.redis_service = RedisService() + self.user_service = UserService(db_session) + + async def sync_single_user_gifts(self, telegram_user_id: int) -> None: + """Public method for single-user sync, manages the indexer lifecycle.""" + async with BotApiGiftIndexer() as indexer: + try: + await self._sync_user_gifts(telegram_user_id, indexer=indexer) + except Exception as e: + if "user not found" in str(e).lower(): + self.user_service.mark_as_blocked([telegram_user_id]) + raise + + async def sync_users_batch( + self, + telegram_user_ids: list[int], + existing_collection_ids: set[int] | None = None, + ) -> None: + """Public method for batch sync, manages a single indexer across multiple users.""" + blocked_user_ids = [] + async with BotApiGiftIndexer() as indexer: + for user_id in telegram_user_ids: + try: + await self._sync_user_gifts( + user_id, + indexer=indexer, + existing_collection_ids=existing_collection_ids, + ) + except Exception as e: + logger.error(f"Error refreshing gifts for user {user_id}: {e}") + if "user not found" in str(e).lower(): + blocked_user_ids.append(user_id) + + if blocked_user_ids: + self.user_service.mark_as_blocked(blocked_user_ids) - async def sync_user_gifts( - self, telegram_user_id: int, existing_collection_ids: set[int] | None = None + async def _sync_user_gifts( + self, + telegram_user_id: int, + indexer: BotApiGiftIndexer, + existing_collection_ids: set[int] | None = None, ) -> None: """ - Syncs unique gifts for a user using the Telegram Bot API. - Identifies gifts the user lost and updates their database ownership. + Internal business logic to sync unique gifts for a user using the provided indexer. """ logger.info(f"Starting gifts synchronizing for user {telegram_user_id}") - indexer = BotApiGiftIndexer() current_owned_slugs = set() lost_ownership_user_ids = set() created_count = 0 @@ -37,102 +73,99 @@ async def sync_user_gifts( if existing_collection_ids is None: existing_collection_ids = self.gift_collection_service.get_all_ids() - try: - # 1. Fetch current gifts from Telegram API - async for owned_gift in indexer.iter_user_gifts(telegram_user_id): - ug = owned_gift.gift - current_owned_slugs.add(ug.name) - - # Check if it exists - existing = self.gift_service.find(ug.name) - - if existing: - # If it was owned by someone else, record they lost it - if ( - existing.telegram_owner_id - and existing.telegram_owner_id != telegram_user_id - ): - lost_ownership_user_ids.add(existing.telegram_owner_id) - - self.gift_service.update_ownership( - slug=ug.name, - telegram_owner_id=telegram_user_id, - blockchain_address=existing.blockchain_address, - owner_address=existing.owner_address, - ) - updated_count += 1 - else: - # Ensure collection exists using in-memory cache - collection_id = int(ug.gift_id) - if collection_id not in existing_collection_ids: - logger.info( - f"Collection {collection_id} missing, creating stub collection." - ) - self.gift_collection_service.create( - id=collection_id, - title=ug.base_name or f"Gift {collection_id}", - preview_url=None, - supply=0, - upgraded_count=0, - ) - existing_collection_ids.add(collection_id) - - # Extract string identifiers for visual attributes if available - model_name = ( - getattr(ug.model, "name", None) - if getattr(ug, "model", None) - else None - ) - backdrop_name = ( - getattr(ug.backdrop, "name", None) - if getattr(ug, "backdrop", None) - else None - ) - pattern_name = ( - getattr(ug.symbol, "name", None) - if getattr(ug, "symbol", None) - else None - ) + # 1. Fetch current gifts from Telegram API + async for owned_gift in indexer.iter_user_gifts(telegram_user_id): + ug = owned_gift.gift + current_owned_slugs.add(ug.name) + + # Check if it exists + existing = self.gift_service.find(ug.name) + + if existing: + # If it was owned by someone else, record they lost it + if ( + existing.telegram_owner_id + and existing.telegram_owner_id != telegram_user_id + ): + lost_ownership_user_ids.add(existing.telegram_owner_id) - # Create new gift - self.gift_service.create( - slug=ug.name, - collection_id=collection_id, - number=ug.number, - model=model_name, - backdrop=backdrop_name, - pattern=pattern_name, - telegram_owner_id=telegram_user_id, - blockchain_address=None, - owner_address=None, + self.gift_service.update_ownership( + slug=ug.name, + telegram_owner_id=telegram_user_id, + blockchain_address=existing.blockchain_address, + owner_address=existing.owner_address, + ) + updated_count += 1 + else: + # Ensure collection exists using in-memory cache + collection_id = int(ug.gift_id) + if collection_id not in existing_collection_ids: + logger.info( + f"Collection {collection_id} missing, creating stub collection." ) - created_count += 1 - - # 2. Fetch all gifts currently assigned to the user in the database - db_gifts = self.gift_service.get_all(telegram_user_id=telegram_user_id) - for db_gift in db_gifts: - if db_gift.slug not in current_owned_slugs: - # User no longer owns this gift - self.gift_service.update_ownership( - slug=db_gift.slug, - telegram_owner_id=None, - blockchain_address=db_gift.blockchain_address, - owner_address=db_gift.owner_address, + self.gift_collection_service.create( + id=collection_id, + title=ug.base_name or f"Gift {collection_id}", + preview_url=None, + supply=0, + upgraded_count=0, ) - lost_ownership_user_ids.add(telegram_user_id) - lost_count += 1 + existing_collection_ids.add(collection_id) - # 3. Add users who lost ownership to the Redis set for community manager checking - if lost_ownership_user_ids: - logger.info( - f"Adding {len(lost_ownership_user_ids)} users to Redis for community manager check due to lost gifts." + # Extract string identifiers for visual attributes if available + model_name = ( + getattr(ug.model, "name", None) + if getattr(ug, "model", None) + else None + ) + backdrop_name = ( + getattr(ug.backdrop, "name", None) + if getattr(ug, "backdrop", None) + else None + ) + pattern_name = ( + getattr(ug.symbol, "name", None) + if getattr(ug, "symbol", None) + else None ) - for uid in lost_ownership_user_ids: - self.redis_service.add_to_set(UPDATED_GIFT_USER_IDS, str(uid)) + # Create new gift + self.gift_service.create( + slug=ug.name, + collection_id=collection_id, + number=ug.number, + model=model_name, + backdrop=backdrop_name, + pattern=pattern_name, + telegram_owner_id=telegram_user_id, + blockchain_address=None, + owner_address=None, + ) + created_count += 1 + + # 2. Fetch all gifts currently assigned to the user in the database + db_gifts = self.gift_service.get_all(telegram_user_id=telegram_user_id) + for db_gift in db_gifts: + if db_gift.slug not in current_owned_slugs: + # User no longer owns this gift + self.gift_service.update_ownership( + slug=db_gift.slug, + telegram_owner_id=None, + blockchain_address=db_gift.blockchain_address, + owner_address=db_gift.owner_address, + ) + lost_ownership_user_ids.add(telegram_user_id) + lost_count += 1 + + # 3. Add users who lost ownership to the Redis set for community manager checking + if lost_ownership_user_ids: logger.info( - f"Finished gifts synchronizing for user {telegram_user_id}. " - f"Created: {created_count}, Updated: {updated_count}, Lost: {lost_count}." + f"Adding {len(lost_ownership_user_ids)} users to Redis for community manager check due to lost gifts." ) - finally: - await indexer.close() + for uid in lost_ownership_user_ids: + self.redis_service.add_to_set(UPDATED_GIFT_USER_IDS, str(uid)) + + logger.info( + f"Finished gifts synchronizing for user {telegram_user_id}. " + f"Created: {created_count}, Updated: {updated_count}, Lost: {lost_count}." + ) diff --git a/backend/indexer_gifts/indexers/bot_api.py b/backend/indexer_gifts/indexers/bot_api.py index f7478533..3732e0ea 100644 --- a/backend/indexer_gifts/indexers/bot_api.py +++ b/backend/indexer_gifts/indexers/bot_api.py @@ -62,6 +62,12 @@ async def iter_user_gifts( logger.error(f"Error fetching gifts for user {telegram_user_id}: {e}") raise + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self.close() + async def close(self): await self.bot.session.close() # Allow underlying aiohttp SSL connections to close gracefully diff --git a/backend/indexer_gifts/tasks.py b/backend/indexer_gifts/tasks.py index 2a8e975a..5c4cd6a0 100644 --- a/backend/indexer_gifts/tasks.py +++ b/backend/indexer_gifts/tasks.py @@ -3,10 +3,10 @@ from core.constants import ( CELERY_GIFT_COLLECTIONS_SYNC_QUEUE, - CELERY_GIFT_USER_PRIORITY_QUEUE, CELERY_GIFT_USER_BATCH_QUEUE, - DEFAULT_CELERY_TASK_RETRY_DELAY, + CELERY_GIFT_USER_PRIORITY_QUEUE, DEFAULT_CELERY_TASK_MAX_RETRIES, + DEFAULT_CELERY_TASK_RETRY_DELAY, ) from core.services.db import DBService from core.services.user import UserService @@ -47,7 +47,7 @@ def sync_gift_collections_from_api() -> None: def index_user_gifts(telegram_user_id: int) -> bool: with DBService().db_session() as db_session: action = IndexerUserGiftAction(db_session) - asyncio.run(action.sync_user_gifts(telegram_user_id)) + asyncio.run(action.sync_single_user_gifts(telegram_user_id)) return True @@ -65,15 +65,7 @@ def refresh_users_gifts_batch(telegram_user_ids: list[int]) -> None: action = IndexerUserGiftAction(db_session) existing_collection_ids = action.gift_collection_service.get_all_ids() - for user_id in telegram_user_ids: - try: - asyncio.run( - action.sync_user_gifts( - user_id, existing_collection_ids=existing_collection_ids - ) - ) - except Exception as e: - logger.error(f"Error refreshing gifts for user {user_id}: {e}") + asyncio.run(action.sync_users_batch(telegram_user_ids, existing_collection_ids)) # === Flow C: Master Dispatcher Task ===