Upload folder using huggingface_hub

app/tasks/celery_tasks.py

@@ -31,9 +31,68 @@ celery_app.conf.update(
     task_acks_late=True,
     worker_prefetch_multiplier=1,
     task_reject_on_worker_lost=True,
+    # Drain the greylist retry queue periodically. Without a beat schedule the
+    # queue would only ever fill — a slow leak of unretried addresses.
+    beat_schedule={
+        "drain-greylist-retries": {
+            "task": "app.tasks.celery_tasks.drain_retries_task",
+            "schedule": 300.0,  # every 5 minutes
+        }
+    },
 )
 
 
+@celery_app.task(bind=True, max_retries=1, rate_limit="12/m")
+def drain_retries_task(self, limit: int = 50):
+    """Re-probe addresses that were greylisted earlier.
+
+    Scheduled by beat every 5 minutes. Without this the retry queue would only
+    fill, leaving valid mailboxes on greylisting domains stranded as "risky".
+    """
+    from ..validator.engine import EmailValidationEngine
+    from ..validator.retry_queue import drain_retry_queue, retry_queue
+    import sys
+
+    async def run():
+        eng = EmailValidationEngine()
+
+        async def prober(email: str, mx_host: str):
+            return await eng.mailbox.verify(email, mx_host)
+
+        stats = await drain_retry_queue(prober, queue=retry_queue, limit=limit)
+
+        # Cache the resolved verdicts so History/Results and future runs see the
+        # retry's outcome (a valid mailbox stops being reported as risky).
+        cached = 0
+        for email, result in stats.get("results", []):
+            if result.verified is None:
+                continue
+            try:
+                await eng.cache.put_verdict(email, True, result, None)
+                cached += 1
+            except Exception as exc:
+                print(f"[RETRY] could not cache {email}: {exc}", file=sys.stderr)
+
+        return {
+            "checked": stats["checked"],
+            "resolved": stats["resolved"],
+            "requeued": stats["requeued"],
+            "errors": stats["errors"],
+            "cached": cached,
+        }
+
+    try:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            return loop.run_until_complete(run())
+        finally:
+            loop.close()
+    except Exception as exc:
+        print(f"[RETRY] drain failed: {exc}", file=sys.stderr)
+        return {"error": str(exc)}
+
+
 @celery_app.task(bind=True, max_retries=3, rate_limit="100/m")
 def bulk_validate_task(self, emails: list, webhook_url: str = None, deep: bool = True):
     """

app/validator/engine.py

@@ -20,6 +20,7 @@ from .syntax import validate_syntax
 from .dns_check import check_dns
 from .disposable import check_disposable
 from .mailbox_provider import MailboxProvider, resolve_provider
+from .retry_queue import is_greylist_result
 from .scoring import calculate_score
 
 logger = structlog.get_logger()
@@ -123,6 +124,7 @@ class EmailValidationEngine:
         self,
         mailbox: MailboxProvider | None = None,
         cache=None,
+        retry_queue=None,
     ):
         self.use_proxies = settings.use_proxies
         # Layers 5 (catch-all) and 6 (SMTP) run through a mailbox provider. On
@@ -136,6 +138,12 @@ class EmailValidationEngine:
             from .validation_cache import validation_cache as cache
 
         self.cache = cache
+        # Greylisted (4xx) addresses get a bounded second chance. Without this a
+        # valid mailbox on a greylisting domain stays permanently "risky".
+        if retry_queue is None:
+            from .retry_queue import retry_queue as retry_queue
+
+        self.retry_queue = retry_queue
 
     async def validate(self, email: str, deep: bool = True) -> ValidationResult:
         start_time = time.monotonic()
@@ -524,6 +532,22 @@ class EmailValidationEngine:
                     confidence=0.9 if ca_value is not None else 0.0,
                 )
                 disposable = check_disposable(normalized, domain)
+
+                # A 4xx means "come back later", not a verdict. Schedule one
+                # bounded retry so a real mailbox on a greylisting domain is not
+                # stranded as permanently risky. The queue never raises.
+                if cached is None:
+                    try:
+                        if is_greylist_result(smtp):
+                            await self.retry_queue.enqueue(
+                                normalized, dns.mx_records[0]["host"], smtp.smtp_code
+                            )
+                        elif smtp.verified is not None:
+                            # Definitive answer — cancel any pending retry.
+                            await self.retry_queue.resolve_email(normalized)
+                    except Exception:
+                        pass
+
                 scoring = calculate_score(
                     syntax, dns, disposable, catch_all, smtp
                 )

app/validator/retry_queue.py

@@ -0,0 +1,312 @@
+"""Greylist retry queue.
+
+Real mail servers often answer the first RCPT for an unknown sender with a
+**4xx** (greylisting / "try again later"). That is not a verdict — it is an
+instruction to come back. Without a retry, a perfectly valid mailbox on such a
+domain stays permanently "risky", which is exactly the false-negative that makes
+a validator feel unreliable.
+
+Two rules keep this safe:
+
+  * **Only 4xx is retried.** A 5xx is final, a 250 is done. A *transport*
+    failure (connect timeout / unreachable) is systemic to the MX — re-queuing
+    every address behind a dead host would flood the queue for no gain, so the
+    MX-level circuit breaker handles that instead.
+  * **Bounded attempts with exponential backoff.** A queue that retries forever
+    is a queue that never drains; entries are dropped after `max_attempts`.
+
+The store is pluggable (Redis in production, memory in tests) and every store
+error degrades to "no queue" rather than breaking a validation run.
+"""
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import asdict, dataclass
+from typing import Awaitable, Callable, Optional, Protocol
+
+from ..models import SMTPResult
+
+
+def is_greylist_result(smtp: Optional[SMTPResult]) -> bool:
+    """True only for a genuine SMTP 4xx — the retryable case."""
+    if smtp is None:
+        return False
+    code = smtp.smtp_code
+    return code is not None and 400 <= code < 500
+
+
+@dataclass
+class RetryEntry:
+    email: str
+    mx_host: str
+    attempts: int = 0
+    due_at: float = 0.0
+    last_code: Optional[int] = None
+
+    def as_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "RetryEntry":
+        return cls(
+            email=data["email"],
+            mx_host=data["mx_host"],
+            attempts=int(data.get("attempts", 0)),
+            due_at=float(data.get("due_at", 0.0)),
+            last_code=data.get("last_code"),
+        )
+
+
+class RetryStore(Protocol):
+    async def add(self, entry: RetryEntry) -> None: ...
+    async def due(self, now: float, limit: int) -> list[RetryEntry]: ...
+    async def remove(self, email: str) -> None: ...
+
+
+class MemoryRetryStore:
+    """In-process store. Adequate for a single worker or tests."""
+
+    def __init__(self) -> None:
+        self._items: dict[str, RetryEntry] = {}
+
+    @staticmethod
+    def _key(email: str) -> str:
+        return email.strip().lower()
+
+    async def add(self, entry: RetryEntry) -> None:
+        self._items[self._key(entry.email)] = entry
+
+    async def due(self, now: float, limit: int) -> list[RetryEntry]:
+        ready = [e for e in self._items.values() if e.due_at <= now]
+        ready.sort(key=lambda e: e.due_at)
+        return ready[:limit]
+
+    async def remove(self, email: str) -> None:
+        self._items.pop(self._key(email), None)
+
+    async def get(self, email: str) -> Optional[RetryEntry]:
+        return self._items.get(self._key(email))
+
+    async def count(self) -> int:
+        return len(self._items)
+
+
+class RedisRetryStore:
+    """Durable, worker-shared store: a hash of entries plus a sorted-set index.
+
+    The sorted set is scored by `due_at`, so "what is due now" is one
+    ZRANGEBYSCORE rather than a full scan of every pending address.
+    """
+
+    HASH_KEY = "vx:retry:entries"
+    DUE_KEY = "vx:retry:due"
+
+    def __init__(self, client=None):
+        self._client = client
+
+    async def _redis(self):
+        if self._client is not None:
+            return self._client
+        from ..utils.cache import get_redis
+
+        self._client = await get_redis()
+        return self._client
+
+    @staticmethod
+    def _key(email: str) -> str:
+        return email.strip().lower()
+
+    async def add(self, entry: RetryEntry) -> None:
+        client = await self._redis()
+        if client is None:
+            raise RuntimeError("redis unavailable")
+        key = self._key(entry.email)
+        await client.hset(self.HASH_KEY, key, json.dumps(entry.as_dict()))
+        await client.zadd(self.DUE_KEY, {key: entry.due_at})
+
+    async def due(self, now: float, limit: int) -> list[RetryEntry]:
+        client = await self._redis()
+        if client is None:
+            raise RuntimeError("redis unavailable")
+        members = await client.zrangebyscore(self.DUE_KEY, float("-inf"), now, start=0, num=limit)
+        entries: list[RetryEntry] = []
+        for member in members:
+            raw = await client.hget(self.HASH_KEY, member)
+            if not raw:
+                # Index without a payload (partial write): drop the orphan.
+                await client.zrem(self.DUE_KEY, member)
+                continue
+            try:
+                entries.append(RetryEntry.from_dict(json.loads(raw)))
+            except Exception:
+                # Corrupt payload must be skipped, never crash the drain.
+                await client.zrem(self.DUE_KEY, member)
+                await client.hdel(self.HASH_KEY, member)
+        return entries
+
+    async def remove(self, email: str) -> None:
+        client = await self._redis()
+        if client is None:
+            raise RuntimeError("redis unavailable")
+        key = self._key(email)
+        await client.hdel(self.HASH_KEY, key)
+        await client.zrem(self.DUE_KEY, key)
+
+    async def get(self, email: str) -> Optional[RetryEntry]:
+        client = await self._redis()
+        if client is None:
+            raise RuntimeError("redis unavailable")
+        raw = await client.hget(self.HASH_KEY, self._key(email))
+        if not raw:
+            return None
+        try:
+            return RetryEntry.from_dict(json.loads(raw))
+        except Exception:
+            return None
+
+    async def count(self) -> int:
+        client = await self._redis()
+        if client is None:
+            raise RuntimeError("redis unavailable")
+        return int(await client.hlen(self.HASH_KEY) or 0)
+
+
+def build_retry_store():
+    """Redis when configured/reachable, else the in-process store."""
+    try:
+        from ..utils.cache import _HAS_REDIS  # type: ignore[attr-defined]
+
+        if _HAS_REDIS:
+            return RedisRetryStore()
+    except Exception:
+        pass
+    return MemoryRetryStore()
+
+
+class RetryQueue:
+    """Schedules greylisted addresses for a later attempt."""
+
+    def __init__(
+        self,
+        store: Optional[RetryStore] = None,
+        backoff_seconds: int = 900,      # 15 min — inside the usual greylist window
+        max_attempts: int = 3,
+        now: Optional[Callable[[], float]] = None,
+    ):
+        self.store = store or MemoryRetryStore()
+        self.backoff_seconds = max(1, backoff_seconds)
+        self.max_attempts = max(1, max_attempts)
+        self._now = now or time.time
+
+    def _delay_for(self, attempts: int) -> float:
+        # attempts=0 -> base, 1 -> 2x base, 2 -> 4x base (capped growth).
+        return self.backoff_seconds * (2 ** min(attempts, 4))
+
+    async def enqueue(self, email: str, mx_host: str, code: Optional[int] = None) -> bool:
+        """Schedule `email` for a later attempt. Returns False if not queued."""
+        try:
+            existing = None
+            if hasattr(self.store, "get"):
+                existing = await self.store.get(email)  # type: ignore[attr-defined]
+            if existing is not None:
+                return False  # already queued; never create duplicates
+            entry = RetryEntry(
+                email=email,
+                mx_host=mx_host,
+                attempts=0,
+                due_at=self._now() + self._delay_for(0),
+                last_code=code,
+            )
+            await self.store.add(entry)
+            return True
+        except Exception:
+            return False  # a queue outage must not break validation
+
+    async def due(self, limit: int = 50) -> list[RetryEntry]:
+        try:
+            return await self.store.due(self._now(), limit)
+        except Exception:
+            return []
+
+    async def mark_attempted(self, entry: RetryEntry) -> None:
+        """Record an attempt; reschedule with backoff or drop when exhausted."""
+        try:
+            attempts = entry.attempts + 1
+            if attempts >= self.max_attempts:
+                await self.store.remove(entry.email)
+                return
+            entry.attempts = attempts
+            entry.due_at = self._now() + self._delay_for(attempts)
+            await self.store.add(entry)
+        except Exception:
+            return
+
+    async def resolve(self, entry: RetryEntry) -> None:
+        """The address got a definitive answer — stop retrying it."""
+        try:
+            await self.store.remove(entry.email)
+        except Exception:
+            return
+
+    async def resolve_email(self, email: str) -> None:
+        """Convenience: stop retrying by address (definitive verdict arrived)."""
+        try:
+            await self.store.remove(email)
+        except Exception:
+            return
+
+    async def size(self) -> int:
+        try:
+            if hasattr(self.store, "count"):
+                return await self.store.count()  # type: ignore[attr-defined]
+            return len(await self.store.due(float("inf"), 10_000))
+        except Exception:
+            return 0
+
+
+async def drain_retry_queue(
+    prober: Callable[[str, str], Awaitable[SMTPResult]],
+    queue: Optional[RetryQueue] = None,
+    limit: int = 50,
+) -> dict:
+    """Re-probe addresses that were greylisted earlier.
+
+    A queue that only fills is a leak, so this is the other half of the feature.
+    For each due entry:
+
+      * definitive answer (250 or 5xx) -> drop it from the queue
+      * still 4xx                       -> re-queue with exponential backoff
+      * probe raised                    -> leave queued for a later attempt
+
+    Returns a stats dict including the raw results so the caller can persist them.
+    """
+    q = queue or retry_queue
+    stats: dict = {"checked": 0, "resolved": 0, "requeued": 0, "errors": 0, "results": []}
+
+    for entry in await q.due(limit=limit):
+        stats["checked"] += 1
+        try:
+            result = await prober(entry.email, entry.mx_host)
+        except Exception:
+            # Transport-level blow-up: keep it queued, try again next drain.
+            stats["errors"] += 1
+            await q.mark_attempted(entry)
+            continue
+
+        stats["results"].append((entry.email, result))
+
+        if is_greylist_result(result):
+            stats["requeued"] += 1
+            await q.mark_attempted(entry)
+        else:
+            # 250 or 5xx — a real answer, so stop retrying this address.
+            stats["resolved"] += 1
+            await q.resolve(entry)
+
+    return stats
+
+
+# Shared queue used by the bulk pipeline. Prefers Redis so the queue survives a
+# restart and is shared across workers; falls back to memory otherwise.
+retry_queue = RetryQueue(store=build_retry_store())

tests/test_engine_retry.py

@@ -0,0 +1,168 @@
+import unittest
+from unittest.mock import AsyncMock, patch
+
+from app.models import DNSResult, DisposableResult, SMTPResult, ValidationStatus
+from app.validator.engine import EmailValidationEngine
+from app.validator.retry_queue import MemoryRetryStore, RetryQueue
+
+
+def _dns(host="mx.corp.test"):
+    return DNSResult(
+        domain_exists=True,
+        has_mx_records=True,
+        mx_records=[{"host": host, "priority": 10}],
+    )
+
+
+def _clean():
+    return DisposableResult(
+        is_disposable=False, is_free_provider=False, is_role_based=False
+    )
+
+
+class NullCache:
+    async def get_dns(self, domain):
+        return None
+
+    async def put_dns(self, domain, result):
+        return None
+
+    async def get_verdict(self, email, deep):
+        return None
+
+    async def put_verdict(self, email, deep, smtp, catch_all):
+        return None
+
+
+class ScriptedProvider:
+    def __init__(self, mapping):
+        self.mapping = mapping
+
+    async def verify_many(self, pairs):
+        return (
+            {e: self.mapping.get(e, SMTPResult(verified=True, smtp_code=250)) for e, _ in pairs},
+            {},
+        )
+
+    async def verify(self, email, mx_host):
+        return self.mapping.get(email, SMTPResult(verified=True, smtp_code=250))
+
+    async def catch_all(self, domain, mx_host):
+        return (False, 0.9)
+
+
+class EngineRetryIntegrationTests(unittest.IsolatedAsyncioTestCase):
+    """4xx rows must be re-queued (bounded), definitive rows must not."""
+
+    async def test_greylisted_row_is_enqueued(self):
+        clock = {"t": 0.0}
+        queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=600, now=lambda: clock["t"])
+        engine = EmailValidationEngine(
+            mailbox=ScriptedProvider(
+                {"a@corp.test": SMTPResult(verified=None, smtp_code=451,
+                                           smtp_message="451 greylisted, try later")}
+            ),
+            cache=NullCache(),
+            retry_queue=queue,
+        )
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())),
+            patch("app.validator.engine.check_disposable", return_value=_clean()),
+        ):
+            results = await engine.validate_batch(["a@corp.test"], deep=True)
+
+        self.assertEqual(await queue.size(), 1)
+        entry = (await queue.due(limit=1)) if clock["t"] > 600 else None
+        # Still pending (not yet due), but queued.
+        self.assertEqual(results[0].status, ValidationStatus.RISKY)
+
+    async def test_verified_row_is_not_enqueued(self):
+        queue = RetryQueue(store=MemoryRetryStore(), now=lambda: 0.0)
+        engine = EmailValidationEngine(
+            mailbox=ScriptedProvider(
+                {"a@corp.test": SMTPResult(verified=True, smtp_code=250)}
+            ),
+            cache=NullCache(),
+            retry_queue=queue,
+        )
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())),
+            patch("app.validator.engine.check_disposable", return_value=_clean()),
+        ):
+            await engine.validate_batch(["a@corp.test"], deep=True)
+
+        self.assertEqual(await queue.size(), 0)
+
+    async def test_rejected_row_is_not_enqueued(self):
+        queue = RetryQueue(store=MemoryRetryStore(), now=lambda: 0.0)
+        engine = EmailValidationEngine(
+            mailbox=ScriptedProvider(
+                {"a@corp.test": SMTPResult(verified=False, smtp_code=550,
+                                           smtp_message="550 no such user")}
+            ),
+            cache=NullCache(),
+            retry_queue=queue,
+        )
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())),
+            patch("app.validator.engine.check_disposable", return_value=_clean()),
+        ):
+            await engine.validate_batch(["a@corp.test"], deep=True)
+
+        self.assertEqual(await queue.size(), 0)
+
+    async def test_transport_failure_is_not_enqueued(self):
+        # Systemic to the MX — the circuit breaker owns this, not the queue.
+        queue = RetryQueue(store=MemoryRetryStore(), now=lambda: 0.0)
+        engine = EmailValidationEngine(
+            mailbox=ScriptedProvider(
+                {"a@corp.test": SMTPResult(verified=None, error="transport_error: timed out")}
+            ),
+            cache=NullCache(),
+            retry_queue=queue,
+        )
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())),
+            patch("app.validator.engine.check_disposable", return_value=_clean()),
+        ):
+            await engine.validate_batch(["a@corp.test"], deep=True)
+
+        self.assertEqual(await queue.size(), 0)
+
+    async def test_queue_failure_does_not_break_validation(self):
+        class BrokenQueue:
+            async def enqueue(self, *a, **k):
+                raise RuntimeError("redis down")
+
+        engine = EmailValidationEngine(
+            mailbox=ScriptedProvider(
+                {"a@corp.test": SMTPResult(verified=None, smtp_code=451)}
+            ),
+            cache=NullCache(),
+            retry_queue=BrokenQueue(),
+        )
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())),
+            patch("app.validator.engine.check_disposable", return_value=_clean()),
+        ):
+            results = await engine.validate_batch(["a@corp.test"], deep=True)
+
+        # The row still has a verdict; a broken queue must never lose it.
+        self.assertEqual(len(results), 1)
+
+    async def test_engine_without_queue_still_works(self):
+        engine = EmailValidationEngine(
+            mailbox=ScriptedProvider({"a@corp.test": SMTPResult(verified=True, smtp_code=250)}),
+            cache=NullCache(),
+        )
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns())),
+            patch("app.validator.engine.check_disposable", return_value=_clean()),
+        ):
+            results = await engine.validate_batch(["a@corp.test"], deep=True)
+
+        self.assertEqual(results[0].status, ValidationStatus.VALID)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_retry_queue.py

@@ -0,0 +1,153 @@
+import unittest
+
+from app.validator.retry_queue import (
+    MemoryRetryStore,
+    RetryEntry,
+    RetryQueue,
+    is_greylist_result,
+)
+
+
+class GreylistDetectionTests(unittest.TestCase):
+    """Only a genuine SMTP 4xx (greylisting / temporary mailbox error) is worth
+    retrying. A transport failure is systemic to the MX — re-queuing every
+    address behind it would flood the queue for no gain."""
+
+    def _result(self, **kw):
+        from app.models import SMTPResult
+
+        base = dict(verified=None)
+        base.update(kw)
+        return SMTPResult(**base)
+
+    def test_4xx_is_retryable(self):
+        self.assertTrue(is_greylist_result(self._result(smtp_code=451)))
+        self.assertTrue(is_greylist_result(self._result(smtp_code=421)))
+
+    def test_5xx_is_not_retryable(self):
+        self.assertFalse(is_greylist_result(self._result(verified=False, smtp_code=550)))
+
+    def test_250_is_not_retryable(self):
+        self.assertFalse(is_greylist_result(self._result(verified=True, smtp_code=250)))
+
+    def test_transport_failure_is_not_retryable(self):
+        # Systemic: affects every address at that MX. Retrying per-address would
+        # flood the queue; the MX-level circuit breaker handles this instead.
+        self.assertFalse(
+            is_greylist_result(self._result(error="transport_error: connect timed out"))
+        )
+
+    def test_none_result_is_not_retryable(self):
+        self.assertFalse(is_greylist_result(None))
+
+
+class RetryQueueTests(unittest.IsolatedAsyncioTestCase):
+    async def test_enqueue_schedules_a_future_attempt(self):
+        q = RetryQueue(store=MemoryRetryStore(), backoff_seconds=900, now=lambda: 1000.0)
+        ok = await q.enqueue("a@corp.test", "mx.corp.test", code=451)
+        self.assertTrue(ok)
+        self.assertEqual(await q.size(), 1)
+        # Not due yet.
+        self.assertEqual(await q.due(), [])
+
+    async def test_entry_becomes_due_after_backoff(self):
+        clock = {"t": 1000.0}
+        q = RetryQueue(store=MemoryRetryStore(), backoff_seconds=900, now=lambda: clock["t"])
+        await q.enqueue("a@corp.test", "mx.corp.test", code=451)
+        clock["t"] = 1901.0
+        due = await q.due()
+        self.assertEqual(len(due), 1)
+        self.assertEqual(due[0].email, "a@corp.test")
+        self.assertEqual(due[0].mx_host, "mx.corp.test")
+
+    async def test_duplicate_enqueue_does_not_create_twice(self):
+        q = RetryQueue(store=MemoryRetryStore(), now=lambda: 1.0)
+        await q.enqueue("a@corp.test", "mx.corp.test", code=451)
+        await q.enqueue("a@corp.test", "mx.corp.test", code=451)
+        self.assertEqual(await q.size(), 1)
+
+    async def test_email_matching_is_case_insensitive(self):
+        q = RetryQueue(store=MemoryRetryStore(), now=lambda: 1.0)
+        await q.enqueue("A@Corp.test", "mx.corp.test", code=451)
+        await q.enqueue("a@corp.test", "mx.corp.test", code=451)
+        self.assertEqual(await q.size(), 1)
+
+    async def test_attempt_increments_and_reschedules_with_backoff(self):
+        clock = {"t": 0.0}
+        q = RetryQueue(
+            store=MemoryRetryStore(), backoff_seconds=600, max_attempts=3, now=lambda: clock["t"]
+        )
+        await q.enqueue("a@corp.test", "mx.corp.test", code=451)
+        clock["t"] = 601.0
+        entry = (await q.due())[0]
+        await q.mark_attempted(entry)
+
+        # Still queued (2nd attempt allowed), and not immediately due.
+        self.assertEqual(await q.size(), 1)
+        self.assertEqual(await q.due(), [])
+        clock["t"] = 601.0 + 1200.0  # exponential: 600 * 2
+        self.assertEqual(len(await q.due()), 1)
+
+    async def test_max_attempts_removes_the_entry(self):
+        clock = {"t": 0.0}
+        q = RetryQueue(
+            store=MemoryRetryStore(), backoff_seconds=10, max_attempts=2, now=lambda: clock["t"]
+        )
+        await q.enqueue("a@corp.test", "mx.corp.test", code=451)
+        clock["t"] = 11.0
+        await q.mark_attempted((await q.due())[0])
+        clock["t"] = 40.0
+        await q.mark_attempted((await q.due())[0])
+        self.assertEqual(await q.size(), 0, "must not retry forever")
+
+    async def test_resolve_removes_the_entry(self):
+        clock = {"t": 0.0}
+        q = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"])
+        await q.enqueue("a@corp.test", "mx.corp.test", code=451)
+        clock["t"] = 11.0
+        await q.resolve((await q.due())[0])
+        self.assertEqual(await q.size(), 0)
+
+    async def test_due_respects_limit(self):
+        clock = {"t": 0.0}
+        q = RetryQueue(store=MemoryRetryStore(), backoff_seconds=1, now=lambda: clock["t"])
+        for i in range(5):
+            await q.enqueue(f"u{i}@corp.test", "mx.corp.test", code=451)
+        clock["t"] = 5.0
+        self.assertEqual(len(await q.due(limit=2)), 2)
+
+    async def test_store_failure_does_not_raise(self):
+        class BrokenStore:
+            async def add(self, *a, **k):
+                raise RuntimeError("redis down")
+
+            async def due(self, *a, **k):
+                raise RuntimeError("redis down")
+
+            async def remove(self, *a, **k):
+                raise RuntimeError("redis down")
+
+            async def get(self, *a, **k):
+                raise RuntimeError("redis down")
+
+            async def count(self, *a, **k):
+                raise RuntimeError("redis down")
+
+        q = RetryQueue(store=BrokenStore())
+        self.assertFalse(await q.enqueue("a@corp.test", "mx.corp.test", code=451))
+        self.assertEqual(await q.due(), [])
+        self.assertEqual(await q.size(), 0)
+
+
+class RetryEntryTests(unittest.TestCase):
+    def test_entry_serialises(self):
+        e = RetryEntry(email="a@corp.test", mx_host="mx.corp.test", attempts=1, due_at=123.0, last_code=451)
+        data = e.as_dict()
+        self.assertEqual(data["email"], "a@corp.test")
+        back = RetryEntry.from_dict(data)
+        self.assertEqual(back.email, e.email)
+        self.assertEqual(back.due_at, e.due_at)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_retry_worker.py

@@ -0,0 +1,202 @@
+import json
+import unittest
+
+from app.models import SMTPResult
+from app.validator.retry_queue import (
+    MemoryRetryStore,
+    RedisRetryStore,
+    RetryEntry,
+    RetryQueue,
+    drain_retry_queue,
+)
+
+
+class FakeRedis:
+    """Minimal async Redis subset (hash + zset) used by RedisRetryStore."""
+
+    def __init__(self):
+        self.hashes: dict[str, dict[str, str]] = {}
+        self.zsets: dict[str, dict[str, float]] = {}
+
+    async def hset(self, key, field, value):
+        self.hashes.setdefault(key, {})[field] = value
+
+    async def hget(self, key, field):
+        return self.hashes.get(key, {}).get(field)
+
+    async def hdel(self, key, field):
+        self.hashes.get(key, {}).pop(field, None)
+
+    async def zadd(self, key, mapping):
+        self.zsets.setdefault(key, {}).update(mapping)
+
+    async def zrem(self, key, member):
+        self.zsets.get(key, {}).pop(member, None)
+
+    async def zrangebyscore(self, key, min_, max_, start=0, num=None):
+        items = [
+            (m, s) for m, s in self.zsets.get(key, {}).items() if min_ <= s <= max_
+        ]
+        items.sort(key=lambda kv: kv[1])
+        members = [m for m, _ in items]
+        return members[start : (start + num) if num else None]
+
+    async def zcard(self, key):
+        return len(self.zsets.get(key, {}))
+
+    async def hlen(self, key):
+        return len(self.hashes.get(key, {}))
+
+
+class RedisRetryStoreTests(unittest.IsolatedAsyncioTestCase):
+    """Production needs the queue to survive a restart and be shared across
+    workers — an in-process dict does neither."""
+
+    async def test_add_persists_entry(self):
+        store = RedisRetryStore(client=FakeRedis())
+        entry = RetryEntry(email="a@corp.test", mx_host="mx.corp.test", due_at=500.0)
+        await store.add(entry)
+        got = await store.get("a@corp.test")
+        self.assertIsNotNone(got)
+        self.assertEqual(got.mx_host, "mx.corp.test")
+        self.assertEqual(got.due_at, 500.0)
+
+    async def test_due_returns_only_matured_entries(self):
+        store = RedisRetryStore(client=FakeRedis())
+        await store.add(RetryEntry(email="early@corp.test", mx_host="mx", due_at=100.0))
+        await store.add(RetryEntry(email="late@corp.test", mx_host="mx", due_at=9999.0))
+        due = await store.due(now=200.0, limit=10)
+        self.assertEqual([e.email for e in due], ["early@corp.test"])
+
+    async def test_due_is_ordered_by_time(self):
+        store = RedisRetryStore(client=FakeRedis())
+        await store.add(RetryEntry(email="b@corp.test", mx_host="mx", due_at=300.0))
+        await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=200.0))
+        due = await store.due(now=400.0, limit=10)
+        self.assertEqual([e.email for e in due], ["a@corp.test", "b@corp.test"])
+
+    async def test_remove_clears_both_indexes(self):
+        redis = FakeRedis()
+        store = RedisRetryStore(client=redis)
+        await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=1.0))
+        await store.remove("a@corp.test")
+        self.assertIsNone(await store.get("a@corp.test"))
+        self.assertEqual(await store.count(), 0)
+
+    async def test_readd_updates_rather_than_duplicates(self):
+        store = RedisRetryStore(client=FakeRedis())
+        await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=1.0))
+        await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=99.0, attempts=1))
+        self.assertEqual(await store.count(), 1)
+        self.assertEqual((await store.get("a@corp.test")).attempts, 1)
+
+    async def test_corrupt_entry_is_skipped_not_raised(self):
+        redis = FakeRedis()
+        store = RedisRetryStore(client=redis)
+        await store.add(RetryEntry(email="ok@corp.test", mx_host="mx", due_at=1.0))
+        # Simulate a half-written/garbage payload for another member.
+        redis.hashes.setdefault("vx:retry:entries", {})["bad@corp.test"] = "{not json"
+        redis.zsets.setdefault("vx:retry:due", {})["bad@corp.test"] = 1.0
+        due = await store.due(now=5.0, limit=10)
+        self.assertEqual([e.email for e in due], ["ok@corp.test"])
+
+
+class DrainWorkerTests(unittest.IsolatedAsyncioTestCase):
+    """A queue that only fills is a leak. Due entries must actually be retried,
+    and a definitive answer must clear the entry."""
+
+    async def test_drain_reprobes_due_entries_and_resolves_them(self):
+        clock = {"t": 0.0}
+        queue = RetryQueue(
+            store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"]
+        )
+        await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
+        clock["t"] = 100.0
+
+        probed: list[tuple[str, str]] = []
+
+        async def prober(email, mx_host):
+            probed.append((email, mx_host))
+            return SMTPResult(verified=True, smtp_code=250)
+
+        stats = await drain_retry_queue(prober, queue=queue, limit=10)
+
+        self.assertEqual(probed, [("a@corp.test", "mx.corp.test")])
+        self.assertEqual(stats["resolved"], 1)
+        self.assertEqual(stats["requeued"], 0)
+        self.assertEqual(await queue.size(), 0)
+
+    async def test_still_greylisted_entry_is_requeued_with_backoff(self):
+        clock = {"t": 0.0}
+        queue = RetryQueue(
+            store=MemoryRetryStore(), backoff_seconds=10, max_attempts=3,
+            now=lambda: clock["t"],
+        )
+        await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
+        clock["t"] = 100.0
+
+        async def prober(email, mx_host):
+            return SMTPResult(verified=None, smtp_code=451, smtp_message="still greylisted")
+
+        stats = await drain_retry_queue(prober, queue=queue, limit=10)
+
+        self.assertEqual(stats["requeued"], 1)
+        self.assertEqual(await queue.size(), 1)
+        # Backed off: not immediately due again.
+        self.assertEqual(await queue.due(), [])
+
+    async def test_rejected_entry_is_resolved_and_removed(self):
+        clock = {"t": 0.0}
+        queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"])
+        await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
+        clock["t"] = 100.0
+
+        async def prober(email, mx_host):
+            return SMTPResult(verified=False, smtp_code=550, smtp_message="550 no such user")
+
+        stats = await drain_retry_queue(prober, queue=queue, limit=10)
+
+        self.assertEqual(stats["resolved"], 1)
+        self.assertEqual(await queue.size(), 0)
+
+    async def test_prober_error_requeues_without_crashing(self):
+        clock = {"t": 0.0}
+        queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"])
+        await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
+        clock["t"] = 100.0
+
+        async def prober(email, mx_host):
+            raise RuntimeError("mx exploded")
+
+        stats = await drain_retry_queue(prober, queue=queue, limit=10)
+
+        self.assertEqual(stats["errors"], 1)
+        self.assertEqual(await queue.size(), 1, "must stay queued for another try")
+
+    async def test_empty_queue_is_a_noop(self):
+        queue = RetryQueue(store=MemoryRetryStore(), now=lambda: 0.0)
+
+        async def prober(email, mx_host):
+            raise AssertionError("should not be called")
+
+        stats = await drain_retry_queue(prober, queue=queue, limit=10)
+        self.assertEqual(
+            stats, {"checked": 0, "resolved": 0, "requeued": 0, "errors": 0, "results": []}
+        )
+
+    async def test_results_are_returned_for_persistence(self):
+        clock = {"t": 0.0}
+        queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"])
+        await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
+        clock["t"] = 100.0
+
+        async def prober(email, mx_host):
+            return SMTPResult(verified=True, smtp_code=250)
+
+        stats = await drain_retry_queue(prober, queue=queue, limit=10)
+        self.assertEqual(stats["results"][0][0], "a@corp.test")
+        self.assertTrue(stats["results"][0][1].verified)
+
+
+if __name__ == "__main__":
+    unittest.main()