Upload folder using huggingface_hub

app/validator/mailbox_provider.py

@@ -178,7 +178,11 @@ class AutonomousSmtpProvider(MailboxProvider):
         pairs = [(e, mx) for e, mx in pairs if e and mx]
         if not pairs:
             return {}, {}
-        results = self.batch.verify_batch(pairs)
+        # Prefer the concurrent path when the batch engine supports it.
+        if hasattr(self.batch, "verify_batch_async"):
+            results = await self.batch.verify_batch_async(pairs)
+        else:
+            results = self.batch.verify_batch(pairs)
         domains = {e.rsplit("@", 1)[-1].lower(): mx for e, mx in pairs}
         catch_all = {
             domain: self.batch.catch_all_cache.get(domain) for domain in domains

app/validator/mx_batch.py

@@ -18,6 +18,7 @@ bulk job.
 """
 from __future__ import annotations
 
+import asyncio
 import random
 import time
 from dataclasses import dataclass
@@ -81,6 +82,7 @@ class MxBatchVerifier:
         max_connections_per_host: int = 2,
         fail_threshold: int = 3,
         cooldown_seconds: int = 300,
+        bulk_concurrency: int = 20,
     ):
         self.timeout = timeout
         self.helo_domain = helo_domain
@@ -96,6 +98,7 @@ class MxBatchVerifier:
         # `cooldown_seconds` and report rows unavailable immediately instead.
         self.fail_threshold = max(1, fail_threshold)
         self.cooldown_seconds = cooldown_seconds
+        self.bulk_concurrency = max(1, bulk_concurrency)
         self._consecutive_failures = 0
         self._open_until = 0.0
 
@@ -123,22 +126,73 @@ class MxBatchVerifier:
     def verify_batch(
         self, pairs: Iterable[tuple[str, str]]
     ) -> dict[str, SMTPResult]:
-        """Verify pairs of (email, mx_host). Returns email -> SMTPResult."""
-        pairs = list(pairs)
-        if not pairs:
+        """Verify pairs of (email, mx_host), sequentially. Returns email -> result."""
+        groups = self._group(pairs)
+        results: dict[str, SMTPResult] = {}
+        for (domain, mx_host), emails in groups.items():
+            results.update(self._verify_domain(domain, mx_host, emails))
+        return results
+
+    async def verify_batch_async(
+        self, pairs: Iterable[tuple[str, str]]
+    ) -> dict[str, SMTPResult]:
+        """Verify many pairs with DOMAINS PROBED CONCURRENTLY.
+
+        Blocking SMTP work runs in a thread pool bounded by `bulk_concurrency`.
+        Sequential probing makes wall-clock the sum of every domain's latency
+        (timeout x domains on a blocked host); real bulk throughput comes from
+        parallelism across domains, not speed per domain.
+        """
+        groups = self._group(pairs)
+        if not groups:
             return {}
 
-        # Group by (domain, mx) — catch-all is a domain property, and one domain
-        # maps to one MX host, so this is the natural session unit.
+        semaphore = asyncio.Semaphore(self.bulk_concurrency)
+
+        async def run_domain(domain: str, mx_host: str, emails: list[str]):
+            async with semaphore:
+                return await asyncio.to_thread(
+                    self._verify_domain, domain, mx_host, emails
+                )
+
+        chunks = await asyncio.gather(
+            *(run_domain(d, m, e) for (d, m), e in groups.items()),
+            return_exceptions=True,
+        )
+        results: dict[str, SMTPResult] = {}
+        for chunk in chunks:
+            if isinstance(chunk, Exception):
+                continue
+            results.update(chunk)
+        # Any group that raised must still return a row per address.
+        missing = [
+            (d, m, e)
+            for (d, m), e in groups.items()
+            if not any(addr in results for addr in e)
+        ]
+        for domain, mx_host, emails in missing:
+            for email in emails:
+                results.setdefault(
+                    email,
+                    SMTPResult(
+                        verified=None,
+                        error=f"smtp_unreachable: {mx_host} (probe aborted)",
+                    ),
+                )
+        return results
+
+    @staticmethod
+    def _group(
+        pairs: Iterable[tuple[str, str]]
+    ) -> dict[tuple[str, str], list[str]]:
+        """Group (email, mx_host) pairs by (domain, mx) — the session unit."""
         groups: dict[tuple[str, str], list[str]] = {}
         for email, mx_host in pairs:
+            if not email or not mx_host:
+                continue
             domain = email.rsplit("@", 1)[-1].lower()
             groups.setdefault((domain, mx_host), []).append(email)
-
-        results: dict[str, SMTPResult] = {}
-        for (domain, mx_host), emails in groups.items():
-            results.update(self._verify_domain(domain, mx_host, emails))
-        return results
+        return groups
 
     def _verify_domain(
         self, domain: str, mx_host: str, emails: list[str]

tests/test_mx_batch.py

@@ -197,5 +197,74 @@ class EgressCircuitBreakerTests(unittest.TestCase):
         self.assertGreaterEqual(v.fail_threshold, 2)
 
 
+class ParallelDomainTests(unittest.IsolatedAsyncioTestCase):
+    """Domains must be probed CONCURRENTLY (with a cap).
+
+    Sequential probing means wall-clock = sum of every domain's latency; on a
+    blocked host that is timeout x domains. Commercial validators get bulk
+    throughput from parallelism across domains, not speed per domain.
+    """
+
+    async def test_domains_are_probed_in_parallel(self):
+        import asyncio
+        import time
+
+        active = {"now": 0, "peak": 0}
+
+        class SlowConn(FakeConnection):
+            def open(self):
+                active["now"] += 1
+                active["peak"] = max(active["peak"], active["now"])
+                time.sleep(0.25)
+                active["now"] -= 1
+                self.ready = True
+                return True
+
+        FakeConnection.registry = {f"mx{i}.test": {"*": 550} for i in range(8)}
+        v = MxBatchVerifier(
+            connection_factory=SlowConn, bulk_concurrency=8, max_connections_per_host=1
+        )
+        pairs = [(f"u{i}@dom{i}.test", f"mx{i}.test") for i in range(8)]
+
+        t0 = time.time()
+        out = await v.verify_batch_async(pairs)
+        elapsed = time.time() - t0
+
+        self.assertEqual(len(out), 8)
+        # 8 x 0.25s sequential = 2.0s; parallel should be well under 1s.
+        self.assertLess(elapsed, 1.2, f"domains ran sequentially ({elapsed:.2f}s)")
+        self.assertGreater(active["peak"], 2, "no real concurrency observed")
+
+    async def test_concurrency_cap_is_respected(self):
+        import time
+
+        active = {"now": 0, "peak": 0}
+
+        class SlowConn(FakeConnection):
+            def open(self):
+                active["now"] += 1
+                active["peak"] = max(active["peak"], active["now"])
+                time.sleep(0.15)
+                active["now"] -= 1
+                self.ready = True
+                return True
+
+        FakeConnection.registry = {f"mx{i}.test": {"*": 550} for i in range(10)}
+        v = MxBatchVerifier(
+            connection_factory=SlowConn, bulk_concurrency=3, max_connections_per_host=1
+        )
+        pairs = [(f"u{i}@dom{i}.test", f"mx{i}.test") for i in range(10)]
+        out = await v.verify_batch_async(pairs)
+
+        self.assertEqual(len(out), 10)
+        self.assertLessEqual(active["peak"], 3, "exceeded the concurrency cap")
+
+    async def test_sync_verify_batch_still_works(self):
+        FakeConnection.registry = {"mx.corp.test": {"a@corp.test": 250, "*": 550}}
+        v = MxBatchVerifier(connection_factory=FakeConnection)
+        out = v.verify_batch([("a@corp.test", "mx.corp.test")])
+        self.assertTrue(out["a@corp.test"].verified)
+
+
 if __name__ == "__main__":
     unittest.main()