File size: 10,967 Bytes
58d0e85 0f6b827 58d0e85 | 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 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 | """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()
# 0 is legitimate (immediate retry — tests and hot-path recovery);
# the bounded max_attempts prevents any runaway.
self.backoff_seconds = max(0, 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()) |