From d4eb61f99439ae39558c4558dddbe989ddfa503c Mon Sep 17 00:00:00 2001 From: vrubelroman Date: Sun, 26 Jul 2026 10:25:34 +0000 Subject: [PATCH] prioritize interactive requests over background checks in shared Lichess-token queue /today, /yesterday, /week were queuing behind the entire periodic-check backlog on the same rate-limited LICHESS_APP_TOKEN (observed ~2min wait on prod for a single command). Add a two-tier priority queue to SharedTokenGate: interactive on-demand requests jump ahead of background periodic checks, which still drain normally when nothing interactive is waiting. Verified locally end-to-end (isolated queue unit tests, live request with priority correctly reaching Lichess, and a real successful accuracy fetch rendered through StatsFormatter). Co-Authored-By: Claude Sonnet 5 --- LichessClientTG_bot/bot.py | 3 ++- LichessClientTG_bot/lichess_api.py | 13 +++++++--- LichessWebServices/lichess_client.py | 31 ++++++++++++++++------- LichessWebServices/main.py | 8 ++++-- LichessWebServices/rate_limiter.py | 38 +++++++++++++++++++++++----- LichessWebServices/stats_service.py | 8 +++--- 6 files changed, 76 insertions(+), 25 deletions(-) diff --git a/LichessClientTG_bot/bot.py b/LichessClientTG_bot/bot.py index 6331fea..cdcfd35 100644 --- a/LichessClientTG_bot/bot.py +++ b/LichessClientTG_bot/bot.py @@ -1119,7 +1119,8 @@ class LichessBot: int(since_dt.timestamp() * 1000), int(until_dt.timestamp() * 1000), rated_only=True, - include_games=want_rows + include_games=want_rows, + priority="interactive" ) except Exception as e: logger.warning(f"⚠️ Supplementary accuracy fetch failed for {username}/{period}: {e}") diff --git a/LichessClientTG_bot/lichess_api.py b/LichessClientTG_bot/lichess_api.py index c21dad5..26cc162 100644 --- a/LichessClientTG_bot/lichess_api.py +++ b/LichessClientTG_bot/lichess_api.py @@ -104,12 +104,19 @@ class LichessAPI: logger.error(f"Error getting week stats: {e}") return None - async def get_games_period(self, username: str, since: int, until: int, rated_only: Optional[bool] = None, include_games: bool = False) -> Optional[Dict[str, Any]]: - """Get games for a specific period""" + async def get_games_period(self, username: str, since: int, until: int, rated_only: Optional[bool] = None, include_games: bool = False, priority: str = "background") -> Optional[Dict[str, Any]]: + """ + Get games for a specific period. + + priority="interactive" (e.g. the accuracy fetch behind /today, /yesterday, + /week) jumps ahead of "background" (periodic_check) in the stats + service's shared Lichess-token queue, so an on-demand command doesn't + wait behind the whole periodic-check backlog. + """ await self.rate_limiter.wait_if_needed() try: url = f"{self.stats_base_url}/games/{username}/period" - params = {"since": since, "until": until} + params = {"since": since, "until": until, "priority": priority} if rated_only is not None: params["rated_only"] = "true" if rated_only else "false" if include_games: diff --git a/LichessWebServices/lichess_client.py b/LichessWebServices/lichess_client.py index 3c3bb81..0ec1624 100644 --- a/LichessWebServices/lichess_client.py +++ b/LichessWebServices/lichess_client.py @@ -84,11 +84,16 @@ class LichessClient: headers = {'Authorization': f'Bearer {self.app_token}'} if self.app_token else {} - # Держим shared_token_gate на весь запрос-ответ и с паузой после — # LICHESS_APP_TOKEN общий с get_games_of_period, и Lichess не - # пускает 2+ запроса подряд без ощутимой паузы между ними. - async with self.shared_token_gate: + # пускает 2+ запроса подряд без ощутимой паузы между ними — держим + # shared_token_gate на весь запрос-ответ. get_user_activity вызывается + # только из интерактивных команд (/today и т.п.), периодические проверки + # его не используют — приоритет всегда "interactive", параметризовать не нужно. + await self.shared_token_gate.acquire(priority="interactive") + try: response = await self.client.get(url, headers=headers) + finally: + self.shared_token_gate.release() logger.info(f"🔍 Lichess API response status: {response.status_code} for {username}") response.raise_for_status() # Проверяем статус ответа @@ -113,19 +118,22 @@ class LichessClient: logger.error(traceback.format_exc()) raise - async def get_games_of_period(self, username: str, since_ms: int, until_ms: int, rated_only: bool = True) -> Optional[List[Dict[str, Any]]]: + async def get_games_of_period(self, username: str, since_ms: int, until_ms: int, rated_only: bool = True, priority: str = "background") -> Optional[List[Dict[str, Any]]]: """ Получает игры пользователя за определенный период. - + Lichess API возвращает игры в формате NDJSON (Newline Delimited JSON), где каждая строка содержит JSON объект с информацией об игре. - + Args: username: Имя пользователя на Lichess since_ms: Начало периода в миллисекундах (Unix timestamp * 1000) until_ms: Конец периода в миллисекундах (Unix timestamp * 1000) rated_only: Только рейтинговые игры (по умолчанию True) - + priority: "high" для интерактивных команд (/today и т.п.), чтобы не + ждать позади бэклога фоновых периодических проверок в общей + очереди shared_token_gate; "background" (по умолчанию) для них. + Returns: Список игр в формате JSON или None при ошибке @@ -157,15 +165,20 @@ class LichessClient: if self.app_token: headers['Authorization'] = f'Bearer {self.app_token}' - logger.info(f"Запрос игр для {username} с {since_ms} по {until_ms}") + logger.info(f"Запрос игр для {username} с {since_ms} по {until_ms} (priority={priority})") # Держим shared_token_gate на весь запрос-ответ и с паузой после — # LICHESS_APP_TOKEN общий с get_user_activity, и Lichess не пускает # 2+ запроса подряд без ощутимой паузы между ними (даже строго # последовательные с интервалом ~100-300мс ловили 429 "Please only # run 1 request(s) at a time" под реальной нагрузкой на проде). - async with self.shared_token_gate: + # priority="interactive" (например, довесочный accuracy-fetch в /today) + # обгоняет фоновые периодические проверки в этой же очереди. + await self.shared_token_gate.acquire(priority=priority) + try: response = await self.client.get(url, params=params, headers=headers) + finally: + self.shared_token_gate.release() response.raise_for_status() # Проверяем статус ответа # Парсим NDJSON (Newline Delimited JSON) diff --git a/LichessWebServices/main.py b/LichessWebServices/main.py index 7e96c83..d4a0fb5 100644 --- a/LichessWebServices/main.py +++ b/LichessWebServices/main.py @@ -503,7 +503,11 @@ async def get_games_of_period( example=True), include_games: bool = Query(False, description="Включить построчный список отдельных партий (только blitz/rapid/classical)", - example=False) + example=False), + priority: str = Query("background", + pattern="^(background|interactive)$", + description="\"interactive\" (запросы из /today и т.п.) обгоняет \"background\" (периодические проверки) в общей очереди shared-token запросов к Lichess", + example="background") ): """ ## Статистика игр за период @@ -557,7 +561,7 @@ async def get_games_of_period( # Конвертируем миллисекунды в секунды для внутренней логики since_seconds = since // 1000 until_seconds = until // 1000 - result = await stats_service.get_games_of_period(username, since_seconds, until_seconds, rated_only, include_games) + result = await stats_service.get_games_of_period(username, since_seconds, until_seconds, rated_only, include_games, priority=priority) if not result.success: # Реальная ошибка при обращении к Lichess — не маскируем её под "0 игр" raise HTTPException(status_code=502, detail=result.message) diff --git a/LichessWebServices/rate_limiter.py b/LichessWebServices/rate_limiter.py index 7789c93..9063042 100644 --- a/LichessWebServices/rate_limiter.py +++ b/LichessWebServices/rate_limiter.py @@ -5,6 +5,7 @@ Ensures minimum delay between requests import asyncio import time import logging +from collections import deque from typing import Optional logger = logging.getLogger(__name__) @@ -65,26 +66,49 @@ class SharedTokenGate: Lichess is enforcing here needs a real cooldown between calls, not just non-overlap, so this gate tracks the last release time and forces a minimum gap before the next request is allowed to start. + + Two-tier priority: background periodic checks queue behind interactive + on-demand requests (/today etc.), so a user command doesn't have to wait + out an entire backlog of checks (observed ~2min on prod) — just whatever + single request is already in flight. acquire()/release() (not a plain + `async with`) because the priority has to be picked at call time. """ def __init__(self, min_gap: float = 6.0): - self._lock = asyncio.Lock() self._min_gap = min_gap self._last_release_time: Optional[float] = None + self._busy = False + self._interactive_waiters: "deque[asyncio.Future]" = deque() + self._background_waiters: "deque[asyncio.Future]" = deque() + + async def acquire(self, priority: str = "background"): + """priority: "interactive" (/today etc.) jumps ahead of "background" (periodic checks).""" + if not self._busy and not self._interactive_waiters and not self._background_waiters: + self._busy = True + else: + fut = asyncio.get_event_loop().create_future() + queue = self._interactive_waiters if priority == "interactive" else self._background_waiters + queue.append(fut) + await fut - async def __aenter__(self): - await self._lock.acquire() if self._last_release_time is not None: elapsed = time.time() - self._last_release_time if elapsed < self._min_gap: wait_time = self._min_gap - elapsed logger.debug(f"Shared token gate: waiting {wait_time:.3f}s before next games/user or user/activity call") await asyncio.sleep(wait_time) - return self - async def __aexit__(self, exc_type, exc, tb): + def release(self): self._last_release_time = time.time() - self._lock.release() + if self._interactive_waiters: + nxt = self._interactive_waiters.popleft() + elif self._background_waiters: + nxt = self._background_waiters.popleft() + else: + self._busy = False + return + if not nxt.done(): + nxt.set_result(None) # min_gap=6.0: no documented number from Lichess for this endpoint's real # limit: the games/user docs only mention a streaming throttle (games/sec @@ -95,6 +119,6 @@ class SharedTokenGate: _shared_token_gate = SharedTokenGate(min_gap=6.0) def get_shared_token_gate() -> SharedTokenGate: - """Async context manager serializing+pacing lichess.org calls made with the shared LICHESS_APP_TOKEN.""" + """Gate serializing+pacing lichess.org calls made with the shared LICHESS_APP_TOKEN. Use acquire()/release().""" return _shared_token_gate diff --git a/LichessWebServices/stats_service.py b/LichessWebServices/stats_service.py index 50b3fb6..3889b52 100644 --- a/LichessWebServices/stats_service.py +++ b/LichessWebServices/stats_service.py @@ -698,7 +698,7 @@ class StatsService: total=create_game_stats(stats['total']) ) - async def get_games_of_period(self, username: str, since_timestamp: int, until_timestamp: int, rated_only: bool = True, include_games: bool = False) -> GamesOfPeriodResponse: + async def get_games_of_period(self, username: str, since_timestamp: int, until_timestamp: int, rated_only: bool = True, include_games: bool = False, priority: str = "background") -> GamesOfPeriodResponse: """ Получает статистику игр пользователя за определенный период. @@ -711,6 +711,8 @@ class StatsService: until_timestamp: Конец периода (Unix timestamp в секундах) rated_only: Только рейтинговые игры (по умолчанию True) include_games: Включить построчный список отдельных партий (только blitz/rapid/classical) + priority: "interactive" для запросов из /today и т.п. (обгоняют фоновые + периодические проверки в общей очереди shared_token_gate), иначе "background" Returns: GamesOfPeriodResponse с статистикой игр @@ -719,9 +721,9 @@ class StatsService: # Конвертируем timestamp в миллисекунды для API Lichess since_ms = since_timestamp * 1000 until_ms = until_timestamp * 1000 - + # Получаем игры (без фильтра rated в запросе к Lichess — см. lichess_client) - games = await self.lichess_client.get_games_of_period(username, since_ms, until_ms, rated_only=False) + games = await self.lichess_client.get_games_of_period(username, since_ms, until_ms, rated_only=False, priority=priority) if games is None: return GamesOfPeriodResponse(