fix periodic-check request queue throughput bottleneck

The bot's request_queue.py 4s FIFO gate wasn't protecting against Lichess's
rate limiter — that's already handled downstream in LichessWebServices/
rate_limiter.py (0.2s, shared across all callers of our stats service). The
bot-side gate only paced calls to our own local service, and since it awaited
each request to full completion before dequeuing the next, real dispatch gaps
were max(4s, previous request's duration) — with 454 tracked gamer/user pairs,
any burst (e.g. after a restart) piled into the queue and took 10-20+ minutes
to drain.

Replace it with a paced-dispatch + bounded-concurrency design: a hard 2s floor
between dispatches (still never lets 2+ requests through in that window),
decoupled from completion time, with up to 10 requests actually in flight at
once via a semaphore. Doesn't touch the real Lichess-facing rate limit at all.

Also add deterministic per-(user,gamer) checkpoint jitter: previously every
pair sharing the same period_minutes re-locked onto the same wall-clock phase
on every restart (backlog collapse snaps period_end_approx to `now` for
everyone overdue at once), recreating the pileup each time. Jitter is stable
across restarts (crc32-based, not Python's salted hash()) and capped well
under the 2h stale-backlog threshold. Small startup stagger added too, purely
cosmetic smoothing on top of the jitter fix.

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:
+            active_gamers = [g for g in gamers_with_periods if g['period_minutes'] > 0]
-                if gamer['period_minutes'] > 0:
+            # Stagger task creation across a short startup window instead of firing all
-                    user_id = gamer['user_id']
+            # ~N tasks' first (blocking) sqlite checkpoint read in the same event loop
-                    username = gamer['username']
+            # tick. Small and one-off — the real anti-lockstep fix is the per-pair
-                    period = gamer['period_minutes']
+            # checkpoint jitter in periodic_check, this just smooths process boot.
-                    # Start periodic task with user_id and gamer
+            stagger_step = config.PERIODIC_STARTUP_STAGGER_MAX_SECONDS / max(1, len(active_gamers))
-                    await self.start_periodic_task(gamer, user_id, period)
+
-                    logger.info(f"✅ Started periodic task for {username} (user {user_id}) with period {period} minutes")
+            for gamer in active_gamers:
-            
+                user_id = gamer['user_id']
-            logger.info(f"✅ All periodic tasks started. Total: {len([g for g in gamers_with_periods if g['period_minutes'] > 0])}")
+                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)
                     
-                    # Получаем текущее время
+                    # Получаем текущее время (с джиттером, см. _checkpoint_jitter_seconds)
-                    period_end_approx = datetime.now()
+                    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}")

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:

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.
+    Queue for managing requests to our local stats API.
-    Ensures minimum delay between requests.
+
+    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
+                # Hard floor: never dispatch two requests closer than
-                await self._wait_if_needed()
+                # min_dispatch_interval apart, regardless of how long the
-                
+                # previous one takes to finish (that's what self.semaphore
-                # Execute the request
+                # bounds separately, not this gate).
-                func = request_item['func']
+                await self._wait_for_dispatch_slot()
-                args = request_item['args']
+
-                kwargs = request_item['kwargs']
+                task = asyncio.create_task(self._execute(request_item))
-                future = request_item['future']
+                self._in_flight.add(task)
-                
+                task.add_done_callback(self._in_flight.discard)
-                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
                 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):
+    async def _execute(self, request_item: dict):
-        """Wait if necessary to maintain minimum delay between requests"""
+        """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:
+            if self.last_dispatch_time is not None:
-                elapsed = now - self.last_request_time
+                elapsed = now - self.last_dispatch_time
-                if elapsed < self.min_delay:
+                if elapsed < self.min_dispatch_interval:
-                    wait_time = self.min_delay - elapsed
+                    wait_time = self.min_dispatch_interval - elapsed
-                    logger.debug(f"⏳ Rate limiter: waiting {wait_time:.2f} seconds")
+                    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
-