Upload folder using huggingface_hub

app/validator/mx_batch.py

@@ -79,6 +79,8 @@ class MxBatchVerifier:
         mail_from: str = "probe@validator.local",
         catch_all_cache: Optional[CatchAllCache] = None,
         max_connections_per_host: int = 2,
+        fail_threshold: int = 3,
+        cooldown_seconds: int = 300,
     ):
         self.timeout = timeout
         self.helo_domain = helo_domain
@@ -88,6 +90,27 @@ class MxBatchVerifier:
         self._factory = connection_factory or (
             lambda host, **kw: SmtpConnection(host, **kw)
         )
+        # Egress circuit breaker. On a host that blocks outbound port 25 every
+        # connect burns the full timeout, so a 10k job would take ~27h. After
+        # `fail_threshold` consecutive connect failures we stop dialling for
+        # `cooldown_seconds` and report rows unavailable immediately instead.
+        self.fail_threshold = max(1, fail_threshold)
+        self.cooldown_seconds = cooldown_seconds
+        self._consecutive_failures = 0
+        self._open_until = 0.0
+
+    @property
+    def circuit_open(self) -> bool:
+        return time.monotonic() < self._open_until
+
+    def _record_connect_failure(self) -> None:
+        self._consecutive_failures += 1
+        if self._consecutive_failures >= self.fail_threshold:
+            self._open_until = time.monotonic() + self.cooldown_seconds
+
+    def _record_connect_success(self) -> None:
+        self._consecutive_failures = 0
+        self._open_until = 0.0
 
     def _new_connection(self, mx_host: str):
         return self._factory(
@@ -130,9 +153,21 @@ class MxBatchVerifier:
             }
 
         out: dict[str, SMTPResult] = {}
+
+        # Fail fast while egress is known-bad: do not dial, do not wait.
+        if self.circuit_open:
+            return {
+                e: SMTPResult(
+                    verified=None,
+                    error=f"smtp_unreachable: {mx_host} (egress circuit open)",
+                )
+                for e in emails
+            }
+
         conn = self._new_connection(mx_host)
         try:
             if not conn.open():
+                self._record_connect_failure()
                 return {
                     e: SMTPResult(
                         verified=None,
@@ -140,6 +175,7 @@ class MxBatchVerifier:
                     )
                     for e in emails
                 }
+            self._record_connect_success()
 
             if cached is None:
                 ca = conn.probe(_gibberish(domain))

tests/test_mx_batch.py

@@ -145,5 +145,57 @@ class CatchAllCacheTests(unittest.TestCase):
         self.assertIsNone(c.get("never-seen.test"))
 
 
+class EgressCircuitBreakerTests(unittest.TestCase):
+    """When outbound port 25 is blocked (HF Spaces, most PaaS) every connect
+    burns the full timeout. A bulk job must detect that once and fail fast
+    instead of spending timeout × N — 10k rows at 10s/row is ~27 hours."""
+
+    def setUp(self):
+        FakeConnection.registry = {}
+        FakeConnection.opened = []
+
+    def test_circuit_opens_after_threshold_and_stops_connecting(self):
+        class DeadConn(FakeConnection):
+            def open(self):
+                self.ready = False
+                return False
+
+        FakeConnection.registry = {"mx.a.test": {"*": 250}}
+        v = MxBatchVerifier(connection_factory=DeadConn, fail_threshold=2, cooldown_seconds=60)
+
+        v.verify_batch([("a@a.test", "mx.a.test")])
+        v.verify_batch([("b@b.test", "mx.b.test")])
+        self.assertTrue(v.circuit_open)
+
+        before = len(FakeConnection.opened)
+        out = v.verify_batch([("c@c.test", "mx.c.test")])
+        # No new connection attempted while the circuit is open.
+        self.assertEqual(len(FakeConnection.opened), before)
+        self.assertIsNone(out["c@c.test"].verified)
+        self.assertIn("unreachable", (out["c@c.test"].error or "").lower())
+
+    def test_successful_connection_resets_failure_count(self):
+        class FlakyConn(FakeConnection):
+            fail_next = True
+
+            def open(self):
+                if FlakyConn.fail_next:
+                    self.ready = False
+                    return False
+                self.ready = True
+                return True
+
+        FakeConnection.registry = {"mx.a.test": {"*": 250}}
+        v = MxBatchVerifier(connection_factory=FlakyConn, fail_threshold=2, cooldown_seconds=60)
+        v.verify_batch([("a@a.test", "mx.a.test")])
+        FlakyConn.fail_next = False
+        v.verify_batch([("b@b.test", "mx.b.test")])
+        self.assertFalse(v.circuit_open)
+
+    def test_default_threshold_is_conservative(self):
+        v = MxBatchVerifier(connection_factory=FakeConnection)
+        self.assertGreaterEqual(v.fail_threshold, 2)
+
+
 if __name__ == "__main__":
     unittest.main()