| """In-process retry-queue drain loop. |
| |
| The Dockerfile CMD runs uvicorn only — no Celery worker, no beat — so the |
| greylist retry queue would fill forever without this loop. It runs as a |
| background asyncio task started in the FastAPI lifespan, draining the shared |
| queue every `interval_seconds`. |
| |
| Every prober error is contained: the loop must survive anything the queue or |
| the network throws at it. |
| """ |
| from __future__ import annotations |
|
|
| import asyncio |
| import logging |
| from typing import Awaitable, Callable |
|
|
| from ..validator.retry_queue import drain_retry_queue, retry_queue |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class DrainLoop: |
| """Periodically drains the greylist retry queue in-process.""" |
|
|
| def __init__( |
| self, |
| prober: Callable[[str, str], Awaitable[object]], |
| interval_seconds: int = 300, |
| limit: int = 50, |
| queue=None, |
| ): |
| self.prober = prober |
| self.interval_seconds = interval_seconds |
| self.limit = limit |
| self.queue = queue if queue is not None else retry_queue |
| self.drain_count = 0 |
| self._stop_event = asyncio.Event() |
|
|
| async def start(self) -> None: |
| """Loop forever until stop() — drain every interval.""" |
| while not self._stop_event.is_set(): |
| try: |
| await asyncio.wait_for(self._stop_event.wait(), timeout=self.interval_seconds) |
| break |
| except asyncio.TimeoutError: |
| pass |
| try: |
| await drain_retry_queue(self.prober, queue=self.queue, limit=self.limit) |
| self.drain_count += 1 |
| except Exception: |
| logger.warning("drain loop iteration failed", exc_info=True) |
|
|
| async def stop(self) -> None: |
| self._stop_event.set() |
