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)