LichessStatTgWeb/LichessClientTG_bot/request_queue.py

"""
Request Queue for managing API requests with 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, Set
from datetime import datetime

import config

logger = logging.getLogger(__name__)

class RequestQueue:
    """
    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_dispatch_interval: float, max_concurrent: int):
        """
        Args:
            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_dispatch_interval = min_dispatch_interval
        self.max_concurrent = max_concurrent
        self.queue = asyncio.Queue()
        self.is_processing = False
        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
        """
        future = asyncio.Future()

        await self.queue.put({
            'func': request_func,
            'args': args,
            'kwargs': kwargs,
            'future': future
        })

        if not self.is_processing:
            self._start_processor()

        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 "
                f"(dispatch interval: {self.min_dispatch_interval}s, "
                f"max concurrent: {self.max_concurrent})"
            )

    async def _process_queue(self):
        """Dequeue requests and dispatch them at a paced rate, without waiting for completion"""
        logger.info("📋 Request queue processor started")

        while True:
            try:
                request_item = await self.queue.get()

                # 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
            except Exception as e:
                logger.error(f"❌ Error in request queue processor: {e}")
                import traceback
                logger.error(traceback.format_exc())
                await asyncio.sleep(1)

    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_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_dispatch_time = now

    async def stop(self):
        """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")

# Global request queue instance
_request_queue: Optional[RequestQueue] = None

def get_request_queue() -> RequestQueue:
    """Get the global request queue instance"""
    global _request_queue
    if _request_queue is None:
        _request_queue = RequestQueue(
            min_dispatch_interval=config.LICHESS_REQUEST_QUEUE_MIN_DISPATCH_INTERVAL,
            max_concurrent=config.LICHESS_REQUEST_QUEUE_MAX_CONCURRENT,
        )
    return _request_queue