File size: 1,831 Bytes
0f6b827 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | """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 # stop event set
except asyncio.TimeoutError:
pass # interval elapsed — drain
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() |