prioritize interactive requests over background checks in shared Lichess-token queue
All checks were successful
CI/CD Pipeline / build-and-deploy (push) Successful in 15s
All checks were successful
CI/CD Pipeline / build-and-deploy (push) Successful in 15s
/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 <noreply@anthropic.com>
This commit is contained in:
parent
0dbd0c400b
commit
d4eb61f994
6 changed files with 76 additions and 25 deletions
|
|
@ -1119,7 +1119,8 @@ class LichessBot:
|
||||||
int(since_dt.timestamp() * 1000),
|
int(since_dt.timestamp() * 1000),
|
||||||
int(until_dt.timestamp() * 1000),
|
int(until_dt.timestamp() * 1000),
|
||||||
rated_only=True,
|
rated_only=True,
|
||||||
include_games=want_rows
|
include_games=want_rows,
|
||||||
|
priority="interactive"
|
||||||
)
|
)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"⚠️ Supplementary accuracy fetch failed for {username}/{period}: {e}")
|
logger.warning(f"⚠️ Supplementary accuracy fetch failed for {username}/{period}: {e}")
|
||||||
|
|
|
||||||
|
|
@ -104,12 +104,19 @@ class LichessAPI:
|
||||||
logger.error(f"Error getting week stats: {e}")
|
logger.error(f"Error getting week stats: {e}")
|
||||||
return None
|
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]]:
|
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"""
|
"""
|
||||||
|
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()
|
await self.rate_limiter.wait_if_needed()
|
||||||
try:
|
try:
|
||||||
url = f"{self.stats_base_url}/games/{username}/period"
|
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:
|
if rated_only is not None:
|
||||||
params["rated_only"] = "true" if rated_only else "false"
|
params["rated_only"] = "true" if rated_only else "false"
|
||||||
if include_games:
|
if include_games:
|
||||||
|
|
|
||||||
|
|
@ -84,11 +84,16 @@ class LichessClient:
|
||||||
|
|
||||||
headers = {'Authorization': f'Bearer {self.app_token}'} if self.app_token else {}
|
headers = {'Authorization': f'Bearer {self.app_token}'} if self.app_token else {}
|
||||||
|
|
||||||
# Держим shared_token_gate на весь запрос-ответ и с паузой после —
|
|
||||||
# LICHESS_APP_TOKEN общий с get_games_of_period, и Lichess не
|
# LICHESS_APP_TOKEN общий с get_games_of_period, и Lichess не
|
||||||
# пускает 2+ запроса подряд без ощутимой паузы между ними.
|
# пускает 2+ запроса подряд без ощутимой паузы между ними — держим
|
||||||
async with self.shared_token_gate:
|
# 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)
|
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}")
|
logger.info(f"🔍 Lichess API response status: {response.status_code} for {username}")
|
||||||
response.raise_for_status() # Проверяем статус ответа
|
response.raise_for_status() # Проверяем статус ответа
|
||||||
|
|
||||||
|
|
@ -113,19 +118,22 @@ class LichessClient:
|
||||||
logger.error(traceback.format_exc())
|
logger.error(traceback.format_exc())
|
||||||
raise
|
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),
|
Lichess API возвращает игры в формате NDJSON (Newline Delimited JSON),
|
||||||
где каждая строка содержит JSON объект с информацией об игре.
|
где каждая строка содержит JSON объект с информацией об игре.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
username: Имя пользователя на Lichess
|
username: Имя пользователя на Lichess
|
||||||
since_ms: Начало периода в миллисекундах (Unix timestamp * 1000)
|
since_ms: Начало периода в миллисекундах (Unix timestamp * 1000)
|
||||||
until_ms: Конец периода в миллисекундах (Unix timestamp * 1000)
|
until_ms: Конец периода в миллисекундах (Unix timestamp * 1000)
|
||||||
rated_only: Только рейтинговые игры (по умолчанию True)
|
rated_only: Только рейтинговые игры (по умолчанию True)
|
||||||
|
priority: "high" для интерактивных команд (/today и т.п.), чтобы не
|
||||||
|
ждать позади бэклога фоновых периодических проверок в общей
|
||||||
|
очереди shared_token_gate; "background" (по умолчанию) для них.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Список игр в формате JSON или None при ошибке
|
Список игр в формате JSON или None при ошибке
|
||||||
|
|
||||||
|
|
@ -157,15 +165,20 @@ class LichessClient:
|
||||||
if self.app_token:
|
if self.app_token:
|
||||||
headers['Authorization'] = f'Bearer {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 на весь запрос-ответ и с паузой после —
|
# Держим shared_token_gate на весь запрос-ответ и с паузой после —
|
||||||
# LICHESS_APP_TOKEN общий с get_user_activity, и Lichess не пускает
|
# LICHESS_APP_TOKEN общий с get_user_activity, и Lichess не пускает
|
||||||
# 2+ запроса подряд без ощутимой паузы между ними (даже строго
|
# 2+ запроса подряд без ощутимой паузы между ними (даже строго
|
||||||
# последовательные с интервалом ~100-300мс ловили 429 "Please only
|
# последовательные с интервалом ~100-300мс ловили 429 "Please only
|
||||||
# run 1 request(s) at a time" под реальной нагрузкой на проде).
|
# 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)
|
response = await self.client.get(url, params=params, headers=headers)
|
||||||
|
finally:
|
||||||
|
self.shared_token_gate.release()
|
||||||
response.raise_for_status() # Проверяем статус ответа
|
response.raise_for_status() # Проверяем статус ответа
|
||||||
|
|
||||||
# Парсим NDJSON (Newline Delimited JSON)
|
# Парсим NDJSON (Newline Delimited JSON)
|
||||||
|
|
|
||||||
|
|
@ -503,7 +503,11 @@ async def get_games_of_period(
|
||||||
example=True),
|
example=True),
|
||||||
include_games: bool = Query(False,
|
include_games: bool = Query(False,
|
||||||
description="Включить построчный список отдельных партий (только blitz/rapid/classical)",
|
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
|
since_seconds = since // 1000
|
||||||
until_seconds = until // 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:
|
if not result.success:
|
||||||
# Реальная ошибка при обращении к Lichess — не маскируем её под "0 игр"
|
# Реальная ошибка при обращении к Lichess — не маскируем её под "0 игр"
|
||||||
raise HTTPException(status_code=502, detail=result.message)
|
raise HTTPException(status_code=502, detail=result.message)
|
||||||
|
|
|
||||||
|
|
@ -5,6 +5,7 @@ Ensures minimum delay between requests
|
||||||
import asyncio
|
import asyncio
|
||||||
import time
|
import time
|
||||||
import logging
|
import logging
|
||||||
|
from collections import deque
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
@ -65,26 +66,49 @@ class SharedTokenGate:
|
||||||
Lichess is enforcing here needs a real cooldown between calls, not just
|
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
|
non-overlap, so this gate tracks the last release time and forces a
|
||||||
minimum gap before the next request is allowed to start.
|
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):
|
def __init__(self, min_gap: float = 6.0):
|
||||||
self._lock = asyncio.Lock()
|
|
||||||
self._min_gap = min_gap
|
self._min_gap = min_gap
|
||||||
self._last_release_time: Optional[float] = None
|
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:
|
if self._last_release_time is not None:
|
||||||
elapsed = time.time() - self._last_release_time
|
elapsed = time.time() - self._last_release_time
|
||||||
if elapsed < self._min_gap:
|
if elapsed < self._min_gap:
|
||||||
wait_time = self._min_gap - elapsed
|
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")
|
logger.debug(f"Shared token gate: waiting {wait_time:.3f}s before next games/user or user/activity call")
|
||||||
await asyncio.sleep(wait_time)
|
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._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
|
# 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
|
# 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)
|
_shared_token_gate = SharedTokenGate(min_gap=6.0)
|
||||||
|
|
||||||
def get_shared_token_gate() -> SharedTokenGate:
|
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
|
return _shared_token_gate
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -698,7 +698,7 @@ class StatsService:
|
||||||
total=create_game_stats(stats['total'])
|
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 в секундах)
|
until_timestamp: Конец периода (Unix timestamp в секундах)
|
||||||
rated_only: Только рейтинговые игры (по умолчанию True)
|
rated_only: Только рейтинговые игры (по умолчанию True)
|
||||||
include_games: Включить построчный список отдельных партий (только blitz/rapid/classical)
|
include_games: Включить построчный список отдельных партий (только blitz/rapid/classical)
|
||||||
|
priority: "interactive" для запросов из /today и т.п. (обгоняют фоновые
|
||||||
|
периодические проверки в общей очереди shared_token_gate), иначе "background"
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
GamesOfPeriodResponse с статистикой игр
|
GamesOfPeriodResponse с статистикой игр
|
||||||
|
|
@ -719,9 +721,9 @@ class StatsService:
|
||||||
# Конвертируем timestamp в миллисекунды для API Lichess
|
# Конвертируем timestamp в миллисекунды для API Lichess
|
||||||
since_ms = since_timestamp * 1000
|
since_ms = since_timestamp * 1000
|
||||||
until_ms = until_timestamp * 1000
|
until_ms = until_timestamp * 1000
|
||||||
|
|
||||||
# Получаем игры (без фильтра rated в запросе к Lichess — см. lichess_client)
|
# Получаем игры (без фильтра 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:
|
if games is None:
|
||||||
return GamesOfPeriodResponse(
|
return GamesOfPeriodResponse(
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue