File size: 7,131 Bytes
5db9737 fd7800b 5db9737 58d0e85 5db9737 58d0e85 5db9737 fd7800b 5db9737 fd7800b 5db9737 45aeb2f 5db9737 45aeb2f 5db9737 45aeb2f 5db9737 fd7800b 45aeb2f fd7800b 45aeb2f 5db9737 45aeb2f 5db9737 978c21c 5db9737 | 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 | from celery import Celery
from ..config import settings
import asyncio
celery_app = Celery(
"emailvalidator",
broker=settings.celery_broker_url,
backend=settings.celery_result_backend,
include=["app.tasks.celery_tasks"]
)
celery_app.conf.update(
task_serializer="json",
accept_content=["json"],
result_serializer="json",
timezone="UTC",
enable_utc=True,
task_track_started=True,
worker_concurrency=10,
task_soft_time_limit=300,
task_time_limit=600,
# Fail fast when Redis/broker is down instead of retrying forever
broker_connection_max_retries=2,
broker_connection_retry_on_startup=False,
task_publish_retry=False,
task_publish_retry_policy={"max_retries": 1},
result_backend_transport_options={
"retry_policy": {"max_retries": 1, "interval_start": 0, "interval_step": 1},
},
# Improve worker acknowledgment and task reliability
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):
"""
Celery task for async bulk email validation.
Runs synchronously inside Celery worker.
`deep` defaults to True: bulk must run the full 7-layer pipeline (catch-all
+ SMTP) to produce a real verdict, matching single verification.
"""
from ..validator.engine import EmailValidationEngine
from ..db import repo
import sys
CHUNK = 200
async def run():
eng = EmailValidationEngine()
results = []
total = len(emails)
# Process in chunks through the engine's batched path: DNS/MX once per
# domain and one mailbox call per chunk, so the SMTP engine reuses a
# session per receiving host instead of a TCP connection per address.
for start in range(0, total, CHUNK):
chunk = emails[start:start + CHUNK]
try:
chunk_results = await eng.validate_batch(chunk, deep=deep)
results.extend(r.model_dump() for r in chunk_results)
except Exception as e:
print(
f"[CELERY TASK] Validation failed for chunk "
f"{start + 1}-{start + len(chunk)}/{total}: {e.__class__.__name__}",
file=sys.stderr,
)
results.extend(
{"email": email, "error": str(e), "status": "error"}
for email in chunk
)
done = min(start + len(chunk), total)
self.update_state(state="PROGRESS", meta={"current": done, "total": total})
# Persist incremental progress to Supabase (best-effort)
try:
repo.update_bulk_job_progress(self.request.id, done, total)
except Exception as e:
print(f"[CELERY TASK] Failed to update progress: {str(e)}", file=sys.stderr)
# Run-quality monitor (report item 10): a run that is mostly
# risky/unknown is a systemic failure — page via the signed webhook.
try:
from ..utils.rate_alerts import maybe_alert_risky_rate
await maybe_alert_risky_rate(
results,
webhook_url=webhook_url,
job_id=self.request.id,
threshold=settings.risky_rate_alert_threshold,
)
except Exception as e:
print(f"[CELERY TASK] Rate alert check failed: {str(e)}", file=sys.stderr)
return results
try:
print(f"[CELERY TASK] Starting bulk validation for {len(emails)} emails, task_id: {self.request.id}")
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
results = loop.run_until_complete(run())
loop.close()
print(f"[CELERY TASK] Completed bulk validation for {len(emails)} emails, task_id: {self.request.id}")
# Persist finished job (best-effort)
try:
repo.finish_bulk_job(self.request.id, results)
except Exception as e:
print(f"[CELERY TASK] Failed to finish job: {str(e)}", file=sys.stderr)
# Webhook notification
if webhook_url:
import httpx
try:
httpx.post(webhook_url, json={"results": results}, timeout=10)
print(f"[CELERY TASK] Webhook notification sent to {webhook_url}")
except Exception as e:
print(f"[CELERY TASK] Webhook notification failed: {str(e)}", file=sys.stderr)
return results
except Exception as exc:
print(f"[CELERY TASK] Fatal error in bulk validation: {str(exc)}", file=sys.stderr)
raise self.retry(exc=exc, countdown=5)
|