diff --git a/LichessClientTG_bot/bot.py b/LichessClientTG_bot/bot.py index 313c3c7..a8699b6 100644 --- a/LichessClientTG_bot/bot.py +++ b/LichessClientTG_bot/bot.py @@ -2,6 +2,7 @@ import asyncio import logging import sqlite3 import os +import zlib from datetime import datetime, timedelta from typing import Dict, Any, Optional from pathlib import Path @@ -13,6 +14,7 @@ from telegram.ext import ( PicklePersistence ) +import config from config import ( TELEGRAM_BOT_TOKEN, PERIOD_OPTIONS, POLL_INTERVAL, POLL_TIMEOUT, DROP_PENDING_UPDATES, ALLOWED_UPDATES, @@ -224,16 +226,24 @@ class LichessBot: self.request_queue._start_processor() logger.info("✅ Request queue processor started") - for gamer in gamers_with_periods: - if gamer['period_minutes'] > 0: - user_id = gamer['user_id'] - username = gamer['username'] - period = gamer['period_minutes'] - # Start periodic task with user_id and gamer - await self.start_periodic_task(gamer, user_id, period) - logger.info(f"✅ Started periodic task for {username} (user {user_id}) with period {period} minutes") - - logger.info(f"✅ All periodic tasks started. Total: {len([g for g in gamers_with_periods if g['period_minutes'] > 0])}") + active_gamers = [g for g in gamers_with_periods if g['period_minutes'] > 0] + # Stagger task creation across a short startup window instead of firing all + # ~N tasks' first (blocking) sqlite checkpoint read in the same event loop + # tick. Small and one-off — the real anti-lockstep fix is the per-pair + # checkpoint jitter in periodic_check, this just smooths process boot. + stagger_step = config.PERIODIC_STARTUP_STAGGER_MAX_SECONDS / max(1, len(active_gamers)) + + for gamer in active_gamers: + user_id = gamer['user_id'] + username = gamer['username'] + period = gamer['period_minutes'] + # Start periodic task with user_id and gamer + await self.start_periodic_task(gamer, user_id, period) + logger.info(f"✅ Started periodic task for {username} (user {user_id}) with period {period} minutes") + if stagger_step > 0: + await asyncio.sleep(stagger_step) + + logger.info(f"✅ All periodic tasks started. Total: {len(active_gamers)}") # Start daily counter reset task asyncio.create_task(self.daily_counter_reset_task()) @@ -1534,6 +1544,24 @@ class LichessBot: # подряд идущих ошибок (~2 часа при капнутом бэкоффе в 300с на попытку) ADMIN_NOTIFY_ERROR_THRESHOLD = 25 + @staticmethod + def _checkpoint_jitter_seconds(user_id: int, gamer_id: int, period_minutes: int) -> float: + """ + Deterministic per-(user, gamer) jitter applied whenever a checkpoint gets + pinned to "now" (backlog collapse / first check). Without this, every pair + sharing the same period_minutes re-locks onto the exact same wall-clock + phase on every bot restart, causing them all to become "due" in the same + instant forever after. Stable across restarts (unlike Python's salted + hash()), and capped well below STALE_BACKLOG_THRESHOLD so it never + interacts with backlog-notification suppression. + """ + jitter_cap = min(period_minutes * 60 * config.PERIODIC_CHECKPOINT_JITTER_FRACTION, + config.PERIODIC_CHECKPOINT_JITTER_MAX_SECONDS) + if jitter_cap <= 0: + return 0.0 + digest = zlib.crc32(f"{user_id}:{gamer_id}".encode()) + return digest % jitter_cap + async def periodic_check(self, gamer: Dict[str, Any], user_id: int, period_minutes: int): """Periodic check for gamer activity""" task_key = f"{gamer['id']}_{user_id}" @@ -1606,8 +1634,11 @@ class LichessBot: # игроков через один общий RequestQueue) чекпоинт никогда не догонит # текущее время — отставание только растёт. Вместо этого закрываем # весь пропущенный промежуток одним запросом и сразу прыгаем к "сейчас". + # Джиттер (см. _checkpoint_jitter_seconds) не даёт всем парам с + # одинаковым period_minutes зафиксироваться на один и тот же момент. since_time = last_check_time - period_end_approx = now + jitter = self._checkpoint_jitter_seconds(user_id, gamer['id'], period_minutes) + period_end_approx = now - timedelta(seconds=jitter) if period_end_approx - next_period_start > timedelta(minutes=period_minutes): logger.warning( f"⏰ {username} is behind schedule by " @@ -1620,8 +1651,9 @@ class LichessBot: logger.info(f"⏳ First check: waiting {period_minutes} minutes before first check for {username}") await asyncio.sleep(period_minutes * 60) - # Получаем текущее время - period_end_approx = datetime.now() + # Получаем текущее время (с джиттером, см. _checkpoint_jitter_seconds) + jitter = self._checkpoint_jitter_seconds(user_id, gamer['id'], period_minutes) + period_end_approx = datetime.now() - timedelta(seconds=jitter) # Начало периода - текущее время минус period_minutes since_time = period_end_approx - timedelta(minutes=period_minutes) logger.info(f"📌 First check: period from {since_time} to {period_end_approx}") diff --git a/LichessClientTG_bot/config.py b/LichessClientTG_bot/config.py index b58ed54..b355981 100644 --- a/LichessClientTG_bot/config.py +++ b/LichessClientTG_bot/config.py @@ -10,8 +10,25 @@ ADMINPANEL_TELEGRAM_BOT_TOKEN = os.getenv("ADMINPANEL_TELEGRAM_BOT_TOKEN") # Lichess API Configuration LICHESS_API_BASE_URL = "https://lichess.org/api" LICHESS_STATS_API_BASE_URL = "http://localhost:8002" # Host port for stats API when bot runs with host networking -# Минимальная задержка (сек) между запросами к Lichess в очереди мониторинга (избежание бана) -LICHESS_REQUEST_QUEUE_MIN_DELAY = 4.0 + +# Пейсинг очереди запросов к НАШЕМУ ЖЕ локальному stats-сервису (LICHESS_STATS_API_BASE_URL). +# Это НЕ защита от рейт-лимитера Lichess — та уже есть ниже по стеку, в +# LichessWebServices/rate_limiter.py (0.2s, единственный process-wide инстанс, +# применяется перед каждым реальным вызовом lichess.org, общий для всех клиентов +# stats-сервиса). Эти числа можно свободно менять — они не увеличивают нагрузку на Lichess. +LICHESS_REQUEST_QUEUE_MIN_DISPATCH_INTERVAL = 2.0 # сек между диспетчами запросов (жёсткий пол) +LICHESS_REQUEST_QUEUE_MAX_CONCURRENT = 10 # макс. запросов к stats-сервису одновременно; 1 = откат к строго последовательному режиму + +# Джиттер чекпоинтов периодических проверок: не даёт парам (user, gamer) с одинаковым +# period_minutes синхронизироваться на один и тот же wall-clock момент при рестарте бота +# (см. periodic_check в bot.py). Потолок держим далеко ниже STALE_BACKLOG_THRESHOLD (2ч), +# чтобы не задевать логику подавления уведомлений по устаревшему бэклогу. +PERIODIC_CHECKPOINT_JITTER_FRACTION = 0.10 +PERIODIC_CHECKPOINT_JITTER_MAX_SECONDS = 300 + +# Разброс старта периодических задач при запуске бота, чтобы не бить по SQLite +# синхронно за один тик event loop при большом числе отслеживаемых игроков. +PERIODIC_STARTUP_STAGGER_MAX_SECONDS = 30 # Database Configuration def _resolve_database_path() -> str: diff --git a/LichessClientTG_bot/request_queue.py b/LichessClientTG_bot/request_queue.py index 3a81a3d..420cd6f 100644 --- a/LichessClientTG_bot/request_queue.py +++ b/LichessClientTG_bot/request_queue.py @@ -1,10 +1,17 @@ """ Request Queue for managing API requests with rate limiting -Ensures minimum delay between requests to avoid DDoS and rate limiting + +Paces how fast we dispatch requests to our own local stats API +(http://localhost:8002), and caps how many of those requests may be in +flight at once. This does NOT protect against Lichess's real rate limiter — +that protection already lives downstream, in LichessWebServices/rate_limiter.py +(0.2s min delay, single shared instance, applied before every actual +lichess.org call, shared by ALL callers of stats_service — not just this +bot). These numbers only govern bot -> local-service traffic. """ import asyncio import logging -from typing import Callable, Any, Optional, Dict +from typing import Callable, Any, Optional, Set from datetime import datetime import config @@ -13,91 +20,89 @@ logger = logging.getLogger(__name__) class RequestQueue: """ - Queue for managing API requests with rate limiting. - Ensures minimum delay between requests. + Queue for managing requests to our local stats API. + + Dispatch is paced by a hard floor (min_dispatch_interval): two requests + can never be dispatched closer together than that, enforced by a lock in + the processor loop, independent of how long any individual request takes + to complete. Actual execution is capped separately by max_concurrent via + a semaphore, so a slow request can't stall the whole queue behind it. """ - - def __init__(self, min_delay: float = 7.0): + + def __init__(self, min_dispatch_interval: float, max_concurrent: int): """ - Initialize request queue. - Args: - min_delay: Minimum delay in seconds between requests (default: 7.0) + min_dispatch_interval: Minimum seconds between successive dispatches + max_concurrent: Maximum requests in flight at once (set to 1 to + fully serialize execution again, e.g. as a rollback lever) """ - self.min_delay = min_delay + self.min_dispatch_interval = min_dispatch_interval + self.max_concurrent = max_concurrent self.queue = asyncio.Queue() self.is_processing = False - self.last_request_time: Optional[float] = None + self.last_dispatch_time: Optional[float] = None self.lock = asyncio.Lock() + self.semaphore = asyncio.Semaphore(max_concurrent) self._processor_task: Optional[asyncio.Task] = None - + self._in_flight: Set[asyncio.Task] = set() + async def add_request(self, request_func: Callable, *args, **kwargs) -> Any: """ Add a request to the queue and wait for its result. - + Args: request_func: Async function to call *args: Positional arguments for the function **kwargs: Keyword arguments for the function - + Returns: Result of the request function """ - # Create a future to wait for the result future = asyncio.Future() - - # Add request to queue + await self.queue.put({ 'func': request_func, 'args': args, 'kwargs': kwargs, 'future': future }) - - # Start processor if not already running + if not self.is_processing: self._start_processor() - - # Wait for result + return await future - + def _start_processor(self): """Start the queue processor task""" if self._processor_task is None or self._processor_task.done(): self.is_processing = True self._processor_task = asyncio.create_task(self._process_queue()) - logger.info(f"🚀 Started request queue processor (delay: {self.min_delay}s)") - + logger.info( + f"🚀 Started request queue processor " + f"(dispatch interval: {self.min_dispatch_interval}s, " + f"max concurrent: {self.max_concurrent})" + ) + async def _process_queue(self): - """Process requests from the queue with rate limiting""" + """Dequeue requests and dispatch them at a paced rate, without waiting for completion""" logger.info("📋 Request queue processor started") - + while True: try: - # Get next request from queue (wait indefinitely) request_item = await self.queue.get() - - # Wait if needed to maintain minimum delay - await self._wait_if_needed() - - # Execute the request - func = request_item['func'] - args = request_item['args'] - kwargs = request_item['kwargs'] - future = request_item['future'] - - try: - logger.debug(f"🔄 Executing request: {func.__name__}") - result = await func(*args, **kwargs) - future.set_result(result) - logger.debug(f"✅ Request completed: {func.__name__}") - except Exception as e: - logger.error(f"❌ Request failed: {func.__name__}: {e}") - future.set_exception(e) - - # Mark task as done + + # Hard floor: never dispatch two requests closer than + # min_dispatch_interval apart, regardless of how long the + # previous one takes to finish (that's what self.semaphore + # bounds separately, not this gate). + await self._wait_for_dispatch_slot() + + task = asyncio.create_task(self._execute(request_item)) + self._in_flight.add(task) + task.add_done_callback(self._in_flight.discard) + self.queue.task_done() - + except asyncio.CancelledError: logger.info("🛑 Request queue processor cancelled") break @@ -105,32 +110,56 @@ class RequestQueue: logger.error(f"❌ Error in request queue processor: {e}") import traceback logger.error(traceback.format_exc()) - # Continue processing await asyncio.sleep(1) - - async def _wait_if_needed(self): - """Wait if necessary to maintain minimum delay between requests""" + + async def _execute(self, request_item: dict): + """Run one queued request, bounded by the concurrency semaphore""" + func = request_item['func'] + args = request_item['args'] + kwargs = request_item['kwargs'] + future = request_item['future'] + + async with self.semaphore: + try: + logger.debug(f"🔄 Executing request: {func.__name__}") + result = await func(*args, **kwargs) + if not future.done(): + future.set_result(result) + logger.debug(f"✅ Request completed: {func.__name__}") + except Exception as e: + logger.error(f"❌ Request failed: {func.__name__}: {e}") + if not future.done(): + future.set_exception(e) + + async def _wait_for_dispatch_slot(self): + """Wait if necessary to maintain minimum delay between dispatches""" async with self.lock: now = datetime.now().timestamp() - - if self.last_request_time is not None: - elapsed = now - self.last_request_time - if elapsed < self.min_delay: - wait_time = self.min_delay - elapsed - logger.debug(f"⏳ Rate limiter: waiting {wait_time:.2f} seconds") + + if self.last_dispatch_time is not None: + elapsed = now - self.last_dispatch_time + if elapsed < self.min_dispatch_interval: + wait_time = self.min_dispatch_interval - elapsed + logger.debug(f"⏳ Dispatch pacing: waiting {wait_time:.2f} seconds") await asyncio.sleep(wait_time) now = datetime.now().timestamp() - - self.last_request_time = now - + + self.last_dispatch_time = now + async def stop(self): - """Stop the queue processor""" + """Stop the queue processor and any in-flight requests""" if self._processor_task and not self._processor_task.done(): self._processor_task.cancel() try: await self._processor_task except asyncio.CancelledError: pass + + for task in list(self._in_flight): + task.cancel() + if self._in_flight: + await asyncio.gather(*self._in_flight, return_exceptions=True) + self.is_processing = False logger.info("🛑 Request queue processor stopped") @@ -141,6 +170,8 @@ def get_request_queue() -> RequestQueue: """Get the global request queue instance""" global _request_queue if _request_queue is None: - _request_queue = RequestQueue(min_delay=config.LICHESS_REQUEST_QUEUE_MIN_DELAY) + _request_queue = RequestQueue( + min_dispatch_interval=config.LICHESS_REQUEST_QUEUE_MIN_DISPATCH_INTERVAL, + max_concurrent=config.LICHESS_REQUEST_QUEUE_MAX_CONCURRENT, + ) return _request_queue -